Redis消息队列

￡神魔★判官ぃ 2024-03-31 08:49 106阅读 0赞

#### Redis消息队列 ####

*  1.认识消息队列
 *  2.Redis消息队列-基于List实现消息队列
 *  3.Redis消息队列-基于PubSub的消息队列
 *  4.Redis消息队列-基于Stream的消息队列
 *  5.Redis消息队列-基于Stream的消息队列-消费者组

## 1.认识消息队列 ##

什么是消息队列：字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

*  消息队列：存储和管理消息，也被称为消息代理（Message Broker）
 *  生产者：发送消息到消息队列
 *  消费者：从消息队列获取消息并处理消息

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

> 使用队列的好处在于解耦：所谓解耦，举一个生活中的例子就是：快递员(生产者)把快递放到快递柜里边(Message Queue)去，我们(消费者)从快递柜里边去拿东西，这就是一个异步，如果耦合，那么这个快递员相当于直接把快递交给你，这事固然好，但是万一你不在家，那么快递员就会一直等你，这就浪费了快递员的时间，所以这种思想在我们日常开发中，是非常有必要的。

这种场景在我们秒杀中就变成了：我们下单之后，利用redis去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快我们的响应速度。

这里我们可以使用一些现成的mq，比如kafka，rabbitmq等等，但是呢，如果没有安装mq，我们也可以直接使用redis提供的mq方案，降低我们的部署和学习成本。

## 2.Redis消息队列-基于List实现消息队列 ##

**基于List结构模拟消息队列**

消息队列（Message Queue），字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表，很容易模拟出队列效果。

队列是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。  
不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。

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

**代码实现：**

> lrange l1 0 -1
    3
    2
    1
    > rpop l1
    1
    > lrange l1 0 -1
    3
    2
    > ping
    PONG
    > brpop l1 200
    l1
    2
    > brpop l1 200
    l1
    3
    > lpush l1 4
    1
    > brpop l1 200
    l1
    4

**基于List的消息队列有哪些优缺点？**  
优点：

*  利用Redis存储，不受限于JVM内存上限
 *  基于Redis的持久化机制，数据安全性有保证
 *  可以满足消息有序性

缺点：

*  无法避免消息丢失
 *  只支持单消费者

## 3.Redis消息队列-基于PubSub的消息队列 ##

PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。

SUBSCRIBE channel \[channel\] ：订阅一个或多个频道  
PUBLISH channel msg ：向一个频道发送消息  
PSUBSCRIBE pattern\[pattern\] ：订阅与pattern格式匹配的所有频道  
![在这里插入图片描述][cc72f0ba6d1a438ebaecfd3889a97a95.png]  
**代码实现：**  
生产者：

> publish order.queue nihao
    0
    > publish order.queue nihao
    1
    > publish order.queue cxk
    1
    > publish order.queue woshixiaoheizi
    1
    > publish order.queue1 cxk
    1

消费者：

> SUBSCRIBE order.queue hello
    2
    
    order.queue
    nihao
    
    order.queue
    cxk
    
    order.queue
    woshixiaoheizi

消费者2：

> psubscribe order.*
    1
    
    order.*
    order.queue1
    cxk

**基于PubSub的消息队列有哪些优缺点？**  
优点：

*  采用发布订阅模型，支持多生产、多消费

缺点：

*  不支持数据持久化
 *  无法避免消息丢失
 *  消息堆积有上限，超出时数据丢失

## 4.Redis消息队列-基于Stream的消息队列 ##

> Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

**相关命令**

**这是发送消息的命令**

> XADD key [NOMKSTREAM] [<MAXLEN | MINID> [= | ~] threshold [LIMIT count]] <* | id> field value [field value ...]
    # [NOMKSTREAM] 默认是自动创建队列，如果队列不存在，自动创建队列
    # [<MAXLEN | MINID> [= | ~] threshold [LIMIT count]]  消息队列的最大消息数量
    # <* | id> 消息的唯一id，*代表由redis自动生成，格式是“时间戳-递增数字”
    # field value [field value ...] 发送到队列中的消息，称为entry，格式就是key-value键值对
    
    > xadd user * name zhangsan age 18
    1668834399741-0

**读取消息的命令**

> XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
    # [COUNT count] 每次读取消息的最大数量
    # [BLOCK milliseconds] 当没有消息时，是否阻塞，阻塞时长 ms
    # STREAMS key [key ...] 从哪个队列中读取 key代表队列名
    # id [id ...] 其实id，只返回大于该id的消息，0代表从第一个消息开始，$：从最新的消息开始
    # 获取之前那条消息
    > xread count 1 streams user 0
    user
    1668834399741-0
    name
    zhangsan
    age
    18
    # 通过阻塞的方式去等待生产者发送消息
    > xread count 1 block 100000 streams user $
    user
    1668834822049-0
    name
    lisi
    age
    14
    # 只要一发送消息，消费者就会接收
    > xadd user * name lisi age 14
    1668834822049-0

注意：当我们指定起始ID为$时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题

**STREAM类型消息队列的XREAD命令特点：**

*  消息可回溯
 *  一个消息可以被多个消费者读取
 *  可以阻塞读取
 *  有消息漏读的风险

## 5.Redis消息队列-基于Stream的消息队列-消费者组 ##

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：  
![在这里插入图片描述][e6690c5d5c1f40bdb2060dac735b87aa.png]  
**相关命令**

**创建消费者组**

> XGROUP CREATE key groupname <id | $> [MKSTREAM][ENTRIESREAD entries_read]
    # key：队列名称
    # groupName：消费者组名称
    # ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
    # MKSTREAM：队列不存在时,自动创建队列

**给指定的消费者组添加消费者**

> XGROUP CREATECONSUMER key groupname consumername

**删除消费者组中的指定消费者**

> XGROUP DELCONSUMER key groupname consumername

**删除消费者组**

> XGROUP DESTROY key groupname

**从消费者组中读取消息**

> XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds][NOACK] STREAMS key [key ...] id [id ...]
    # group 消费者组名
    # consumer 消费者名 ，如果消费者不存在，会自动创建一个消费者
    # COUNT 读取多少条消息，本次查询的最大数量
    # BLOCK 当没有消息时最长等待时间
    # NOACK 无需手动ack，获取消息自动确认
    # STREAMS key [key ...] 指定队列名称
    # id [id ...] - 获取消息的起始ID：
    ">"：从下一个未消费的消息开始
    其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

读取消息

# 创建一个消费者组
    > xgroup create user g1 0
    OK
    # 给消费者组添加一个消费者
    > xgroup createconsumer user g1 c1
    1
    # 读取消息
    > xreadgroup group g1 c1 count 1 streams user >
    user
    1668834399741-0
    name
    zhangsan
    age
    18
    > xreadgroup group g1 c1 count 1 streams user >
    user
    1668834822049-0
    name
    lisi
    age
    14
    > xreadgroup group g1 c1 count 1 streams user >
    null
    # 确认消息
    > XACK key group id [id ...]
    > XACk user g1 1668834399741-0
    1
    # 未确认的消息
    > XPENDING key group [[IDLE min-idle-time] start end count [consumer]]
    # key 队列名称
    # group 消费者组名称
    # [IDLE min-idle-time] start end  最小时间到最大时间
    # count [consumer] 消费者数量
    > xpending user g1 - + 10
    1668834822049-0
    c1
    361620
    1

**STREAM类型消息队列的XREADGROUP命令特点：**

*  消息可回溯
 *  可以多消费者争抢消息，加快消费速度
 *  可以阻塞读取
 *  没有消息漏读的风险
 *  有消息确认机制，保证消息至少被消费一次

最后我们来个小对比

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>List</th> 
   <th>PubSub</th> 
   <th>Stream</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>消息持久化</td> 
   <td>支持</td> 
   <td>不支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>阻塞读取</td> 
   <td>支持</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>消息堆积处理</td> 
   <td>受限于内存空间，可以利用多消费者加快处理</td> 
   <td>受限于消费者缓冲区</td> 
   <td>受限于队列长度，可以利用消费者组提高消费速度，减少堆积</td> 
  </tr> 
  <tr> 
   <td>消息确认机制</td> 
   <td>不支持</td> 
   <td>不支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>消息回溯</td> 
   <td>不支持</td> 
   <td>不支持</td> 
   <td>支持</td> 
  </tr> 
 </tbody> 
</table>

[eaf6fd94c71b475182c21d35b7d3ade7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/c7293f07dc0d4fa3ab8c4b8ad2856c8c.png
[c4ec1578381e468aa1cbefc7492515bb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/c8064707b6c2411d9259cc0a2b8764d1.png
[cc72f0ba6d1a438ebaecfd3889a97a95.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/2e138b77a2ff43cf95c5998899c7ff6d.png
[e6690c5d5c1f40bdb2060dac735b87aa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/77a227ad5e1a4567a76bf6aea856a373.png