【2023】Redis实现消息队列的方式汇总以及代码实现

ゞ浴缸里的玫瑰 2023-10-14 11:12 70阅读 0赞

#### Redis实现消息队列的方式汇总以及代码实现 ####

*  前言
 *  开始前准备
 *   *   *  1、添加依赖
         *  2、添加配置的Bean
 *  具体实现
 *   *  一、从最简单的开始：List 队列
     *   *   *  代码实现
     *  二、发布订阅模式：Pub/Sub
     *   *  1、使用RedisMessageListenerContainer实现订阅
         *  2、还可以使用redisTemplate实现订阅
     *  三、、 趋于成熟的队列：Stream
     *   *  具体java代码实现：
 *  总结

## 前言 ##

经常听到很多人讨论，关于「把 Redis 当作队列来用是否合适」的问题。

有些人表示赞成，他们认为 Redis 很轻量，用作队列很方便。也些人则反对，认为 Redis 会「丢」数据，最好还是用「专业」的队列中间件更稳妥。

这篇文章就聊一聊把 Redis 当作队列，究竟是否合适这个问题。我们会从简单到复杂，一步步带你梳理其中的细节，把常用的实现方式展现一遍。

## 开始前准备 ##

#### 1、添加依赖 ####

<dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
            </dependency>
                    <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
            </dependency>
                    <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
                    <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
                <version>5.4.0</version>
            </dependency>

#### 2、添加配置的Bean ####

避免不方便用软件查看存储的数据

/**
         * redisTemplate 序列化使用的jdkSerializeable, 存储二进制字节码, 所以自定义序列化类
         * @param redisConnectionFactory
         * @return
         */
        @Bean
        public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        
            RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
            redisTemplate.setConnectionFactory(redisConnectionFactory);
    
            // 使用Jackson2JsonRedisSerialize 替换默认序列化
            Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
    
            ObjectMapper objectMapper = new ObjectMapper();
            objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
    
            jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
    
            // 设置value的序列化规则和 key的序列化规则
            redisTemplate.setKeySerializer(new StringRedisSerializer());
            //jackson2JsonRedisSerializer就是JSON序列号规则，
            redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
            redisTemplate.afterPropertiesSet();
            return redisTemplate;
        }

## 具体实现 ##

### 一、从最简单的开始：List 队列 ###

首先，我们先从最简单的场景开始讲起。

如果你的业务需求足够简单，想把 Redis 当作队列来使用，肯定最先想到的就是使用 List 这个数据类型。

因为 List 底层的实现就是一个「链表」，在头部和尾部操作元素，时间复杂度都是 O(1)，这意味着它非常符合消息队列的模型。

如果把 List 当作队列，你可以这么来用。

##### 代码实现 #####

生产者端读取：

@RestController
    @RequestMapping("/redis01")
    public class RedisTest1 {
        
    
        @Autowired
        private RedisTemplate<String, Object> redisTemplate;
    
    
    	//LPUSH 发布消息
        @GetMapping("/set")
        public void set(String code){
        
            redisTemplate.opsForList().leftPush("code",code);
        }
        // RPOP 拉取消息
        @GetMapping("/get1")
        public String get1(String key){
        
            Object code =  redisTemplate.opsForList().rightPop(key);
            if (code!=null){
        
                return code.toString();
            }
            return "redis中没数据！";
        }

> **实现模型：**  
> 这个模型也非常简单容易理解。 ![在这里插入图片描述][c3a6aef6424549cb80f70ba849301c77.png]

但这里有个小问题，当队列中已经没有消息了，消费者在执行 RPOP 时，会返回 NULL。  
![在这里插入图片描述][3b43ac5ce4514844824b8e2c8ec5656f.png]  
一般在编写消费者时，会采用一个死循环，这个实现方式就是不断去队列中拉取数据。

@GetMapping("/get2")
        public String get2(String key) throws InterruptedException {
        
            while (true){
        
                Object code = redisTemplate.opsForList().rightPop(key);
                System.out.println(code);
     //            读取到消息，退出，没读到继续循环
                if (code!=null){
        
                    return code.toString();
                }
            }
        }

如果此时队列为空，那消费者依旧会频繁拉取消息，这会造成「CPU 空转」，不仅浪费 CPU 资源，还会对 Redis 造成压力。

怎么解决这个问题呢？

也很简单，**当队列为空时，我们可以「休眠」一会**，再去尝试拉取消息。代码可以修改成这样：

这就解决了 CPU 空转问题。

这个问题虽然解决了，但又带来另外一个问题：当消费者在休眠等待时，有新消息来了，那消费者处理新消息就会存在「延迟」。

假设设置的休眠时间是 2s，那新消息最多存在 2s 的延迟。

要想缩短这个延迟，只能减小休眠的时间。但休眠时间越小，又有可能引发 CPU 空转问题。

鱼和熊掌不可兼得。

那如何做，既能及时处理新消息，还能避免 CPU 空转呢？

Redis 是否存在这样一种机制：如果队列为空，消费者在拉取消息时就「阻塞等待」，一旦有新消息过来，就通知我的消费者立即处理新消息呢？

幸运的是，**Redis 确实提供了「阻塞式」拉取消息的命令：BRPOP / BLPOP**，这里的 B 指的是阻塞（Block）。  
![在这里插入图片描述][d5b41978010a47c18f9fd2a17ec77eed.png]  
在java中也已经封装好了，调用pop方法时，直接设置一个过期时间就行

@GetMapping("/get3")
        public String get3(String key) throws InterruptedException {
        
            Object code = redisTemplate.opsForList().rightPop(key,0, TimeUnit.SECONDS);
            if (code==null){
        
                return "数据读取超时！";
            }
            return code.toString();
        }

使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个「超时时间」，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL。

这个方案不错，既兼顾了效率，还避免了 CPU 空转问题，一举两得。

> 注意：如果设置的超时时间太长，这个连接太久没有活跃过，可能会被 Redis Server 判定为无效连接，之后 Redis Server  
> 会强制把这个客户端踢下线。所以，采用这种方案，客户端要有重连机制。

解决了消息处理不及时的问题，你可以再思考一下，这种队列模型，有什么缺点？

我们一起来分析一下：

1.  **不支持重复消费**：消费者拉取消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费，即不支持多个消费者消费同一批数据
2.  **消息丢失**：消费者拉取到消息后，如果发生异常宕机，那这条消息就丢失了  
    第一个问题是功能上的，使用 List 做消息队列，它仅仅支持最简单的，一组生产者对应一组消费者，不能满足多组生产者和消费者的业务场景。

第二个问题就比较棘手了，因为从 List 中 POP 一条消息出来后，这条消息就会立即从链表中删除了。也就是说，无论消费者是否处理成功，这条消息都没办法再次消费了。

这也意味着，如果消费者在处理消息时异常宕机，那这条消息就相当于丢失了。

针对这 2 个问题怎么解决呢？我们一个个来看。

### 二、发布订阅模式：Pub/Sub ###

从名字就能看出来，这个模块是 Redis 专门是针对「发布/订阅」这种队列模型设计的。

它正好可以解决前面提到的第一个问题：重复消费。

即多组生产者、消费者的场景，我们来看它是如何做的。

Redis 提供了 PUBLISH / SUBSCRIBE 命令，来完成发布、订阅的操作。  
![在这里插入图片描述][dc51ccf2ee334154b6e22d33232cd28c.png]  
依赖继续用前面的就行

#### 1、使用RedisMessageListenerContainer实现订阅 ####

>  *  通过实现MessageListener接口来处理接收到的消息。这允许您在Spring应用程序中以更高级的方式处理消息，例如使用依赖注入和其他Spring功能。它还支持基于注解的消息监听器，使消息处理更加简洁和灵活。
>  *  该方式是Spring Data Redis库提供的方法，用于在Spring应用程序中使用Redis的发布订阅功能。它需要创建一个MessageListenerContainer对象，并通过调用addMessageListener方法来添加消息监听器。

*  添加监控器，用于监听通道  
    为了便于更加直观的对比测试，我添加了两个

/**
     * @author zhengfuping
     * @version 1.0
     * @description: TODO  配置监控器
     * @date 2023/7/28 17:10
     */
    @Component
    public class RedisMessaeListener1 implements MessageListener {
        
        @Override
        public void onMessage(Message message, byte[] pattern) {
        
    
            String channel = new String(message.getChannel());
            String body = new String(message.getBody());
            System.out.println("监听器1号：消息: " + body + " 通道QQ: " + channel);
        }
    }
    
    /*########################*/
    
    /**
     * @author zhengfuping
     * @version 1.0
     * @description: TODO 配置监控器1
     * @date 2023/8/2 11:24
     */
    @Component
    public class RedisMessaeListener2 implements MessageListener {
        
        @Override
        public void onMessage(Message message, byte[] pattern) {
        
            String channel = new String(message.getChannel());
            String body = new String(message.getBody());
            System.out.println("监听器2号：消息: " + body + " 通道QQ: " + channel);
        }
    }

*  配置订阅可以单和多个  
    用于绑定主题（通道）和监听器，该发送是

/**
     * @author zhengfuping
     * @version 1.0
     * @description: TODO 使用RedisMessageListenerContainer直接注入到bean进行监听
     * @date 2023/7/28 15:43
     */
    @Configuration
    public class RedisPubSubExample {
        
    
        @Autowired
        private  RedisMessaeListener1 redisMessaeListener1;
        @Autowired
        private  RedisMessaeListener2 redisMessaeListener2;
    
        /**
         * 订阅三个频道
         * @author zhengfuping
         * @date 2023/8/2 11:19
         * @param redisConnectionFactory redis线程工厂
         * @return RedisMessageListenerContainer
         */
    //    @Bean
        public RedisMessageListenerContainer subscribeToChannel(RedisConnectionFactory redisConnectionFactory){
        
            RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
    
            listenerContainer.setConnectionFactory(redisConnectionFactory);
            List<Topic> list = new ArrayList<>();
            list.add(new PatternTopic("TEST01"));
            list.add(new PatternTopic("TEST02"));
            list.add(new PatternTopic("TEST03"));
            /*
             *  redisMessaeListener  消息监听器
             *  list  订阅的主题（可以单个和多个）
             */
            listenerContainer.addMessageListener(redisMessaeListener1,list);
            listenerContainer.addMessageListener(redisMessaeListener2,new PatternTopic("TEST01"));
            return listenerContainer;
        }
    }

*  向指定频道发送消息

/**
         *  PUBLISH 发送消息到指定频道
         * @author zhengfuping
         * @date 2023/8/2 11:14
         * @param channel 通道
         * @param name  数据
         * @param age
         * @return Object
         */
        @GetMapping("/pub")
        public Object pub(String channel,String name,Integer age) {
        
            User user = new User(name, age);
            redisTemplate.convertAndSend(channel,user);
            return user;
        }

![在这里插入图片描述][834146cd0518481786af60676ef09899.png]

#### 2、还可以使用redisTemplate实现订阅 ####

>  *  该方式是Redis客户端的方法，用于在独立的Redis客户端中直接使用发布订阅功能。它需要创建一个Redis连接对象，并通过调用subscribe方法来订阅一个或多个频道。
>  *  如果您只是在独立的Redis客户端中使用发布订阅功能，并且不需要使用Spring的其他功能，则可以选择connection.subscribe

/**
     * 自行添加订阅
     */
        @GetMapping("/sub")
        public void sub(String channel) {
        
            RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
    
            /*
             *  MessageListener:监听器，直接使用内部类实现绑定监听可以把数据传递出去
             *  channel 订阅频道
             */
            connection.subscribe((message, pattern) -> {
        
                String channel1 = new String(message.getChannel());
                String body = new String(message.getBody());
                System.out.println("subscribe方式监听：消息: " + body + " 通道QQ: " + channel1);
            }, channel.getBytes());
    
    //            connection.close();
        }

![在这里插入图片描述][e71924468d324164b2b80c59fc0f4cca.png]

发送消息

@GetMapping("/pub")
        public Object pub(String channel,String name,Integer age) {
        
            User user = new User(name, age);
            redisTemplate.convertAndSend(channel,user);
            return user;
        }

![在这里插入图片描述][edfe9e41cdb24ebba0703eac5841c686.png]

最终监听到的结果  
![在这里插入图片描述][e4bcbd0aea4a4bfdaee0c791ca527c81.png]

### 三、、 趋于成熟的队列：Stream ###

我们来看 Stream 是如何解决上面这些问题的。

我们依旧从简单到复杂，依次来看 Stream 在做消息队列时，是如何处理的？

首先，Stream 通过 XADD 和 XREAD 完成最简单的生产、消费模型：

*  生产者发布 2 条消息：

// *表示让Redis自动生成消息ID
    127.0.0.1:6379> XADD queue * name zhangsan
    "1618469123380-0"
    127.0.0.1:6379> XADD queue * name lisi
    "1618469127777-0"

*  消费者拉取消息：

// 从开头读取5条消息，0-0表示从开头读取
    127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 0-0
    1) 1) "queue"
       2) 1) 1) "1618469123380-0"
             2) 1) "name"
                2) "zhangsan"
          2) 1) "1618469127777-0"
             2) 1) "name"
                2) "lisi"

流程图  
![在这里插入图片描述][5b6180092342486fa31c3d489af43f2e.png]

#### 具体java代码实现： ####

*  先配置监听消息类

@Slf4j
    @Component
    public class ListenerMessage implements StreamListener<String, MapRecord<String, String, String>> {
        
    
        @Override
        public void onMessage(MapRecord<String, String, String> entries) {
        
            log.info("接受到来自redis的消息");
            System.out.println("message id "+entries.getId());
            System.out.println("stream "+entries.getStream());
            System.out.println("body "+entries.getValue());
        }
    }

*  添加工具类，实现初始化

@Component
    @Slf4j
    public class RedisStreamUtil {
        
        @Autowired
        private RedisTemplate<String,Object> redisTemplate;
    
    
        /**
         * @author zhengfuping 添加数据
         * @param streamKey
         * @param map
         * @return RecordId
         */
        public RecordId addStream(String streamKey,Map<String, Object> map){
        
            RecordId recordId = redisTemplate.opsForStream().add(streamKey, map);
            return recordId;
        }
    
        /**
         * 用来创建绑定流和组
         */
        public void addGroup(String key, String groupName){
        
            redisTemplate.opsForStream().createGroup(key,groupName);
        }
    
        /**
         * 用来判断key是否存在
         */
        public boolean hasKey(String key){
        
            if(key==null){
        
                return false;
            }else{
        
                return redisTemplate.hasKey(key);
            }
    
        }
        /**
         * 用来删除掉消费了的消息
         */
        public void delField(String key,String recordIds){
        
            redisTemplate.opsForStream().delete(key,recordIds);
        }
    
        /**
         * 用来初始化 实现绑定
         */
        public void initStream(String key, String group){
        
            //判断key是否存在，如果不存在则创建
            boolean hasKey = hasKey(key);
            if(!hasKey){
        
                Map<String,Object> map = new HashMap<>();
                map.put("field","value");
                RecordId recordId = addStream(key, map);
                addGroup(key,group);   //把Stream和gropu绑定
                delField(key,recordId.getValue());
                log.info("stream:{}-group:{} initialize success",key,group);
            }
        }
    }

*  添加配置类，配置Stream

/**
     * @author zhengfuping
     * @version 1.0
     * @description: TODO  添加配置类，配置Stream
     */
    @Configuration
    @Slf4j
    public class RedisStreamConfig {
        
    
        @Autowired
        private RedisStreamUtil redisStream;
        @Autowired
        private ListenerMessage listenerMessage;
    
    
        @Bean
        public Subscription subscription(RedisConnectionFactory factory){
        
    //        代码中的var是使用了Lombok的可变局部变量。主要是为了方便
    //        StreamMessageListenerContainer: 消息侦听容器，不能在外部实现。创建后，StreamMessageListenerContainer可以订阅Redis流并使用传入的消息
            var options = StreamMessageListenerContainer
                    .StreamMessageListenerContainerOptions
                    .builder()
                    .pollTimeout(Duration.ofSeconds(1))
                    .build();
            redisStream.initStream("mystream","mygroup");	//调用初始化
            var listenerContainer = StreamMessageListenerContainer.create(factory,options);
    
            /*
             *  注意这里接受到消息后会被自动的确认，如果不想自动确认请使用其他的创建订阅方式
             * 消费组 consumer group ，它不能为null （Consumer类型）
             * stream offset ，stream的偏移量（StreamOffset 类型）
             * listener 不能为null （StreamListener<K,V> 类型）
             */
            var subscription = listenerContainer.receiveAutoAck(Consumer.from("mygroup","huhailong"),
                    StreamOffset.create("mystream", ReadOffset.lastConsumed()),listenerMessage);
    
            listenerContainer.start();
            return subscription;
        }
    }

*  调用测试

/**
     * @author zhengfuping
     * @version 1.0
     * @description: TODO
     * @date 2023/8/2 16:06
     */
    @RestController
    @RequestMapping("/redisStream")
    public class RedisStreamTest {
        
    
        @Autowired
        private RedisStreamUtil redisStream;
    
        @GetMapping("add")
        public void add(String key,String data){
        
            Map<String, Object> map = new HashMap<>();
            map.put(key,data);
    //        添加数据到mystream流中
            RecordId recordId = redisStream.addStream("mystream", map);
    //        删除流中消费了的指定key的数据
            redisStream.delField("mystream",recordId.getValue());
        }
    }

> **Stream的好处在于可以写入到 RDB 和 AOF 做持久化。**  
> Stream是新增加的数据类型，它与其它数据类型一样，每个写操作，也都会写入到 RDB 和 AOF 中。  
> 我们只需要配置好持久化策略，这样的话，就算 Redis 宕机重启，Stream 中的数据也可以从 RDB 或 AOF 中恢复回来。

## 总结 ##

好了，总结一下。这篇文章我们从「Redis 能否用作队列」这个角度出发，介绍了 List、Pub/Sub、Stream 在做队列的使用方式，以及它们各自的优劣。

之后又把 Redis 和专业的消息队列中间件做对比，发现 Redis 的不足之处。

最后，我们得出 Redis 做队列的合适场景。

![在这里插入图片描述][72416f0cfdfb459caef675cb2928124a.png]

[c3a6aef6424549cb80f70ba849301c77.png]: https://img-blog.csdnimg.cn/c3a6aef6424549cb80f70ba849301c77.png
[3b43ac5ce4514844824b8e2c8ec5656f.png]: https://img-blog.csdnimg.cn/3b43ac5ce4514844824b8e2c8ec5656f.png
[d5b41978010a47c18f9fd2a17ec77eed.png]: https://img-blog.csdnimg.cn/d5b41978010a47c18f9fd2a17ec77eed.png
[dc51ccf2ee334154b6e22d33232cd28c.png]: https://img-blog.csdnimg.cn/dc51ccf2ee334154b6e22d33232cd28c.png
[834146cd0518481786af60676ef09899.png]: https://img-blog.csdnimg.cn/834146cd0518481786af60676ef09899.png
[e71924468d324164b2b80c59fc0f4cca.png]: https://img-blog.csdnimg.cn/e71924468d324164b2b80c59fc0f4cca.png
[edfe9e41cdb24ebba0703eac5841c686.png]: https://img-blog.csdnimg.cn/edfe9e41cdb24ebba0703eac5841c686.png
[e4bcbd0aea4a4bfdaee0c791ca527c81.png]: https://img-blog.csdnimg.cn/e4bcbd0aea4a4bfdaee0c791ca527c81.png
[5b6180092342486fa31c3d489af43f2e.png]: https://img-blog.csdnimg.cn/5b6180092342486fa31c3d489af43f2e.png
[72416f0cfdfb459caef675cb2928124a.png]: https://img-blog.csdnimg.cn/72416f0cfdfb459caef675cb2928124a.png