使用Redis解决WebSocket分布式场景下的Session共享问题

您所在的位置：网站首页 › 数据共享问题解决方案 › 使用Redis解决WebSocket分布式场景下的Session共享问题

使用Redis解决WebSocket分布式场景下的Session共享问题

2024-07-15 22:46| 来源: 网络整理| 查看: 265

在显示项目中遇到了一个问题，需要使用到websocket与小程序建立长链接。由于项目是负载均衡的，存在项目部署在多台机器上。这样就会存在一个问题，当一次请求负载到第一台服务器时，socketsession在第一台服务器线程上，第二次请求，负载到第二台服务器上，需要通过id查找当前用户的session时，是查找不到的。

可以看到，由于websocket的session并没有实现序列化接口。所以无法将session序列化到redis中。web的中的httpsession 主要是通过下面的两个管理器实现序列化的。 org.apache.catalina.session.StandardManager org.apache.catalina.session.PersistentManager

复制

StandardManager是Tomcat默认使用的，在web应用程序关闭时，对内存中的所有HttpSession对象进行持久化，把他们保存到文件系统中。

默认的存储文件为：/work/Catalina///sessions.ser

PersistentManager比StandardManager更为灵活，只要某个设备提供了实现org.apache.catalina.Store接口的驱动类，PersistentManager就可以将HttpSession对象保存到该设备。

所以spring-session-redis 解决分布场景下的session共享就是将session序列化到redis中间件中，使用filter 加装饰器模式解决分布式场景httpsession 共享问题。

解决方案使用消息中间件解决websocket session共享问题。使用redis的发布订阅模式解决

本文使用方式二

使用StringRedisTemplate的convertAndSend方法向指定频道发送指定消息： this.execute((connection) -> { connection.publish(rawChannel, rawMessage); return null; }, true);

redis的命令publish channel message

添加一个监听的容器以及一个监听器适配器 package com.wdhcr.socket.config; import com.wdhcr.socket.constant.Constants; import org.springframework.cache.annotation.EnableCaching; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; /** * redis配置 * */ @Configuration @EnableCaching public class RedisConfig { @Bean RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapterWebSocket) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); // 可以添加多个 messageListener，配置不同的交换机 container.addMessageListener(listenerAdapterWebSocket, new PatternTopic(Constants.REDIS_CHANNEL));// 订阅最新消息频道 return container; } @Bean MessageListenerAdapter listenerAdapterWebSocket(RedisReceiver receiver) { // 消息监听适配器 return new MessageListenerAdapter(receiver, "onMessage"); } @Bean StringRedisTemplate template(RedisConnectionFactory connectionFactory) { return new StringRedisTemplate(connectionFactory); } }

添加消息接收器 package com.wdhcr.socket.config; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.wdhcr.socket.component.WebSocketServer; import com.wdhcr.socket.constant.Constants; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; /** * 消息监听对象，接收订阅消息 */ @Component public class RedisReceiver implements MessageListener { Logger log = LoggerFactory.getLogger(this.getClass()); @Autowired private WebSocketServer webSocketServer; /** * 处理接收到的订阅消息 */ @Override public void onMessage(Message message, byte[] pattern) { String channel = new String(message.getChannel());// 订阅的频道名称 String msg = ""; try { msg = new String(message.getBody(), Constants.UTF8);//注意与发布消息编码一致，否则会乱码 if (!StringUtils.isEmpty(msg)){ if (Constants.REDIS_CHANNEL.endsWith(channel))// 最新消息 { JSONObject jsonObject = JSON.parseObject(msg); webSocketServer.sendMessageByWayBillId( Long.parseLong(jsonObject.get(Constants.REDIS_MESSAGE_KEY).toString()) ,jsonObject.get(Constants.REDIS_MESSAGE_VALUE).toString()); }else{ //TODO 其他订阅的消息处理 } }else{ log.info("消息内容为空，不处理。"); } } catch (Exception e) { log.error("处理消息异常："+e.toString()); e.printStackTrace(); } } }

websocket的配置类 package com.wdhcr.socket.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.server.standard.ServerEndpointExporter; /** * @description: websocket的配置类 */ @Configuration @EnableWebSocket public class WebSocketConfiguration { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }

添加websocket的服务组件 package com.wdhcr.socket.component; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.wdhcr.socket.constant.Constants; import com.wdhcr.socket.utils.SpringUtils; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.io.IOException; import java.io.UnsupportedEncodingException; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; @ServerEndpoint("/websocket/{id}") @Component public class WebSocketServer { private static final long sessionTimeout = 600000; private static final Logger log = LoggerFactory.getLogger(WebSocketServer.class); /** * 当前在线连接数 */ private static AtomicInteger onlineCount = new AtomicInteger(0); /** * 用来存放每个客户端对应的 WebSocketServer 对象 */ private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap(); /** * 与某个客户端的连接会话，需要通过它来给客户端发送数据 */ private Session session; /** * 接收 id */ private Long id; /** * 连接建立成功调用的方法 */ @OnOpen public void onOpen(Session session, @PathParam("id") Long id) { session.setMaxIdleTimeout(sessionTimeout); this.session = session; this.id = id; if (webSocketMap.containsKey(id)) { webSocketMap.remove(id); webSocketMap.put(id, this); } else { webSocketMap.put(id, this); addOnlineCount(); } log.info("编号id:" + id + "连接,当前在线数为:" + getOnlineCount()); try { sendMessage("连接成功！"); } catch (IOException e) { log.error("编号id:" + id + ",网络异常!!!!!!"); } } /** * 连接关闭调用的方法 */ @OnClose public void onClose() { if (webSocketMap.containsKey(id)) { webSocketMap.remove(id); subOnlineCount(); } log.info("编号id:" + id + "退出,当前在线数为:" + getOnlineCount()); } /** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息 */ @OnMessage public void onMessage(String message, Session session) { JSONObject jsonObject = JSON.parseObject(message); String key = jsonObject.get(Constants.REDIS_MESSAGE_KEY).toString(); String value = jsonObject.get(Constants.REDIS_MESSAGE_VALUE).toString(); sendMessage(key,value); log.info("编号id消息:" + id + ",报文:" + message); } /** * 发生错误时调用 * * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.error("编号id错误:" + this.id + ",原因:" + error.getMessage()); error.printStackTrace(); } /** * @description: 分布式使用redis 去发布消息 * @dateTime: 2021/6/17 10:31 */ public void sendMessage(@NotNull String key, String message) { String newMessge= null; try { newMessge = new String(message.getBytes(Constants.UTF8), Constants.UTF8); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } Map map = new HashMap(); map.put(Constants.REDIS_MESSAGE_KEY, key); map.put(Constants.REDIS_MESSAGE_VALUE, newMessge); /** * * spring管理的都是单例（singleton）和 websocket （多对象）相冲突。 * * 需要了解一个事实：websocket 是多对象的，每个用户的聊天客户端对应 java 后台的一个 websocket 对象，前后台一对一（多对多）实时连接， * 所以 websocket 不可能像 servlet 一样做成单例的，让所有聊天用户连接到一个 websocket对象，这样无法保存所有用户的实时连接信息。 * 可能 spring 开发者考虑到这个问题，没有让 spring 创建管理 websocket ，而是由 java 原来的机制管理websocket ，所以用户聊天时创建的 * websocket 连接对象不是 spring 创建的，spring 也不会为不是他创建的对象进行依赖注入，所以如果不用static关键字，每个 websocket 对象的 service 都是 null。 */ StringRedisTemplate template = SpringUtils.getBean(StringRedisTemplate.class); template.convertAndSend(Constants.REDIS_CHANNEL, JSON.toJSONString(map)); } /** * @description: 单机使用外部接口通过指定的客户id向该客户推送消息。 * @dateTime: 2021/6/16 17:49 */ public void sendMessageByWayBillId(@NotNull Long key, String message) { WebSocketServer webSocketServer = webSocketMap.get(key); if (!StringUtils.isEmpty(webSocketServer)) { try { webSocketServer.sendMessage(message); log.info("编号id为："+key+"发送消息："+message); } catch (IOException e) { e.printStackTrace(); log.error("编号id为："+key+"发送消息失败"); } } log.error("编号id号为："+key+"未连接"); } /** * 实现服务器主动推送 */ public void sendMessage(String message) throws IOException { this.session.getBasicRemote().sendText(message); } public static synchronized AtomicInteger getOnlineCount() { return onlineCount; } public static synchronized void addOnlineCount() { WebSocketServer.onlineCount.getAndIncrement(); } public static synchronized void subOnlineCount() { WebSocketServer.onlineCount.getAndDecrement(); } }

项目结构