SpringBoot Redis 發布訂閱模式 Pub/Sub

2021 年 12 月 18 日
筆記
Redis, spring

SpringBoot Redis 發布訂閱模式 Pub/Sub

注意：redis的發布訂閱模式不可以將消息進行持久化，訂閱者發生網路斷開、宕機等可能導致錯過消息。

Redis命令行下使用發布訂閱

publish 發布

發布者通過以下命令可以往指定channel發布message

redis> publish channel message

訂閱者通過以下命令可以訂閱一個或多個頻道，如果頻道不存在則會創建

redis> subscribe channel [channel ...]

對於redis的發布訂閱的命令就這麼簡單。那麼接下來我們在springboot中如何使用發布訂閱的功能呢？

SpringBoot中使用Redis的發布訂閱功能

添加依賴配置redis資訊和連接池什麼的就不說了，如果添加的有commons-pool2依賴的話，會自動幫我們配置redis連接池的

發布者

相對於訂閱者來說，發布者的實現方式很簡單，以下方式就可以往channel中發送message了。


    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    public void publish(){
    // 使用高級的redisTemplate
    redisTemplate.convertAndSend("channel","message");
    
    // 使用低級的connection 實際上redisTemplate的底層就是使用的下面的方式
    redisTemplate.execute(new RedisCallback<Object>() {
          @Override
          public Object doInRedis(RedisConnection connection) throws DataAccessException {
              connection.publish("channel".getBytes(StandardCharsets.UTF_8), "message".getBytes(StandardCharsets.UTF_8));
              return null;
         }
     }, true);
     // true這個參數意思是 是否將redis連接暴露給回調程式碼，大多數情況下設置true就可以了，往後深入的話可以看到
     RedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse)); 如果為false的話會創建redis連接的代理
}

訂閱者

訂閱者因為涉及到連接、執行緒等所以內容相對會多一點

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    public void subscribe() {
        redisTemplate.execute(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                // 我定義了一個全局的 ConcurrentHashMap 用來存放連接 因為後面的取消訂閱的執行緒要和訂閱的執行緒用同一個連接
                map.put("connection",connection);

                // subscribe 按頻道訂閱 該方法會阻塞該執行緒 只有取消訂閱才會釋放該執行緒
                connection.subscribe(new MessageListener() {
                    @Override
                    public void onMessage(Message message, byte[] pattern) {
                        log.info("接收到消息");
                        System.out.println(new String(message.getBody()));
                    }
                }, "channelOne".getBytes(StandardCharsets.UTF_8), "channelTwo".getBytes(StandardCharsets.UTF_8));

                // 按模式訂閱 pSubscribe 只有取消訂閱才會釋放該執行緒
//                connection.pSubscribe(new MessageListener() {
//                    @Override
//                    public void onMessage(Message message, byte[] pattern) {
//                        System.out.println(new String(message.getBody()));
//                    }
//                }, "patternOne".getBytes(StandardCharsets.UTF_8), "patternOne".getBytes(StandardCharsets.UTF_8));
                return null;
            }
        }, true);
    }

如何取消訂閱呢？從剛才的map里取到連接

    RedisConnection the = map.get("connection");
    Subscription subscription = the.getSubscription();
    subscription.unsubscribe();

消息監聽容器

上面的那種訂閱為低級訂閱，由於連接在調用subscribe的時候會導致當前執行緒阻塞，這種方式需要對每個監聽器連接和執行緒管理，所以spring提供了RedisMessageListenerContainer類來幫我們完成這些工作。

RedisMessageListenerContainer顧名思義可以知道它是一個消息監聽容器
詳情請參考官方文檔

如何實現

@Configuration
public class DefaultMessageListenerContainerConfig {

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory factory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        // 官方推薦我們使用自定義的執行緒池或者使用TaskExecutor
        container.setTaskExecutor(executor());
        container.addMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] pattern) {
                System.out.println(Thread.currentThread().getName() + ": " + new String(message.getBody()));
            }
        }, new ChannelTopic("message"));
        return container;
    }

    @Bean
    public TaskExecutor executor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
        executor.setQueueCapacity(100);
        executor.initialize();
        return executor;
    }
}

這個時候我們在redis命令行內使用 publish channel message 的時候，我們的spring程式就可以訂閱到消息了。

再說下 MessageListenerAdapter
我們可以通過 MessageListenerAdapter 消息接收者包裝進去，消息接收者不會和redis有任何耦合。
官方文檔給了spring傳統的xml的方式配置的，下面我給出基於configuration配置的程式碼

public interface MessageDelegate {
    void handleMessage(String message);
}

public class DefaultMessageDelegate implements MessageDelegate {
    @Override
    public void handleMessage(String message) {
        System.out.println(message);
    }
}

@Configuration
public class MessageListenerContainerConfig {

    @Autowired
    private DefaultMessageDelegate defaultMessageDelegate;

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory factory,
                                                   MessageListenerAdapter messageListenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.setTaskExecutor(executor());

        Map<MessageListenerAdapter, Collection<? extends Topic>> map = new HashMap<>();
        List<ChannelTopic> channelTopics = new ArrayList<>();
        ChannelTopic channelTopic = new ChannelTopic("message");
        channelTopics.add(channelTopic);
        map.put(messageListenerAdapter, channelTopics);
        container.setMessageListeners(map);

        return container;
    }

    @Bean
    public TaskExecutor executor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
        executor.setQueueCapacity(100);
        executor.initialize();
        return executor;
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter() {
        // handleMessage 參數消息來的時候要調用的方法 默認是 handleMessage
        return new MessageListenerAdapter(defaultMessageDelegate, "handleMessage");
    }
}

如果我們要在程式運行時添加訂閱或者取消訂閱的時候該怎麼辦呢？
我們需要提前準備好消息偵聽器，添加的時候把偵聽器注入到消息容器
取消的時候就調用消息容器的remove方法把偵聽器刪除掉即可。

Tags: Redis spring