SpringBoot+WebSocket基於Redis訂閱發布實現集群化

SpringBoot+WebSocket基於Redis訂閱發布實現集群化

前面講了單機版的websocket如何使用發送群聊(2020-03-24-springboot快速集成websocket實現群聊)，那么要是部署多個服務實現集群話怎么實現呢？

由於websocket是長連接，session保持在一個server中，所以在不同server在使用websocket推送消息時就需要獲取對應的session進行推送，在分布式系統中就無法獲取到所有session，這里就需要使用一個中間件將消息推送到各個系統中，在這里使用的redis，使用redis的sub/pub功能。

實現步驟

• Redis要配置消息監聽容器

package cn.pconline.pcloud.admin.config;

import cn.pconline.pcloud.admin.mq.ChatMessageListener;
import cn.pconline.pcloud.base.util.RedisKey;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

/**
 * @Description 消息訂閱配置類
 * @Author jie.zhao
 * @Date 2020/3/31 13:54
 */
@Configuration 
public class RedisSubscriberConfig {
    /**
     * 消息監聽適配器，注入接受消息方法
     *
     * @param receiver
     * @return
     */
    @Bean
    public MessageListenerAdapter messageListenerAdapter(ChatMessageListener receiver) {
        return new MessageListenerAdapter(receiver);
    }

    /**
     * 創建消息監聽容器
     *
     * @param redisConnectionFactory 
     * @param messageListenerAdapter2
     * @return
     */
    @Bean
    public RedisMessageListenerContainer getRedisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory, MessageListenerAdapter messageListenerAdapter) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
        redisMessageListenerContainer.addMessageListener(messageListenerAdapter, new PatternTopic(RedisKey.REDIS_MQ_CHAT));
        return redisMessageListenerContainer;
    }
}

• Redis消息處理類

收到消息后，給當前Server的session發送消息

package cn.pconline.pcloud.admin.mq;

import cn.pconline.framework.util.StringUtils;
import cn.pconline.pcloud.admin.ws.ChatWebsocketEndpoint;
import cn.pconline.pcloud.base.dto.MessageDto;
import cn.pconline.pcloud.base.util.JsonUtils;
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.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

/**
 * @Description 集群聊天消息監聽器
 * @Author jie.zhao
 * @Date 2020/3/29 15:07
 */
@Component
public class ChatMessageListener implements MessageListener {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private ChatWebsocketEndpoint websocketEndpoint;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        RedisSerializer<String> valueSerializer = redisTemplate.getStringSerializer();
        String value = valueSerializer.deserialize(message.getBody());

        if (StringUtils.isNotBlank(value)) {
            MessageDto dto = JsonUtils.jsonToPojo(value, MessageDto.class);
            logger.info("監聽集群websocket消息--- {}", value);
            //集群模式 推送消息
            websocketEndpoint.sendClusterWebsocketMessage(dto);
        }
    }

}

• websoketEndpoint修改

原先直接發消息的地方，改為發送Rdis訂閱消息。

package cn.pconline.pcloud.admin.ws;

import cn.pconline.pcloud.base.dto.MessageDto;
import cn.pconline.pcloud.base.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
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.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Description 群聊websocket改造為集群模式
 * @Author jie.zhao
 * @Date 2020/3/29 15:07
 */
@Component
@ServerEndpoint("/groupChat/{groupNo}/{uuid}/{name}")
public class ChatWebsocketEndpoint {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private StringRedisTemplate stringRedisTemplate = SpringBeanUtils.getBean(StringRedisTemplate.class);

    private RedisTemplateUtil redisTemplateUtil = SpringBeanUtils.getBean(RedisTemplateUtil.class);

    /**
     * 保存 組id->組成員 的映射關系
     * 之所以使用ConcurrentHashMap因為這個是線程安全的
     */
    private static ConcurrentHashMap<String, List<Session>> groupMemberInfoMap = new ConcurrentHashMap<>();

    /**
     * 收到消息調用的方法，群成員發送消息
     *
     * @param groupNo
     * @param uuid
     * @param message
     */
    @OnMessage
    public void onMessage(Session session, @PathParam("groupNo") String groupNo, @PathParam("uuid") String uuid,
                          @PathParam("name") String name, String message) {
        //前端心跳檢測
        if("PING".equals(message)){
            try {
                session.getBasicRemote().sendText("PING");
            } catch (Exception e) {
                logger.info("會話異常！" + e.getMessage());
                e.printStackTrace();
            }
        }else{
            //message過濾html、script、css標簽
            message = HtmlUtil.htmlEncode(message);
            MessageDto messageDto = new MessageDto(groupNo, MessageDto.MessageTypeEnum.CHAT, uuid, name, message, DateUtil.format(new Date()));
            this.sendClusterMessage(messageDto);
            this.writeRedisChatList(messageDto);
        }
    }

    /**
     * 建立連接調用的方法，群成員加入
     *
     * @param session 會話
     * @param groupNo 群id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("groupNo") String groupNo, @PathParam("uuid") String uuid, @PathParam("name") String name) {
        List<Session> sessionList = groupMemberInfoMap.get(groupNo);
        if (sessionList == null) {
            sessionList = new ArrayList<>();
            groupMemberInfoMap.put(groupNo, sessionList);
        }
        sessionList.add(session);
        this.addClusterSessionCount(groupNo, uuid);
        long onlineNum = this.getClusterSessionCount(groupNo);
        MessageDto messageDto = new MessageDto(groupNo, MessageDto.MessageTypeEnum.SYSTEM, uuid, name, "連接建立", DateUtil.format(new Date()), onlineNum);
        this.sendClusterMessage(messageDto);
        logger.info("連接建立");
        logger.info("直播房間號: {},當前集群在線人數:{} ", groupNo, onlineNum);
    }

    /**
     * 關閉連接調用的方法，群成員退出
     *
     * @param session
     * @param groupNo
     */
    @OnClose
    public void onClose(Session session, @PathParam("groupNo") String groupNo, @PathParam("uuid") String uuid, @PathParam("name") String name) {
        List<Session> sessionList = groupMemberInfoMap.get(groupNo);
        sessionList.remove(session);
        this.removeClusterSessionCount(groupNo, uuid);
        long onlineNum = this.getClusterSessionCount(groupNo);
        MessageDto messageDto = new MessageDto(groupNo, MessageDto.MessageTypeEnum.SYSTEM, uuid, name, "連接關閉", DateUtil.format(new Date()), onlineNum);
        this.sendClusterMessage(messageDto);
        logger.info("連接關閉");
        logger.info("直播房間號: {},當前集群在線人數:{} ", groupNo, onlineNum);
    }

    /**
     * 傳輸消息錯誤調用的方法
     *
     * @param error
     */
    @OnError
    public void OnError(Throwable error) {
        logger.info("連接出錯：{}", error.getMessage());
    }

    /**
     * 發布集群的消息
     *
     * @param dto
     */
    private void sendClusterMessage(MessageDto dto) {
        stringRedisTemplate.convertAndSend(RedisKey.REDIS_MQ_CHAT, JsonUtils.objectToJson(dto));
    }

    /**
     * 添加集群的在線人數
     *
     * @param groupNo
     * @param uuid
     * @return
     */
    private void addClusterSessionCount(String groupNo, String uuid) {
        redisTemplateUtil.sSet(RedisKey.REDIS_MQ_CHAT_SESSION_ID + groupNo, uuid);
    }

    /**
     * 移除集群的在線人數
     *
     * @param groupNo
     * @param uuid
     * @return
     */
    private void removeClusterSessionCount(String groupNo, String uuid) {
        redisTemplateUtil.setRemove(RedisKey.REDIS_MQ_CHAT_SESSION_ID + groupNo, uuid);
    }

    /**
     * 獲取集群的在線人數
     *
     * @param groupNo
     * @return
     */
    private long getClusterSessionCount(String groupNo) {
        return redisTemplateUtil.sGetSetSize(RedisKey.REDIS_MQ_CHAT_SESSION_ID + groupNo);
    }

    /**
     * 將聊天內容寫入redis，定時同步mysql
     *
     * @param dto
     * @return
     */
    private void writeRedisChatList(MessageDto dto) {
        if (dto != null) {
            redisTemplateUtil.lSet(RedisKey.REDIS_CHAT_LIST + dto.getGroupNo(), JsonUtils.objectToJson(dto));
            //雙寫模式 同時寫入歷史消息
            redisTemplateUtil.lSet(RedisKey.REDIS_CHAT_LIST_HISTORY + dto.getGroupNo(), JsonUtils.objectToJson(dto));
        }
    }

    /**
     * 發布websocket消息
     *
     * @param dto
     * @return
     */
    public void sendClusterWebsocketMessage(MessageDto dto) {
        if (dto != null) {
            //得到當前群的所有會話，也就是所有用戶
            List<Session> sessionList = groupMemberInfoMap.get(dto.getGroupNo());

            if (sessionList != null && sessionList.size() > 0) {
                // 遍歷Session集合給每個會話發送文本消息
                sessionList.forEach(item -> {
                    try {
                        item.getBasicRemote().sendText(JsonUtils.objectToJson(dto));
                    } catch (Exception e) {
                        logger.info("會話異常！" + e.getMessage());
                        e.printStackTrace();
                    }
                });
            }
        }
    }
}

至此，websocket基於Redis訂閱發布實現集群化改造完成。