SpringBoot中消息中间件Kafka实现Websocket的集群

SpringBoot中消息中间件Kafka实现Websocket的集群

1、在实际项目中，由于数据量的增大及并发数的增多，我们不可能只用一台Websocket服务，这个时候就需要用到Webscoket的集群。但是Websocket集群会遇到一些问题。首先我们肯定会想到直接将Websocket的Session放到Redis等缓存服务器中，然后用的时候直接在Redis中获取。但是Webscoket的Session比较特殊，它不能被序列化，因为 WebSocket的session是有状态的，还有就是 WebSocket的session是有时效性的，只要连接一断开，该Session就会失效。

2、解决Websocket集群的三种方法

　　2.1、通过相应的算法，将有关联的用户（即有可能发生聊天的对象）全部指定到一台Webscoket服务。这样就不会存在聊天对象收不到消息的情况。但是这种方法有局限性，就是用户只能和有关联的用户聊天，不能和其他未建立关联的用户聊天。

　　2.2、使用Redis的消息订阅功能来实现WebSocket集群。大致思路如下图。

2.3使用Kafka等消息中间件来实现Webscoket集群。这也是目前我选用的方式。其实该方法和Redis的消息订阅大致思路差不多。但是Redis我们只把他作为缓存使用，不想Redis涉及太多的业务处理，因此就选用了Kafka。

　2.3.3、在项目的pom文件中添加Kafka依赖（注：Kafka依赖的版本必须和服务器上安装的版本一致）

        <!-- kafka依赖 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>1.1.0.RELEASE</version></dependency>

　　　　　2.3.4、建立Kafka的生产者Bean

l.configuration;import t.annotation.Bean;
import t.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.DefaultKafkaProducerFactory;
import org.KafkaTemplate;
import org.ProducerFactory;import java.util.HashMap;
import java.util.Map;/*** @Author: yxl* @Description: Kafka生产者（消息发送者）* @DATE: Created in 2018/11/14*/
@Configuration
@EnableKafka
public class KafkaProducerConfig {public Map<String, Object> producerConfigs() {Map<String, Object> properties = new HashMap<>();properties.put("bootstrap.servers", "kafka集群IP1:9092,kafka集群IP2:9092");properties.put("acks", "all");//ack是判别请求是否为完整的条件（就是是判断是不是成功发送了）。我们指定了“all”将会阻塞消息，这种设置性能最低，但是是最可靠的。properties.put("retries", 0);//如果请求失败，生产者会自动重试，我们指定是0次，如果启用重试，则会有重复消息的可能性。properties.put("batch.size", 16384);//producer(生产者)缓存每个分区未发送消息。缓存的大小是通过 batch.size 配置指定的properties.put("linger.ms", 1);properties.put(&#", 33554432);properties.put("key.serializer", "org.apache.kafkamon.serialization.StringSerializer");properties.put("value.serializer", "org.apache.kafkamon.serialization.StringSerializer");return properties;}public ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<String, String>(producerFactory());}}

　　　　2.3.4、建立Kafka的消费者Bean以及消费者监听

l.configuration;l.myListener.MyKafkaListener;
import org.apachemons.lang3.StringUtils;
import t.annotation.Bean;
import t.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.fig.ConcurrentKafkaListenerContainerFactory;
import org.fig.KafkaListenerContainerFactory;
import org.ConsumerFactory;
import org.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import java.InetAddress;
import java.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;/*** @Author: yxl* @Description: Kafka消费者* @DATE: Created in 2018/11/14*/
@Configuration
@EnableKafka
public class KafkaConsumerConfig {@Beanpublic KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());ContainerProperties().setPollTimeout(1500);return factory;}public ConsumerFactory<String, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs());}public Map<String, Object> consumerConfigs() {Map<String, Object> properties = new HashMap<>();properties.put("bootstrap.servers", "kafka集群IP1:9092,kafka集群IP2:9092");properties.put("group.id", getIPAddress()); //获取服务器Ip作为groupIdproperties.put("enable.automit", "true");properties.put("automit.interval.ms", "1000");properties.put("set", "earliest");properties.put("session.timeout.ms", "30000");properties.put("key.deserializer", "org.apache.kafkamon.serialization.StringDeserializer");properties.put("value.deserializer", "org.apache.kafkamon.serialization.StringDeserializer");return properties;}public String getIPAddress() {try {InetAddress address = LocalHost();if (address != null && StringUtils.HostAddress())) {HostAddress();}}catch (UnknownHostException e) {return UUID.randomUUID().toString().replace("-","");}return UUID.randomUUID().toString().replace("-","");}/*** 自定义监听*/@Beanpublic MyKafkaListener listener() {return new MyKafkaListener();}
}

　　　　2.3.4、消费者监听

l.myListener;l.websocket.ChatWebsocket;
import org.apache.sumer.ConsumerRecord;
import org.apache.log4j.Logger;
import org.springframework.kafka.annotation.KafkaListener;/*** @Author: yxl* @Description:* @DATE: Created in 2018/11/14*/
public class MyKafkaListener {Logger logger = Logger(MyKafkaListener.class);/*** 发送聊天消息时的监听* @param record*/@KafkaListener(topics = {"chatMessage"})public void listen(ConsumerRecord<?, ?> record) {logger.info("chatMessage发送聊天消息监听："+record.value().toString());ChatWebsocket chatWebsocket = new ChatWebsocket();chatWebsocket.kafkaReceiveMsg(record.value().toString());}/*** 关闭连接时的监听* @param record*/@KafkaListener(topics = {"closeWebsocket"})private void closeListener(ConsumerRecord<?, ?> record) {logger.info("closeWebsocket关闭websocket连接监听："+record.value().toString());ChatWebsocket chatWebsocket = new ChatWebsocket();chatWebsocket.kafkaCloseWebsocket(record.value().toString());}}

　　　　2.3.6、Websocket集群java代码

package com.kk.server.chat.websocket;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apachemons.lang.StringUtils;
import org.apache.log4j.Logger;
import t.ApplicationContext;
import org.KafkaTemplate;
import org.springframework.stereotype.Component;import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import urrent.ConcurrentHashMap;/*** Websocket集群* Created by yxl on 2018-11-17.*/
@ServerEndpoint("/chat/{userId}")
@Component
public class ChatWebsocket {private Logger logger = Logger(ChatWebsocket.class);private static ApplicationContext applicationContext;private KafkaTemplate kafkaTemplate;//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。private static int onlineCount = 0;//concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识private static Map<String, Session> drWebSocketSet = new ConcurrentHashMap<>(); //医生web/*** 连接建立成功调用的方法** @param userId     用户标识*/@OnOpenpublic void onOpen(@PathParam("userId") String userId, Session session) {if (kafkaTemplate == null) {kafkaTemplate = Bean(KafkaTemplate.class); //获取kafka的Bean实例}drWebSocketSet.put(userId, session);}/*** s* 收到客户端消息后调用的方法** @param message 客户端发送过来的消息* @param session 可选的参数*/@OnMessagepublic void onMessage(String message, Session session) throws IOException {if ("ping".equals(message)) {BasicRemote().sendText("pong"); //心跳} else {sendMessage(message, session); //调用Kafka进行消息分发}}/*** 发送消息** @param message* @param session* @throws IOException*/public void sendMessage(String message, Session session) throws IOException {if (StringUtils.isNotBlank(message)) {JSONObject jsonObject = JSONObject.parseObject(message);String sender_id = String("sender_id"); //发送者IDString receiver_id = String("receiver_id"); //接受者ID//TODO 这里可以进行优化。可以首先根据接收方的userId,即receiver_id判断接收方是否在当前服务器，若在，直接获取session发送即可就不需要走Kafka了，节约资源kafkaTemplate.send("chatMessage", message);}}/*** 连接关闭调用的方法*/@OnClosepublic void onClose(Session session) {Map<String, String> pathParameters = PathParameters();String userId = ("userId"); //从session中获取userIdMap<String, String> map = new HashMap<>();map.put("username", userId);kafkaTemplate.send("closeWebsocket", JSONString(map));}
}/*** 关闭连接** @param map 当前登录客户端的map*/private void close(Map<String, Session> map, String username) {if (StringUtils.isNotBlank(username)) {logger.info("关闭websocket链接,关闭客户端username:" + username);if ((username) != null) {ve(username);}}}/*** kafka发送消息监听事件，有消息分发** @param message* @author yxl*/public void kafkaReceiveMsg(String message) {JSONObject jsonObject = JSONObject.parseObject(message);String receiver_id = String("receiver_id"); //接受者IDif ((receiver_id) != null) {(receiver_id).getBasicRemote.sendText(message); //进行消息发送}}/*** kafka监听关闭websocket连接** @param closeMessage*/public void kafkaCloseWebsocket(String closeMessage) {JSONObject jsonObject = JSONObject.parseObject(closeMessage);String userId = String("userId");ve(userId);}/*** 发生错误时调用** @param session* @param error*/@OnErrorpublic void onError(Session session, Throwable error) {logger.info("webscoket发生错误!关闭websocket链接");//onClose(session);error.printStackTrace();logger.info("webscoket发生错误!" + Message());}}

websocket中不能直接注入相应的Bean实例，这个时候可以看我的另一篇博客.html