redis|springboot使用redisTemplate+websocket实现集群消息的发布订阅 boot|websocket|spring

本文主要介绍小项目中如何通过redisTemplate+websocket实现后端向前端推送消息
一、websocket配置类

@Component public class WebsocketConfig { /** * ServerEndpointExporter 作用 * * 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint * * @return */ @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }

二、websocket接口类

@Component @ServerEndpoint(value = "https://www.it610.com/websocket") public class WebSocket {private static final Logger logger = LoggerFactory.getLogger(WebSocket.class); // 每个客户端都会有相应的session,服务端可以发送相关消息 private Session session; // 此处定义静态变量，以在其他方法中获取到所有连接 public static CopyOnWriteArraySet wbSockets = new CopyOnWriteArraySet(); /** * 建立连接。 * 建立连接时入参为session */ @OnOpen public void onOpen(Session session) { this.session = session; wbSockets.add(this); //将此对象存入集合中以在之后广播用，如果要实现一对一订阅，则类型对应为Map。由于这里广播就可以了随意用Set logger.info("New session insert,sessionId is {}", session.getId()); }/** * 关闭连接 */ @OnClose public void onClose() { wbSockets.remove(this); //将socket对象从集合中移除，以便广播时不发送次连接。如果不移除会报错(需要测试) logger.info("A session insert,sessionId is {}", session.getId()); }/** * 接收消息 */ @OnMessage public void OnMessage(String message, Session session) { logger.info("A message insert,sessionId is {}", session.getId()); }/** * 发送消息 */ public void sendMessage(Object message) throws Exception { //遍历客户端 for (WebSocket webSocket : wbSockets) { logger.info("websocket info is {}", IJsonUtil.obj2Json(message)); //服务器主动推送 webSocket.session.getBasicRemote().sendObject(message); } }/** * 发送消息 */ public void sendMessage(String message) throws Exception { //遍历客户端 for (WebSocket webSocket : wbSockets) { logger.info("websocket info is {}", message); //服务器主动推送 webSocket.session.getBasicRemote().sendText(message); } }/** * 发送消息（用于发送给指定客户端消息） */ public void sendMessage(String sessionId, String message) throws Exception { Session session = null; WebSocket tempWebSocket = null; logger.info("websocket info is {}", IJsonUtil.obj2Json(message)); //遍历客户端 for (WebSocket webSocket : wbSockets) { if (webSocket.session.getId().equals(sessionId)) { tempWebSocket = webSocket; session = webSocket.session; break; } } if (session != null) { tempWebSocket.session.getBasicRemote().sendText(message); } else { logger.info("没有找到你指定ID的会话:{}", sessionId); } } }

三、redis消息接收处理类

@Component public class RedisListenerHandle {private static final Logger logger = LoggerFactory.getLogger(RedisListenerHandle.class); @Autowired private WebSocket webSocket; /** * @param message * @description 接收到消息的业务处理 * @create 2020-06-23 17:08:21 */ public void receiveMessage(String message) { logger.info("RedisListenerHandle get message info:{}", message); try { webSocket.sendMessage(message); logger.info("RedisListenerHandle push message success:{}", InetAddress.getLocalHost().getHostAddress()); } catch (Exception e) { logger.info("RedisListenerHandle push message exception", e); } } }

四、消息监听类

@Component public class RedisListenerBean { /** * redis消息监听器容器 * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器 * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理 * @param connectionFactory * @return */ @Bean public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.addMessageListener(listenerAdapter, new PatternTopic(CommonConstants.REDIS_CHANNEL)); return container; }@Bean public MessageListenerAdapter messageListenerAdapter(RedisListenerHandle redisListenerHandle){ return new MessageListenerAdapter(redisListenerHandle,"receiveMessage"); } }

再通过redisTemplate.convertAndSend()即可实现消息推送

总结：
1、为什么websocket不支持集群？
假如有A，B两台应用服务器，用户A与服务器A建立了连接，用户B与服务器B建立了连接，当要把某一消息发送给A，B两个用户时，如果是应用A在处理发送请求，那么用户A可以接收到消息，用户B则不能收到。如果是应用B在处理发送请求，那么用户B可以接收到消息，用户A则不能收到。这是因为websocket只能与一台服务器建立连接，而与哪台服务器建立连接又是通过nginx做的负载均衡。
2、redis实现websocket集群原理？
通过上面那个问题我们知道websocket只能与一台应用服务器建立连接，那么要实现消息推送给所有用户，无非就两种方案：
a、服务器之间共享websocket连接
b、让所有服务应用都发送消息
那么这里采用的则是第二种方案。先让A、B两台服务器都监听同一个topic，然后无论是A还是B服务处理发送请求，都把消息放到redis中，然后A、B监听到消息，都会处理消息，再通过websocket发送给各自连接的用户。
3、替代方案
上述所说的可以替换成rabbitmq，如果是并发较高的可以替换成rocketmq、kafka等吞吐量更好的消息推送服务。当然并发高的时候换成rocketmq或者kafka也需要考虑redis能不能承受的了。but，前端也有更好的推送技术，比如以mqtt协议实现推送的activeMq等等。

【redis|springboot使用redisTemplate+websocket实现集群消息的发布订阅】