Redis发布订阅模式

1、说明

Redis自身提供了发布/订阅(publish/subscribe)模式。实现方式大致流程如下图：

image.png

2、redis发布订阅命令

• 发布者使用命令publish + channel + msg

127.0.0.1:6379[1]> publish channel01 "important things"
(integer) 0
127.0.0.1:6379[1]> publish channel01 "important things"
(integer) 1
127.0.0.1:6379[1]> publish channel01 "hello girl"
(integer) 2

注意：发布返回的是订阅者数量，发布的消息不会持久化，没有订阅者时候，发布消息会丢失，当在发布消息之后对channel进行订阅不会收到之前发布的消息。

• 订阅者使用命令subscribe + channel

127.0.0.1:6379[1]> subscribe channel01
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel01"
3) (integer) 1
1) "message"
2) "channel01"
3) "important things"
1) "message"
2) "channel01"
3) "hello somebody"

使用subscribe客户端进入订阅状态，该状态下不能使用与“发布/订阅”无关的其他命令。

3、RedisTemplate 实现发布订阅

• 发布者，使用convertAndSend(channel , message)方法实现消息发布

@RequestMapping("/pubSub")
@RestController
public class RedisComprehensive {

    private static final String CHANNEL = "channel_01";

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping(value = "/publishMessage")
    public void publishMessage(String message){
        // 发布者
        redisTemplate.convertAndSend(CHANNEL,message);
    }
}

• 订阅者分析

（1）配置redis定义消息容器RedisMessageListenerContainer。
addMessageListener(MessageListenerAdapter，PatternTopic): 新增订阅频道及订阅者，订阅者必须有相关方法处理收到的消息。
setTopicSerializer(RedisSerializer) :对频道内容进行序列化解析

（2）配置适配器MessageListenerAdapter
MessageListenerAdapter(Object delegate, String defaultListenerMethod)
delegate消息订阅者类，defaultListenerMethod消息处理方法

【实现代码如下】：

@Configuration
public class RedisConfig {

    private static final String CHANNEL = "channel_01";
     /**
     * @name: container
     * @description: Redis订阅消息监听容器
     * @param connectionFactory
     * @param adapter
     * @return: org.springframework.data.redis.listener.RedisMessageListenerContainer
    */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter adapter){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 可以添加多个MessageListener
        container.addMessageListener(adapter,new PatternTopic(CHANNEL));
        return container;
    }

    /**
     * @name: listenerAdapter
     * @description: 配置消息处理适配器
     * @param msgReceiveListener
     * @return: org.springframework.data.redis.listener.adapter.MessageListenerAdapter
     *
    */
    @Bean
    public MessageListenerAdapter listenerAdapter(MsgReceiveListener msgReceiveListener){
         // messageListenerAdapter 传入一个消息接受的处理器，利用反射的方式调用对应的处理方法
        return new MessageListenerAdapter(msgReceiveListener,"onMessage");
    }
}

import lombok.extern.slf4j.Slf4j;
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.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class MsgReceiveListener implements MessageListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
        String body = serializer.deserialize(message.getBody());
        String chanel = serializer.deserialize(message.getChannel());
        log.info("接收的消息：{},使用的chanel：{}",body,chanel);
    }
}

执行结果：

2021-06-24 15:06:18.436  INFO 28520 --- [nio-8081-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 13 ms
2021-06-24 15:06:18.540  INFO 28520 --- [    container-2] c.l.r.listener.MsgReceiveListener        : 接收的消息：这是一个发布消息,使用的chanel：channel_01