Reids实现消息队列的思路与方式

忘是亡心i 2023-07-14 10:41 139阅读 0赞

**目录**

前言

1 队列的特点

2 使用List实现简单队列

2.1 思路

2.2 实现简单队列的相关命令

2.3 RedisTemplate操作List实现消息队列

3 使用SortedSet实现延时队列

3.1 延时队列应用场景

3.2 思路

3.3 实现延时队列相关命令

3.4 RedisTemplate操作SortedSet实现延时队列

4 Spring Boot环境中连接Redis实现发布/订阅

4.1 两个消费者

4.2 消息监听配置

4.3 生产者

4.4 测试结果

--------------------

## 前言 ##

Redis是现在最流行的key-value数据库，有诸多应用场景，包括缓存、分布式锁、计数器等，本文介绍的是用Redis实现消息队列。消息队列有专门的中间件，包括RabbitMQ、RocketMQ、kafka等，它们都属于重量级。在项目中，如果需求没必要用到重量级的消息中间件，可以采用Redis来实现消息队列。

## 1 队列的特点 ##

队列是一个线性的数据结构，有两个基本操作，**入队**与**出队**——入队是将数据写入队尾，出队是取出队头的一个数据，也就是常说的FIFO（First Input First Out），**先进先出**。

![2020031417192518.png][]

## 2 使用List实现简单队列 ##

### 2.1 思路 ###

Redis的List类型可以从列表的表头（最左边）或者表尾（最右边）插入元素，当然同样也可以从表头或者表尾删除元素。基于这个特性，假设List的**最左边元素是队头**，**最右边元素是队尾**，那么往List的右边插入元素就是**入队**，往List的最左边删除元素就是**出队**。

### 2.2 实现简单队列的相关命令 ###

刚好，Redis的List类型就支持这样的操作，会用到的命令包括**Rpush 、Lpop、Blpop。**其中，**Rpush**命令用于将一个或多个值插入到列表的尾部（最右边），这里可以用来入队。**Lpop与Blpop**命令都可用于出队，其中**Lpop**命令用于移除并返回列表的第一个元素，当指定的key不存在时，返回nil；**Blpop**也是移除并返回列表的第一个元素，与Lpop命令不同的是，**Blpop在列表中没有元素时会阻塞直到等待超时或发现可弹出元素为止。**

![2020031510220778.png][]

### 2.3 RedisTemplate操作List实现消息队列 ###

这里用SpringBoot + RedisTemplate实现简单消息队列，代码包含两部分，生产者（入队）与消费者（出队）。

首先是**入队（生产者）**代码，这里相当于是使用Rpush在key为"simpleQueue"的List中写入了三个值"Java"、"C++"与"Python"。

然后是**出队（消费者）**代码，redisTemplate.opsForList().leftPop有多个重载方法，这里的写法相当于是调用了Blpop命令，并设置了60秒的超时时间，以阻塞的方式弹出元素。

## 3 使用SortedSet实现延时队列 ##

### 3.1 延时队列应用场景 ###

*  创建的所有订单如果15分钟内未操作，自动取消；
 *  发送短信，11点整发送短信，2分钟后再次执行；
 *  用户下了外卖的单，商家在5分钟内未接单，需要自动取消订单。

### 3.2 思路 ###

这里使用Redis的**SortedSet**类型实现延时队列。首先，SortedSet与Set一样都是String类型元素的集合，并且元素不允许有重复，与Set不同的地方在于SortedSet每一个元素会关联一个double类型的分数，redis通过这一个分数为SortedSet的成员排序，另外这一个分数值是可以重复的。

现在有这样的需求，将创建时间超过15分钟的订单关闭，基于SortedSet的特性可以这样做。

在生产者这边我们以一个**SortedSet**集合为延时队列，**key为常量，这里定义为orderId。**在创建订单时，将订单号写入这个key为orderId的SortedSet，也就是**value存订单号的值，score存订单创建时往后推十五分钟的时间戳。**

**   **    在消费者这边根据key获取集合中的所有元素，遍历这些元素，将score小于当前时间戳的value值删除，并消费这些消息。

### 3.3 实现延时队列相关命令 ###

**       **实现延时队列会用到SortedSet相关的命令有：

*  **Zadd**：用于将一个或多个元素及其分数值加入到有序集中；
 *  **Zrangebyscore：**用于返回指定分数区间的成员元素列表，结果按分数由小到大排列；
 *  **Zrem**：移除集合中一个或多个元素。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3k1MDY3OTgyNzg_size_16_color_FFFFFF_t_70][]

### 3.4 RedisTemplate操作SortedSet实现延时队列 ###

依然是用SpringBoot + RedisTemplate，包含了生产者与消费者的代码，当然RedisTemplate对命令作了封装，API的名称与redis本身的命令命名是不同的。

**生产者：**

**消费者：**

import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.core.ZSetOperations;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.Iterator;
    import java.util.Set;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class MessageListener {
        @Autowired
        private RedisTemplate<String, String> redisTemplate;
    
        @Test
        public void listenMessage() {
            while(true){
                Set<ZSetOperations.TypedTuple<String>> orderIdSet = redisTemplate.opsForZSet().rangeWithScores("orderId", 0, -1);
                if(orderIdSet == null || orderIdSet.isEmpty()){
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        break;
                    }
                    continue;
                }
                Iterator<ZSetOperations.TypedTuple<String>> iterator = orderIdSet.iterator();
                ZSetOperations.TypedTuple<String> next = iterator.next();
                Double score = next.getScore();
                if(score == null) {
                    continue;
                }
                double nowTime = System.currentTimeMillis();
                if(nowTime >= score) {
                    String value = next.getValue();
                    redisTemplate.opsForZSet().remove("orderId", value);
                    System.out.println("已成功处理一条订单，订单id为" + value);
                }
            }
        }
    }

## 4 Spring Boot环境中连接Redis实现发布/订阅 ##

发布/订阅即**Publish/Subscribe**，是一种消息通信模式。在这种模式下可以有多个消费者订阅任意数量的频道，生产者往频道发送消息，订阅了该频道的**所有**消费者会收到消息。

Redis本身就支持发布/订阅，订阅频道的命令为SUBSCRIBE，发布消息的命令为PUBLISH，下面就以**两个消费者订阅同一个频道，**另有**一个生产者发布消息**为例，用Spring Boot + RedisTemplate实现功能并测试。

### 4.1 两个消费者 ###

定义两个消费者，两个消费者中各有一个方法消费消息。

/**
     * 消费者一
     */
    public class SubscribeOne {
        public void receive(String message) {
            System.out.println("这里是一号订阅客户端，接收到信息：" + message);
        }
    }

/**
     * 消费者二
     */
    public class SubscribeTwo {
        public void receive(String message) {
            System.out.println("这里是二号订阅客户端，接收到信息：" + message);
        }
    }

### 4.2 消息监听配置 ###

在这一个配置中，定义了消息监听者容器与两个消息监听适配器。

import com.bigsea.Controller.ChatController;
    import com.bigsea.subscribe.SubscribeOne;
    import com.bigsea.subscribe.SubscribeTwo;
    import org.springframework.context.annotation.Bean;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    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.stereotype.Component;
    
    @Component
    public class subscribeConfig {
        private static final String RECEIVE_NAME = "receive";
        /**
         * 消息监听适配器一
         * @return MessageListenerAdapter
         */
        @Bean
        public MessageListenerAdapter listenerAdapterOne() {
            return new MessageListenerAdapter(new SubscribeOne(), RECEIVE_NAME);
        }
    
        /**
         * 消息监听适配器二
         * @return MessageListenerAdapter
         */
        @Bean
        public MessageListenerAdapter listenerAdapterTwo() {
            return new MessageListenerAdapter(new SubscribeTwo(), RECEIVE_NAME);
        }
    
        /**
         * 定义消息监听者容器
         * @param connectionFactory 连接工厂
         * @param listenerAdapterOne MessageListenerAdapter
         * @param listenerAdapterTwo MessageListenerAdapter
         * @return RedisMessageListenerContainer
         */
        @Bean
        public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                       MessageListenerAdapter listenerAdapterOne,
                                                       MessageListenerAdapter listenerAdapterTwo) {
            RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
            listenerContainer.setConnectionFactory(connectionFactory);
            listenerContainer.addMessageListener(listenerAdapterOne, new PatternTopic(ChatController.CHAT_NAME));
            listenerContainer.addMessageListener(listenerAdapterTwo, new PatternTopic(ChatController.CHAT_NAME));
            return listenerContainer;
        }
    }

### 4.3 生产者 ###

这里的生产者发布消息给“myMessage”频道，两个消费者同样也是监听的这一频道。

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @RequestMapping("/chat")
    public class ChatController {
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
    
        public static final String CHAT_NAME = "myMessage";
    
        private static final int MESSAGE_COUNT = 10;
    
        @GetMapping("/pub")
        public void publish() {
            for(int i = 1; i <= MESSAGE_COUNT; i++) {
                stringRedisTemplate.convertAndSend(CHAT_NAME, "发布的第" + i + "条消息");
            }
        }
    }

### 4.4 测试结果 ###

启动这一个Spring Boot应用，直接在浏览器中通过URL访问，控制台打印结果如下。

这里是一号订阅客户端，接收到信息：发布的第1条消息
    这里是一号订阅客户端，接收到信息：发布的第2条消息
    这里是二号订阅客户端，接收到信息：发布的第2条消息
    这里是二号订阅客户端，接收到信息：发布的第1条消息
    这里是一号订阅客户端，接收到信息：发布的第3条消息
    这里是二号订阅客户端，接收到信息：发布的第3条消息
    这里是二号订阅客户端，接收到信息：发布的第4条消息
    这里是一号订阅客户端，接收到信息：发布的第4条消息
    这里是二号订阅客户端，接收到信息：发布的第5条消息
    这里是一号订阅客户端，接收到信息：发布的第5条消息
    这里是一号订阅客户端，接收到信息：发布的第6条消息
    这里是二号订阅客户端，接收到信息：发布的第6条消息
    这里是一号订阅客户端，接收到信息：发布的第7条消息
    这里是二号订阅客户端，接收到信息：发布的第7条消息
    这里是一号订阅客户端，接收到信息：发布的第8条消息
    这里是二号订阅客户端，接收到信息：发布的第8条消息
    这里是一号订阅客户端，接收到信息：发布的第9条消息
    这里是二号订阅客户端，接收到信息：发布的第9条消息
    这里是二号订阅客户端，接收到信息：发布的第10条消息
    这里是一号订阅客户端，接收到信息：发布的第10条消息

[2020031417192518.png]: /images/20230529/474bb765cbc5446487cdc1de5cc8a83d.png
[2020031510220778.png]: /images/20230529/249a4c3f84a2494a8efc544fecc53fc7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3k1MDY3OTgyNzg_size_16_color_FFFFFF_t_70]: /images/20230529/3a18fb51ecea428f9540680b769582cd.png