Spring WebSocket通信应用二[基于Redis实现Ws分布式]

概要

Spring WebSocket通信应用-后续
基于Redis实现Spring WebSocket分布式通信。
环境说明：spring-boot-2.5.6、springframework-5.3.12，spring-websocket-5.3.12、spring-data-redis-2.5.6
框架说明：基于 若依框架
采用技术：springboot、websocket、redis发布订阅

需求

实现服务端多节点的情况下，主动推送消息到客户端（即设备1、2、…N）

核心流程说明

1)客户端与服务端建立WebSocket通信，内存集合public final ConcurrentHashMap<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();存储WebSocketSession sessionId;
redis同步存储,格式<sessionId,sessionId+YYYYMMDDHHMMSS>
2)客户端（即设备1、2、…N，设备的id为设备唯一性属性deviceId）向服务端多节点注册[“/ems/robot/register”]，设备的请求参数有sessionId、deviceId;
redis同步存储设备和WebSocketSession正反双向信息,
格式:[deviceidandsessionid:{sessionId},{deviceId}]、[deviceidandsessionid:{deviceId},{sessionId}];
3)服务端某节点接收请求消息，根据订阅的不同主题向所有服务节点发布消息
4)服务端消息处理器RedisReceiver接收发布消息，开始在每个服务节点处理
5)服务节点根据sessionId匹配内存集合ConcurrentHashMap<String, WebSocketSession> sessionMap ，命中后获取对应WebSocketSession
6)服务节点根据sessionId匹配deviceId，找到设备，并主动推送消息。

websocket配置

参考Spring WebSocket通信应用

Redis发布订阅

提示：redis提供publish/subscribe命令来完成发布、订阅操作,利用此机制，就可以基本实现多节点下的websocket通信

redis缓存配置
具体见RedisCache.java，忽略。

StringRedisTemplate模板配置-RedisConfig.java
配置redis、StringRedisTemplate、增加消息监听器容器绑定主题、消息监听器适配器绑定消息处理器RedisReceiver

package com.ems.mgr.framework.config;
import com.ems.mgr.common.constant.RedisConstants;
import com.ems.mgr.framework.receiver.RedisReceiver;
import org.springframework.cache.annotation.CachingConfigurerSupport;
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.RedisTemplate;
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;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator;
/**
 * redis配置
 */
@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport
{
    @Bean
    @SuppressWarnings(value = { "unchecked", "rawtypes" })
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory)
    {
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);

        FastJson2JsonRedisSerializer serializer = new FastJson2JsonRedisSerializer(Object.class);

        ObjectMapper mapper = new ObjectMapper();
        mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        mapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
        serializer.setObjectMapper(mapper);

        // 使用StringRedisSerializer来序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(serializer);

        // Hash的key也采用StringRedisSerializer的序列化方式
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(serializer);

        template.afterPropertiesSet();
        return template;
    }

    /**
     * @description:增加redis不同主题的消息监听器容器
     * 添加多个监听不同主题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定
     * 该消息监听器通过反射技术调用消息订阅处理器的相关方法进行业务处理
     * @param: [connectionFactory, listenerAdapter]
     * @return: RedisMessageListenerContainer
     */
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 1订阅redis发布响应设备消息通知[onMessage]
        container.addMessageListener(listenerAdapter, new PatternTopic(RedisConstants.REDIS_CHANNEL));
        // 2订阅redis发布Websocket断开连接通知[close]
        container.addMessageListener(listenerAdapter,new PatternTopic(RedisConstants.REDIS_CHANNEL_CLOSE));
        // 3订阅redis发布设备消息通知[send]
        container.addMessageListener(listenerAdapter,new PatternTopic(RedisConstants.REDIS_CHANNEL_SEND));
        //4订阅删除Websocket和Redis内存信息消息通知[websocketremove]
        container.addMessageListener(listenerAdapter, new PatternTopic(RedisConstants.WEBSOCKETSESSIONID_REMOVE));
        return container;
    }

    // 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法
    @Bean
    MessageListenerAdapter listenerAdapter(RedisReceiver receiver) {
        // 消息监听适配器
        return new MessageListenerAdapter(receiver, "onMessage");
    }

    @Bean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory)
    {
        return new StringRedisTemplate(connectionFactory);
    }
}

消息处理器RedisReceiver
服务端接收订阅消息，按不同类型主题分别处理。

package com.ems.mgr.framework.receiver;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ems.mgr.common.constant.RedisConstants;
import com.ems.mgr.websocket.commons.JacksonUtil;
import com.ems.mgr.websocket.handler.MerakSocketHandler;
import com.ems.mgr.websocket.model.MessageBodyBean;
import com.ems.mgr.websocket.model.WsResponse;
import com.fasterxml.jackson.core.type.TypeReference;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;

import javax.annotation.Resource;

/**
 * @author harry
 * @version 1.0
 * @description: 消息监听对象，接收订阅消息
 */
@Component
public class RedisReceiver implements MessageListener {
    Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private MerakSocketHandler webSocketServer;

    /**
     * 处理接收到的订阅消息
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 获取订阅的频道名称
        String channel = new String(message.getChannel());
        String msg = "";
        try {
            // 注意与发布消息编码一致，否则会乱码
            msg = new String(message.getBody(), RedisConstants.UTF8);
            if (StringUtils.isNotEmpty(msg)) {
                if (channel.endsWith(RedisConstants.REDIS_CHANNEL_CLOSE)) {
                    log.info("订阅redis发布Websocket断开连接通知[close],消息体:" + msg);
                    // websocket连接关闭的消息-调用关闭接口
                    webSocketServer.closeMessageBySessionId(msg);
                } else if (channel.endsWith(RedisConstants.REDIS_CHANNEL_SEND)) {
                    MessageBodyBean wsParam = JacksonUtil.json2Bean(msg, new TypeReference<MessageBodyBean>() {
                    });
                    log.info("订阅redis发布设备消息通知[send],消息体:" + JacksonUtil.bean2Json(wsParam));
                    // 向客户端推向消息的接口
                    String sessionId = wsParam.getSessionId();
                    WsResponse<MessageBodyBean> response = new WsResponse<>();
                    response.setResult(wsParam);
                    webSocketServer.sendMessageBySessionId(sessionId, new TextMessage(JacksonUtil.bean2Json(response)));
                } else if (channel.endsWith(RedisConstants.REDIS_CHANNEL)) {
                    log.info("订阅redis发布响应设备消息通知[onMessage],消息体: ........");
                    MessageBodyBean wsParam = JacksonUtil.json2Bean(msg, new TypeReference<MessageBodyBean>() {
                    });
                    // 向客户端推向消息的接口
                    String sessionId = wsParam.getSessionId();
                    WsResponse<MessageBodyBean> response = new WsResponse<>();
                    response.setResult(wsParam);
                    webSocketServer.sendMessageBySessionId(sessionId, new TextMessage(JacksonUtil.bean2Json(response)));
                }else if (channel.endsWith(RedisConstants.WEBSOCKETSESSIONID_REMOVE)) {
                    //删除Websocket内存集合信息
                    log.info("订阅消息-websocketremove,删除Websocket和Redis内存信息消息通知[websocketremove],消息体:" + msg);
                    JSONObject jsonObject = JSON.parseObject(msg);
                    webSocketServer.removeWebsocketBySessionId(jsonObject.get("sessionId").toString(),jsonObject.get("deviceId").toString());
                }
            } else {
                log.info("消息内容为空，不处理。");
            }
        } catch (Exception e) {
            log.error("处理消息异常：" + e.getMessage());
        }
    }
}

典型场景：redis发布设备响应消息通知
流程：设备（客户端）向某个服务节点（如node2）发送消息------服务节点基于redis特性发布设备响应消息通知------消息处理器RedisReceiver接收订阅消息、处理【所有服务节点】-----------某个服务节点（如node1）根据sessionId匹配内存集合ConcurrentHashMap<String, WebSocketSession> sessionMap ，命中后获取对应WebSocketSession-----------服务节点node1向设备（客户端）响应消息

某个服务节点（如node2）接收设备（客户端）消息：

@Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String sessionId = session.getId();
        logger.info("handleTextMessage from robot...,sessionId:" + sessionId);
        String msg = message.getPayload();
        logger.info("handleTextMessage msg:" + msg);
        MessageBodyBean wsParam = JacksonUtil.json2Bean(msg, new TypeReference<MessageBodyBean>() {
        });
        if ("signaling".equals(wsParam.getMessageType()) && "heartbeat".equals(wsParam.getMessageContext())) {
            logger.info("handleTextMessage-Robot心跳监测响应");
            webSocketSessionHandler.subtractHeartbeatMonitorMap(sessionId);
        } else {
            logger.info("handleTextMessage-普通响应");
//            WsResponse<MessageBodyBean> response = new WsResponse<>();
            MessageBodyBean messageBodyBean = new MessageBodyBean();
            messageBodyBean.setSessionId(sessionId);
            messageBodyBean.setMessageType("signaling");
            messageBodyBean.setMessageContext("response");
//            response.setResult(messageBodyBean);
//            ConcurrentWebSocketSessionDecorator sessionDecorator = (ConcurrentWebSocketSessionDecorator) webSocketSessionHandler.getSession(sessionId);
//            sendMessageToUser(sessionDecorator, new TextMessage(JacksonUtil.bean2Json(response)));
            publishSendMessage(messageBodyBean);
        }
        logger.info("handleTextMessage end...");
    }

服务节点基于redis特性RedisTemplate.convertAndSend(String channel, Object message) API方法，发布设备响应消息通知

/**
     * @description:分布式-使用redis发布设备响应消息通知
     */
    public void publishSendMessage(MessageBodyBean wsParam) {
        try {
            logger.info("publishSendMessage:redis发布响设备应消息通知,sessionId:" + wsParam.getSessionId());
            StringRedisTemplate template = SpringUtils.getBean(StringRedisTemplate.class);
            template.convertAndSend(RedisConstants.REDIS_CHANNEL, JSON.toJSONString(wsParam));
            logger.info("publishSendMessage:redis发布响应设备消息通知,sessionId:" + wsParam.getSessionId());
        } catch (Exception e) {
            logger.error("publishSendMessage:redis发布响应设备消息通知异常,sessionId:" + wsParam.getSessionId() + ",error:" + e.getMessage());
        }
    }

消息处理器RedisReceiver接收订阅消息、处理【所有服务节点】

/**
     * 处理接收到的订阅消息
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 获取订阅的频道名称
        String channel = new String(message.getChannel());
        String msg = "";
        try {
            // 注意与发布消息编码一致，否则会乱码
            msg = new String(message.getBody(), RedisConstants.UTF8);
            if (StringUtils.isNotEmpty(msg)) {
                if (channel.endsWith(RedisConstants.REDIS_CHANNEL_CLOSE)) {
                   。。。
                } else if (channel.endsWith(RedisConstants.REDIS_CHANNEL_SEND)) {
                   。。。
                } else if (channel.endsWith(RedisConstants.REDIS_CHANNEL)) {
                    log.info("订阅redis发布响应设备消息通知[onMessage],消息体: ........");
                    MessageBodyBean wsParam = JacksonUtil.json2Bean(msg, new TypeReference<MessageBodyBean>() {
                    });
                    // 向客户端推向消息的接口
                    String sessionId = wsParam.getSessionId();
                    WsResponse<MessageBodyBean> response = new WsResponse<>();
                    response.setResult(wsParam);
                    webSocketServer.sendMessageBySessionId(sessionId, new TextMessage(JacksonUtil.bean2Json(response)));
                }else if (channel.endsWith(RedisConstants.WEBSOCKETSESSIONID_REMOVE)) {
                   。。。
                }
            } else {
                log.info("消息内容为空，不处理。");
            }
        } catch (Exception e) {
            log.error("处理消息异常：" + e.getMessage());
        }
    }

某个服务节点（如node1）根据sessionId匹配内存集合ConcurrentHashMap<String, WebSocketSession> sessionMap ，命中后获取对应WebSocketSession

 /**
     * @description: 外部接口通过指定的sessionId向该设备推送消息
     */
    public void sendMessageBySessionId(String sessionId, TextMessage message) {
        logger.info("sendMessageBySessionId start,sessionId:" + sessionId);
        try {
            ConcurrentWebSocketSessionDecorator sessionDecorator = (ConcurrentWebSocketSessionDecorator) webSocketSessionHandler.getSession(sessionId);
            if (null != sessionDecorator) {
                if (sessionDecorator.isOpen()) {
                    sessionDecorator.sendMessage(message);
                    logger.info("redis发布设备消息通知:匹配指定设备,sessionId=" + sessionId);
                } else {
                    logger.info("redis发布设备消息通知:匹配指定设备已经离线,sessionId=" + sessionId);
                }
            } else {
                logger.info("redis发布设备消息通知:其它指定设备,sessionId=" + sessionId);
            }
            logger.info("sendMessageBySessionId end,sessionId:" + sessionId);
        } catch (IOException e) {
            logger.error("sendMessageBySessionId error,sessionId:" + sessionId + ",error:" + e.getMessage());
        }
    }

服务节点node1向设备（客户端）响应消息

sessionDecorator.sendMessage(message);

源代码

源代码下载