springboot集成Redis队列的发布订阅功能

注册监听：

package com.example.demo.config;
 
import com.example.demo.controller.RedisController;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
 
@Configuration
public class RedisConfig {
 
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter listenerAdapter1,
                                            MessageListenerAdapter listenerAdapter2) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 每一个监听器messageListener都需要一个适配器,队列可以是同一个
        container.addMessageListener(listenerAdapter1, new PatternTopic("redis:test"));
        container.addMessageListener(listenerAdapter2, new PatternTopic("redis:prod"));
        return container;
    }
 
    /**
     * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法
     * @return
     */
    @Bean
    MessageListenerAdapter listenerAdapter1(RedisController redisController) {
        return new MessageListenerAdapter(redisController, "receiveMessage1");
    }
 
    /**
     * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法
     * @return
     */
    @Bean
    MessageListenerAdapter listenerAdapter2(RedisController redisController) {
        return new MessageListenerAdapter(redisController, "receiveMessage2");
    }
 
    @Bean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }
 
}

消息发布：

/**
     * 消息发布者
     * @return
     */
    @RequestMapping("/redisTest")
    public String redisTest() {
        for(int i = 1; i <= 5; i++) {
            redisTemplate.convertAndSend("redis:test","通知你该开始了");
        }
        redisTemplate.convertAndSend("redis:prod","通知你该结束了");
        return "";
    }

消息订阅：

package com.example.demo.controller;
 
import org.springframework.stereotype.Component;
 
@Component
public class RedisController {
 
    public void receiveMessage1(String message) {
        System.out.println(message+"：收到");
    }
 
    public void receiveMessage2(String message) {
        System.out.println(message+"：OK");
    }
 
}

打赏支持