在websocket中，服务端主要使用的是session打交道，但是由于session无法实现序列化，不能存储到redis这些中间存储里面，因此这里我们只能把session存储在本地的内存中，那么如果是集群的话，我们如何实现session准确的发送消息呢，其实就是session共享。在websocket中，其实是无法做到session共享的，目前通用的解决方案都是通过消息中间件，实现消息的发布与订阅，也就是每一个服务端实例都订阅某个消息队列的topic，根据对应的sessionid来判断是否在本地存储，如果在本地通过sessionid找到了session，则给客户端发送消息，如果在本地找不到对应的session，那么就直接把这条消息丢弃掉。具体的如下图所示：
这里的图来自于网上，网上大多都是基于redis做发布与订阅，在真实的环境中，我们一般用kafka或者rocketmq等。根据上面的图示，我们介绍下整个流程：
1、我们同时有A,B,C,D四个websocket服务端，同时订阅消息队列的topic: test8
2、我们发送一条消息a1到消息队列的topic：test8
3、此时A,B,C,D四个websocket服务端都会收到这条消息a1
4、A根据a1的消息体，获取到对应的sessionid，然后在本地的map中查找是否有对应的session，如果没有直接放弃掉此条消息。
5、B根据a1的消息体，获取到对应的sessionid，然后在本地的map中查找是否有对应的session，如果没有直接放弃掉此条消息。
6、C根据a1的消息体，获取到对应的sessionid，然后在本地的map中查找是否有对应的session，如果没有直接放弃掉此条消息。
7、D根据a1的消息体，获取到对应的sessionid，然后在本地的map中查找是否有对应的session，结果找到有对应的session，此时我们就把这条消息发送给=这个session。
8、客户端就收到了对应的消息。一、创建一个公共的map，用来存放session
package com.websocket.utils;import java.util.concurrent.ConcurrentHashMap;import javax.websocket.Session;import org.springframework.stereotype.Component;@Component
public class OnlineSessionCache {private ConcurrentHashMap<Integer, Session> onlines = new ConcurrentHashMap<Integer, Session>();public void setUserSession(Integer userId, Session session) {onlines.put(userId, session);}public Session getUserSession(Integer userId) {return onlines.get(userId);}public void removeUserSession(Integer userId) {onlines.remove(userId);}public ConcurrentHashMap<Integer, Session> getAllSession() {return this.onlines;}}二、在websocket连接和关闭的时候，把session关闭掉
@OnOpenpublic void onOpen(Session session,EndpointConfig config) {this.session = session;log.info("当前session id ： {} 登录进来了", session.getId());OnlineCalUtils.addOnlineCount();onlineSessionCache.setUserSession(Integer.valueOf(session.getId()), session);log.info("存储session了多少个session:{}", onlineSessionCache.getAllSession().size());log.info("有新连接加入！当前在线人数为 :{} ", getOnlineCount());}@OnClosepublic void onClose() {OnlineCalUtils.subOnlineCount();log.info("有一连接关闭！当前在线人数为: {}", getOnlineCount());onlineSessionCache.removeUserSession(Integer.valueOf(this.session.getId()));log.info("当前session id ： {} 退出去了");}三、编写一个接口，用来给指定的用户发送消息
package com.websocket.controller;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import com.websocket.model.ChatModel;
import com.websocket.producer.RocketProducer;
import com.websocket.utils.ChatModelUtils;import lombok.extern.slf4j.Slf4j;@RestController
@Slf4j
public class ChatMsgController {@Autowiredprivate RocketProducer rocketProducer;@RequestMapping("/sendToSimpleUser")public String sendToSimpleUser(Integer fromUserId,Integer toUserId) {ChatModel model = ChatModelUtils.createNewChatModel(fromUserId, toUserId, "手动发送消息");rocketProducer.sendDirectMessage(model);return "成功";}}这里我们是把消息直接发送给了rocketmq里面，发送者代码如下：
package com.websocket.producer;import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import com.alibaba.fastjson.JSON;
import com.websocket.model.ChatModel;@Component
public class RocketProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;public void sendDirectMessage(ChatModel message) {String msg = JSON.toJSONString(message);rocketMQTemplate.syncSend("test8", msg);}}四、编写消费者，获取mq的消息，并且发送消息给对应的session
package com.websocket.producer;import javax.websocket.Session;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import com.alibaba.fastjson.JSON;
import com.websocket.model.ChatModel;
import com.websocket.product.SocketServerProduct;
import com.websocket.utils.OnlineSessionCache;import lombok.extern.slf4j.Slf4j;@Component
@Slf4j
@RocketMQMessageListener(topic = "test8", consumerGroup = "${chat.group.groupname}")
public class RocketConsumer implements RocketMQListener<String>{@Autowiredprivate OnlineSessionCache onlineSessionCache;@Autowiredprivate SocketServerProduct socketServerProduct;@Value("${chat.group.groupname}")private String groupName;@Overridepublic void onMessage(String message) {log.info("监听到的topic是：{} groupname是：{}","test8",groupName);ChatModel model = JSON.parseObject(message, ChatModel.class);Integer userId = model.getToUserId();Session session = onlineSessionCache.getUserSession(userId);if (null != session) {log.info("找到了对应的session，准备回复消息");socketServerProduct.sendMessage(session, model.getMessage());}else {log.info("没有找到对应的session，准备丢弃");}}
}以上就是一个完整的关于websocket服务端集群关于session共享的解决方案。
WebSocket服务端数据推送及心跳机制(Spring Boot + VUE)_websocket心跳机制-CSDN博客
