2020-04-05-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订阅发布实现集群化改造完成。

原文地址：https://www.cnblogs.com/cnsyear/p/12638121.html

时间： 2024-10-16 12:32:21

2020-04-05-SpringBoot+WebSocket基于Redis订阅发布实现集群化的相关文章

基于redis的spring task集群配置

项目从单节点环境变为集群环境,这个时候要确保项目中的定时任务同一时刻只能在集群中的其中一个服务器中运行,但又不能写死在哪一台服务器上运行,怎样才能实现这样的需求? 思路: 可以做一个切面,扫描定时任务,在任务开始之前使用redis缓存判断任务是否启动,由于Redis是单线程按顺序执行命令,可以在每个定时任务执行前,使用setnx方法判断是否可以添加值,如果添加成功,说明这个定时任务没有被执行,设置key的过期时间并让定时任务执行,如果不可以添加值,说明该定时任务已经在其他服务器上跑了,方法之间返

基于redis(订阅发布)实现python和java进程间通信

主要结构为: python进程发布消息,java进程订阅消息. 依赖环境: python:   pip install redis java:  jedis 1. python端: PubSub.py import redis class PubSub(object): def __init__(self, host, port, db): self.__conn = redis.Redis(host, port, db) def publish(self, channel, msg): sel

Redis订阅发布功能

Redis安装使用 1.下载Redis:http://redis.io/download 2.解压安装: #tar zxf redis-3.0.6.tar.gz #make 3.启动 #cd /usr/local/src/redis-3.0.6 #src/redis-server & 4.检查 5.登陆 #src/redis-cli 6.常用命令  set name 'yangmv'        存数据            get name                        取数

ServiceStack.Redis订阅发布服务的调用（Z）

1.Redis订阅发布介绍Redis订阅发布是一种消息通信模式:发布者(publisher)发送消息,订阅者(Subscriber)接受消息.类似于设计模式中的观察者模式.发布者和订阅者之间使用频道进行通信,当需要发送消息时,发布者通过publish命令将消息发送到频道上,该消息就会发送给订阅这个频道的订阅者. 图片来自于http://www.runoob.com/redis/redis-pub-sub.html 2.ServiceStack.Redis ServiceStack.Redis是R

责任链模式的具体应用 ServiceStack.Redis订阅发布服务的调用

1.业务场景 生产车间中使用的条码扫描,往往一把扫描枪需要扫描不同的条码来处理不同的业务逻辑,比如,扫描投入料工位条码.扫描投入料条码.扫描产出工装条码等,每种类型的条码位数是不一样,因此通过条码长度来进行业务区分. 2.初步设计 面对此场景,能够想到的最简单的设计就是使用if...else if...或者swith进行判断,因此,我们编写的代码如下 1 switch(barCode.length) 2 { 3 case 3: 4 DoSomething1(); 5 break; 6 case

PHP + Redis 订阅/发布 实现即时通讯功能

最近比较忙,没时间更新博客,先暂时记下,待后续补充 Redis Sub/Pub 订阅/发布 待编辑 PHP + Redis 订阅/发布 实现即时通讯 待编辑 原文地址:http://blog.51cto.com/12750968/2105579

redis解决方案之三种集群模式的概念与部署

上篇文章为大家总结了redis命令并讲述了持久化,今天我们来看一下redis的三种集群模式:主从复制,哨兵集群,Cluster集群 本篇文章先介绍redis-cluster,然后再依次介绍它的哨兵集群与主从复制 一.Cluster集群模式概念 redis集群是一个分布式与容错的redis实现.在集群中不存在代理节点与中心节点.后期可以很好的将其进行扩展 此模式也解决了redis高可用与可扩展的问题.但是redis集群不支持需要同时处理多个Key的redis命令 因为执行这些命令需要在多个redi

基于VMware的虚拟Linux集群搭建-lvs+keepalived

通过keepalived实现lvs服务器的的双机热备和真实服务器之间的负载均衡的blog挺多的,在搭建的时候也参考了相关博文,不同人搭建的实验环境不一样,本文是基于VM搭建的虚拟集群环境,全部虚拟机网卡模式为NAT方式相连,处于同一网段. 搭建环境: 使用redhead 2.6.32-431.el6.x86_64版本的linux,创建四台,配置相关网络信息,确保同一网段下. 配置好各自ip即可,虚拟ip在配置lvs时设定. 安装配置LVS 1.在主备Lvs上安装ipvsadm和keepalive

（转）redis 3.0的集群部署

一:关于redis cluster 1:redis cluster的现状 reids-cluster计划在redis3.0中推出,可以看作者antirez的声明:http://antirez.com/news/49 (ps:跳票了好久,今年貌似加快速度了),目前的最新版本是redis3 beta2(2.9.51). 作者的目标:Redis Cluster will support up to ~1000 nodes. 赞... 目前redis支持的cluster特性(已亲测): 1):节点自动发