SpringBoot Redis 发布订阅模式(Pub/Sub)的具体使用

    @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);
    }

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

@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;
    }
}

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");
    }
}