SpringBoot 使用redis发布订阅

- 日理万妓 2021-08-19 22:42 441阅读 0赞

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。  
本文链接：https://blog.csdn.net/qq\_32867467/article/details/82944209  
消息是许多软件架构的重要组成部分。消息传递解决方案提供了高性能，可扩展性，队列持久性和耐用性，故障转移支持等，以及许多更漂亮具备的功能，在Java世界中大多总是使用JMS实现。后来使用Apache ActiveMQ的，有时只是需要简单的排队支持，而Apache ActiveMQ显得过于复杂。事实上，Redis 不仅提供一个NoSQL数据库，同时提供了一套消息系统。

使用Redis作为pub/sub的队列，将消息存放在Redis内存中。

话不多说直接上配置

yml配置

编写监听类来接收订阅的消息

Cat监听

package com.maoxs.listener;
    
    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.stereotype.Component;
    
    /**
     * 监听发送的消息
     */
    @Component
    public class CatListener implements MessageListener {
    
        @Autowired
        RedisTemplate redisTemplate;
    
        @Override
        public void onMessage(Message message, byte[] bytes) {
            System.out.println("我是Cat监听" + message.toString());
        }
    }

Fish监听

package com.maoxs.listener;
    
    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;
    
    /**
     * 监听发送的消息
     */
    public class FishListener implements MessageListener {
    
        @Autowired
        RedisTemplate redisTemplate;
    
        @Override
        public void onMessage(Message message, byte[] bytes) {
            System.out.println("我是Fish监听" + message.toString());
        }
    }

Redis的相关配置

package com.maoxs.redis;
    
    import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.maoxs.listener.CatListener;
    import com.maoxs.listener.FishListener;
    import com.maoxs.pojo.MessageReceiver;
    import org.springframework.cache.CacheManager;
    import org.springframework.cache.annotation.CachingConfigurerSupport;
    import org.springframework.cache.annotation.EnableCaching;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.cache.RedisCacheManager;
    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 java.util.Arrays;
    
    @SuppressWarnings({"all"})
    @Configuration
    @EnableCaching
    public class CacheConfig extends CachingConfigurerSupport {
    
    
        /**
         * redis消息监听器容器
         * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
         * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
         * @param connectionFactory
         * @param listenerAdapter
         * @return
         */
        @Bean
        //相当于xml中的bean
        RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                MessageListenerAdapter CatAdapter, MessageListenerAdapter FishAdapter) {
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            //订阅了一个叫chat 的通道
            container.addMessageListener(CatAdapter, new PatternTopic("cat"));
            container.addMessageListener(FishAdapter, new PatternTopic("fish"));
            //这个container 可以添加多个 messageListener
            return container;
        }
    
        /**
         * 消息监听器适配器，绑定消息处理器
         *
         * @param receiver
         * @return
         */
        @Bean
        MessageListenerAdapter CatAdapter() {
            return new MessageListenerAdapter(new CatListener());
        }
    
        /**
         * 消息监听器适配器，绑定消息处理器
         *
         * @param receiver
         * @return
         */
        @Bean
        MessageListenerAdapter FishAdapter() {
            return new MessageListenerAdapter(new FishListener());
        }
    
        /**
         * redis 读取内容的template
         */
        @Bean
        StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
            return new StringRedisTemplate(connectionFactory);
        }
    
        @Bean
        public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
            StringRedisTemplate template = new StringRedisTemplate(factory);
    
            //定义value的序列化方式
            Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
            ObjectMapper om = new ObjectMapper();
            om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
            jackson2JsonRedisSerializer.setObjectMapper(om);
    
    
            template.setValueSerializer(jackson2JsonRedisSerializer);
            template.setHashKeySerializer(jackson2JsonRedisSerializer);
            template.afterPropertiesSet();
            return template;
        }
    }

创建一个定时器模拟发布消息

package com.maoxs.service;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.scheduling.annotation.EnableScheduling;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    
    @EnableScheduling //开启定时器功能
    @Component
    public class MessageSender {
    
        @Autowired
        @Resource(name = "stringRedisTemplate")
        private StringRedisTemplate stringRedisTemplate;
    
        @Scheduled(fixedRate = 2000) //间隔2s 通过StringRedisTemplate对象向redis消息队列cat频道发布消息
        public void sendCatMessage(){
            stringRedisTemplate.convertAndSend("cat","猫");
        }
    
        @Scheduled(fixedRate = 1000) //间隔1s 通过StringRedisTemplate对象向redis消息队列fish频道发布消息
        public void sendFishMessage(){
            stringRedisTemplate.convertAndSend("fish","鱼");
        }
    
    }

最后就是启动类

package com.maoxs;
    
    import org.mybatis.spring.annotation.MapperScan;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.ComponentScan;
    
    
    @SpringBootApplication
    @ComponentScan("com.maoxs")
    public class DemoApplication {
        private static final Logger LOGGER = LoggerFactory.getLogger(DemoApplication.class);
        public static void main(String[] args) {
            SpringApplication.run(DemoApplication.class, args);
        }
    
    }

然后就启动看下效果

![å¨è¿éæå¥å¾çæè¿°][20181005145742959]

ok 大功告成!!  
————————————————  
版权声明：本文为CSDN博主「MelancholyCat」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。  
原文链接：https://blog.csdn.net/qq\_32867467/article/details/82944209

[20181005145742959]: /images/20210527/ae373a06572444eab32a8544f5bf945a.png