springboot 整合使用redis发布订阅功能

Posted 2023-04-06 逆风飞翔的小叔

前言

发布订阅作为一种设计思想在很多开源组件中都有体现，比如大家熟知的消息中间件等，可谓把发布订阅这一思想体现的淋漓尽致了；

一、redis发布订阅简介

Redis发布订阅（pub/sub）是一种消息通信模式：发送者（pub）发送消息，订阅者（sub）接收信息。可以参考下面两张图进行理解。

 

二、几个核心概念解释

1.频道

频道（channel）类似于一个快递柜，快递员往里面放快递，收件人去里面取快递。管道（channel）是由中间件（redis）提供的，一个redisServer中有多个channel。

2、消息发布者

可以理解为消息的生产者，消息发布者通过中间件（redis、mq等）向某个频道（管道）发送消息。

3、消息接收者 

也可以理解为消息消费者，消息接收者通过订阅某个频道（管道）来接收发布者发布的消息。

发布者无需关心是否有人接收消息，发布者只需要把消息发布到某个管道中即可；

三、适用场景

1、核心业务完成后，非核心业务需要记录日志，发邮件，发短信之类的操作，一般来说，通过这种方式，核心业务与非核心业务起到了一个解耦的作用；

2、事件订阅，比如订阅UP主，博主相关的消息等；

3、监听事件，比如在分布式微服务场景下，当应用A的某个数据发生变化时，应用B需要同步更新自己的数据做自身业务操作，对于应用A来说并不关心哪个应用，就可以通过这种方式实现；

四、与springboot的整合

1、导入基础依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

2、配置文件

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://IP:3306/school?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false
    username: root
    password: root
    druid:
      max-active: 100
      initial-size: 10
      max-wait: 60000
      min-idle: 5

  redis:
    host: localhost
    port:  6379

  cache:
    type: redis

3、自定义RedisSubConfig

往容器（RedisMessageListenerContainer）内添加消息监听器，注意，container的参数列表是可以传多个监听器的，但是要定义监听器的bean。在定义监听器的方法体内绑定消息处理器和管道（channel），一个监听器可以监听多个管道，可以通过数组或者添加多个channel的方式定义；

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.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

@Configuration
public class RedisSubConfig 

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory factory, RedisMessageListener listener) 
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        //订阅频道redis.news 和 redis.life  这个container 可以添加多个 messageListener
        container.addMessageListener(listener, new ChannelTopic("redis.user"));
        //container.addMessageListener(listener, new ChannelTopic("redis.news"));
        return container;
    

4、自定义消息监听器

需要实现MessageListener 接口，重写里面的onMessage方法，方法体内需要创建一个MessageListenerAdapter（这是一种规范写法，用于绑定消息处理器和监听器）。

这种写法和很多消息中间件对应的API很相似，即通过一个监听的代码块来完成监听到消息后具体的业务操作；

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.RedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class RedisMessageListener implements MessageListener 

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(Message message, byte[] pattern) 

        // 获取消息
        byte[] messageBody = message.getBody();
        // 使用值序列化器转换
        Object msg = redisTemplate.getValueSerializer().deserialize(messageBody);
        // 获取监听的频道
        byte[] channelByte = message.getChannel();
        // 使用字符串序列化器转换
        Object channel = redisTemplate.getStringSerializer().deserialize(channelByte);
        // 渠道名称转换
        String patternStr = new String(pattern);
        System.out.println(patternStr);
        System.out.println("---频道---: " + channel);
        System.out.println("---消息内容---: " + msg);
    

5、redistemplate的序列化

    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) 
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        //使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper mapper = new ObjectMapper();
        mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(mapper);
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //使用StringRedisSerializer来序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    

6、功能测试

下面写一个测试的接口，模拟业务处理完毕后，向 redis.user 这个通道发送一条消息，看看监听器中是否能够正常接收到消息即可；

@RestController
public class RedisPubController 

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/getUserById")
    public String getUserById(String userId)
        //TODO 执行主业务
        redisTemplate.convertAndSend("redis.user", userId);
        return "hello:" + userId;
    

启动工程后调用接口：

http://localhost:8083/getByUserIdId?userId=1

通过断点可以观察发现，通道中的消息被监听器逻辑监听到了，接下来就可以愉快的处理自己的业务了；

 

本篇通过案例演示了如何基于springboot整合使用redis的发布订阅功能，更多的细节需要结合实际的开发场景进行完善，本篇到此结束，最后感谢观看！

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