SpringBoot整合Redis实践消息发布订阅特性

介绍

发布订阅是消息的一种常用模式。例如，在企业分配任务之后，可以通过邮件、短信或者微信通知到相关的责任人，这就是一种典型的发布订阅模式。首先是 Redis 提供一个渠道，让消息能够发送到这个渠道上 ，而多个系统可以监听这个渠道，如短信、微信和邮件系统都可以监听这个渠道，当一条消息发送到渠道，渠道就会通知它的监听者，这样短信、微信和邮件系统就能够得到这个渠道给它们的消息了，这些监听者会根据自己的需要去处理这个消息，于是我们就可以得到各种各样的通知了。

实践

创建一个消息的监听者（通道的订阅者）

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

@Component
public class RedisMessageSubscriber implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        System.out.println("[监听到redis消息]");
        // 消息体
        String body = new String(message.getBody());
        // 渠道名称
        String topic = new String(pattern);
        System.out.println(body);
        System.out.println(topic);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

这里的 onMessage 方法是得到消息后的处理方法，其中 message 参数代表 Redis 发送过来的消息，pattern 是渠道名称，onMessage 方法里打印了它们的内容。这里因为标注了@Component 注解，所以在 Spring Boot 扫描后，会把它自动装配到 IoC 容器中。

创建RedisMessageListenerContainer对象，这个主要是用来注册发布订阅关联关系的。

先创建一个共用的主题名称，类似于队列

public class RedisMessageConstant {
    public final static String TOPIC = "zy-topic";
}
1
2
3

/**
 *  redis订阅注册
 * @author zhangyu
 * @date 2022/10/27
 **/
@Configuration
public class RedisMessageListenerHolder {

    @Autowired
    private RedisTemplate redisTemplate = null;

    // Redis 连接工厂
    @Autowired
    private RedisConnectionFactory connectionFactory = null;

    // Redis 消息监听器
    @Autowired
    private MessageListener redisMsgListener = null;

    // 任务池
    private ThreadPoolTaskScheduler taskScheduler = null;

    /**
     * 创建任务池，运行线程等待处理 Redis 的消息
     */
    @Bean
    public ThreadPoolTaskScheduler initTaskScheduler() {
        if (taskScheduler != null) {
            return taskScheduler;
        }
        taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(20);
        return taskScheduler;
    }

    /**
     * 定义 Redis 的监听容器
     *
     * @return 监听容器
     */
    @Bean
    public RedisMessageListenerContainer initRedisContainer() {
        RedisMessageListenerContainer container
                = new RedisMessageListenerContainer();
        // Redis 连接工厂
        container.setConnectionFactory(connectionFactory);
        // 设置运行任务池
        container.setTaskExecutor(initTaskScheduler());
        // 定义监听渠道，名称为 topic1
        Topic topic = new ChannelTopic(RedisMessageConstant.TOPIC);
        // 使用监听器监听 Redis 的消息
        container.addMessageListener(redisMsgListener, topic);
        return container;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

这里 RedisTemplate 和 RedisConnectionFactory 对象都是 Spring Boot 自动创建的，所以这里只是把它们注入进来，只需要使用@Autowired 注解即可。然后定义了一个任务池，并设置了任务池大小为 20，这样它将可以运行线程，并进行阻塞，等待 Redis 消息的传入。接着再定义了一个 Redis 消息监听的容器 RedisMessageListenerContainer，并且往容器设置了 Redis 连接工厂和指定运行消息的线程池，以及对应的监听的Topic通道。

创建消息的发送者

【1】在 Redis 的客户端输入命令：publish topic1 msg
【2】Spring 中，我们可以使用 RedisTemplate 来发送消息

/**
 * 消息的发布者
 *
 * @author zhangyu
 * @date 2022/10/27
 **/
@RestController
@RequestMapping("/rmp")
public class RedisMessagePublisher {


    @Autowired
    private RedisTemplate redisTemplate;


    @GetMapping("/publish")
    public ResponseEntity publish(String msg) {
        redisTemplate.convertAndSend(RedisMessageConstant.TOPIC, msg);
        return ResponseEntity.ok("success");

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

测试接口发布消息：
http://localhost:8080/rmp/publish?msg=hello

小结

【1】针对一些特定的场景，我们可以灵活的应用 Redis 发布订阅特性帮助我们实现需求。
【2】这种方式是一种异步的方式处理，对于类似短信通知，邮件通知，消息通知都是可以IDE。
【3】但是注意Redis 目前发布订阅基于 发送即忘 策略，且没有 ACK 机制，意味着客户端重启掉线期间，消息会丢失。加上 Pub/Sub 消息 没有持久化机制，假如当订阅客户端由于网络原因没收到，想再次重试，这是没法实现的。