当前位置：首页 > news >正文

Redis：基于PubSub（发布/订阅）、Stream流实现消息队列

news 2026/5/30 12:26:54

Redis - PubSub、Stream流

文章目录

Redis - PubSub、Stream流
- 1.基于List的消息队列
- 2.基于PubSub的消息队列
- 3.基于Stream的消息队列
- - 1.Redis Streams简介
  - 2.Redis Streams基本命令
  - - 1.XADD 添加消息到末尾
    - 2.XLEN 获取消息长度
    - 3.XREAD 读取消息（单消费模式）
    - 4.XGROUP 消费组操作
    - 5.XREADGROUP GROUP 从消费组读取消息
    - 6.XACK 消息确认
    - 7.XPENDING 查看pend数据

1.基于List的消息队列

由于redis的list数据结构为双向链表，则可以通过lpush和rpop来模拟队列效果
由于队列没有消息时候，需要阻塞获取队列数据，而lpop和rpop在空队列获取数据时会返回null，所以需要使用brpop和blpop来进行阻塞获取

#向data1的list存两个数据
lpush data1  aaa bbb

#右监听data1  等待20秒
brpop data1 20

在这里插入图片描述

缺点:

无法避免消息丢失,
只支持单消费者，无法广播

2.基于PubSub的消息队列

基于发布订阅形式，可以广播，生产者向channel(信道)发送消息，可以由多个消息者去订阅，订阅的消费者都可以收到消息

 SUBSCRIBE channel [channel ...]  #订阅一个或多个信道

 PUBLISH channel message  #向一个信道发送消息

 PSUBSCRIBE pattern [pattern ...]  #通过通配符匹配订阅的信道  匹配规则  ？代表一个字符   []代表中括号内的可选字符  *代表任意字符

SUBSCRIBE  log 
PUBLISH  log zhangsan

在这里插入图片描述

缺点:

不支持持久化
消息有上限，超出会导致消息丢失

3.基于Stream的消息队列

1.Redis Streams简介

官方文档：https://redis.io/docs/latest/commands/xadd/

Redis Stream是redis在5.x版本引入的新特性，Redis流是一种数据结构，它类似于一个只可追加的日志，但也实现了多种操作，以克服典型只可追加日志的一些限制。这些操作包括O(1)时间的随机访问和复杂的消费策略，如消费者组。你可以使用流来记录并同时实时分发事件。Redis流的使用案例包括：

事件溯源（例如，跟踪用户操作、点击等）
传感器监测（例如，现场设备的读数）
通知（例如，将每个用户的通知记录存储在单独的流中）

Redis为每个流条目生成一个唯一的ID。你可以使用这些ID在后续检索与其关联的条目，或者读取并处理流中的所有后续条目。请注意，由于这些ID与时间相关，这里显示的ID可能会有所不同，与你自己的Redis实例中看到的ID也会有所不同。
Redis流支持多种修剪策略（以防止流无限制地增长）和多种消费策略（参见XREAD、XREADGROUP和XRANGE）。

2.Redis Streams基本命令

stream消息队列相关命令：

XADD - 添加消息到末尾
XTRIM - 对流进行修剪，限制长度
XDEL - 删除消息
XLEN - 获取流包含的元素数量，即消息长度
XRANGE - 获取消息列表，会自动过滤已经删除的消息
XREVRANGE - 反向获取消息列表，ID 从大到小
XREAD - 以阻塞或非阻塞方式获取消息列表

消费者组相关命令：

XGROUP CREATE - 创建消费者组
XGROUP CREATECONSUMER 给指定的消费者组添加消费者
XREADGROUP GROUP - 读取消费者组中的消息
XACK - 将消息标记为"已处理"
XGROUP SETID - 为消费者组设置新的最后递送消息ID
XGROUP DELCONSUMER - 删除消费者
XGROUP DESTROY - 删除消费者组
XPENDING - 显示待处理消息的相关信息
XCLAIM - 转移消息的归属权
XINFO - 查看流和消费者组的相关信息；
XINFO GROUPS - 打印消费者组的信息；
XINFO STREAM - 打印流信息

1.XADD 添加消息到末尾

1.基本语法

XADD是唯一可以向流中添加数据的 Redis 命令，但还有其他命令，例如 XDEL 和 XTRIM，能够从流中删除数据。

XADD key [NOMKSTREAM] [<MAXLEN | MINID> [= | ~] threshold [LIMIT count]] <* | id> field value [field value ...]

key: 队列名
[NOMKSTREAM]：队列不存在是否自动创建，默认自动创建
[<MAXLEN | MINID> [= | ~] threshold [LIMIT count]] :设置消息队列的最大消息数量
<* | id>：消息唯一id，*代表消息由redis自动生成，格式为时间戳-递增序列,可以手动指定
field value：字段和值（键值对），可以一次添加多个

XADD users * name zhangsan age 18  #向user发送一条name为zhangsan，age为18的消息，返回消息id

在这里插入图片描述

2.指定stream的id参数

1526919030474-55

ID标识流中的给定消息数据。
如果指定的ID参数是*字符，XADD命令将自动生成一个唯一的ID。然而，尽管仅在极少数情况下有用，但可以指定一个格式良好的ID，以便新条目将与指定的ID完全相同。

XADD mystream 1526919030474-55 message "Hello,"

当自动生成ID时，第一部分是Redis实例生成ID的Unix时间（毫秒）。第二部分只是一个序列号，用于区分在同一毫秒内生成的ID。

XADD mystream 1526919030474-* message " World!"

还可以指定一个不完整的ID，只自动生成序列号部分（注意：6.0版本不支持报错）

stream的数据是有序的，所以消息的id始终的递增的，如果手动指定一个小于上一条数据的id则会出错

2.XLEN 获取消息长度

XLEN users  #返回消息个数

3.XREAD 读取消息（单消费模式）

基础语法

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]

[COUNT count] :每次读取的最大数量
[BLOCK milliseconds]:当消息没有时，是否阻塞，阻塞时间（毫秒），不阻塞就不给值，如果给0则永久阻塞等待
STREAMS key: 从那个队列读取消息，key为读取的队列名
id [id ...]：起始id，代表从那个id的消息开始读取；0代表从第一个，$代表从最新的消息读取

测试

 xread count 1 streams users 0  #读取users中最开始的一条数据

在这里插入图片描述

消息读取后不会删除，所有消费者都可以重复获取

 xread count 1 streams users $  #读取最新消息 ，返回为nil空xread count 1 block 0 streams users $    #永久阻塞读取

在这里插入图片描述

但是阻塞方式监听到消息后会关闭，需要重新监听

此时在开发中我们可以使用死循环来无限读取最新消息进行监听

但是：当指定起始id为$时，代表读取最新消息，如果处理消息过程中，又有超过一条以上的消息到达，则下次也只能获取一条最新的消息会导致其他数据漏读

4.XGROUP 消费组操作

消费者组：将多个消费者划分到同一个消费组，监听同一个队列
分流消费：队列中的消费将会分流给消费者组中的消费者，不会重复消费，加快消息消费速度
消息标识：消费者组会维护一个标识，记录最后一个被处理（非最新）的消息，即使redis挂机重启，也可以按照标识恢复读取，确保消息消费
消息确认机制：消费者获取消息后，消息处于pending状态，并存入一个pend-list，当处理完成时通过XACK来确认消息，标记为已处理，才pend-list移除

XGROUP CREATE 创建消费者组

XGROUP CREATE key group <id | $> [MKSTREAM] [ENTRIESREAD entries-read]

key：队列名称
group： 消费者组名称
<id | $>：起始id标识， 0代表第一个， $代表最新消息
[MKSTREAM]：队列不存在时自动创建队列，如果不存在且不指定会报错
[ENTRIESREAD entries-read]: redis7.0后的参数，

创建消费者组g1

XGROUP CREATE users g1 0

在这里插入图片描述

XGROUP CREATECONSUMER 给指定的消费者组添加消费者

XGROUP CREATECONSUMER key group consumer

XGROUP DESTROY 删除指定的消费者组

XGROUP DESTROY key group

XGROUP DELCONSUMER 删除消费者组中指定的消费者

XGROUP DELCONSUMER key group consumer

5.XREADGROUP GROUP 从消费组读取消息

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] id [id ...]

group： 消费者组名称
consumer：消费者名称，如果不存在会自动创建
[COUNT count] :每次读取的最大数量
[BLOCK milliseconds]:当消息没有时，是否阻塞，阻塞时间（毫秒），不阻塞就不给值，如果给0则永久阻塞等待
STREAMS key: 从那个队列读取消息，key为读取的队列名
[NOACK] 无需消息确认(类似自动确认)。
id [id ...]：起始id

**注意：**id取值：">" ：从下一个未消费的消息开始，非最新消息，确保都消费
其他数字：根据指定id从pend-list中获取已消费但未确认消息，例如0，从pend-list第一个消息开始

所以当正常处理时的id都采用">" 进行消费，如果出现异常可以指定0，每次都读取第一个pend-list的消息，即每次都是读取最新的未处理数据，将异常数据处理掉

测试

XREADGROUP GROUP  g1 c1 COUNT 1 BLOCK 2000 STREAMS users >

在这里插入图片描述

6.XACK 消息确认

XACK key group id [id ...]

测试

XACK users g1 1733738565351-0 1733738570018-0 1733738567511-0 1733738587327-0

在这里插入图片描述

7.XPENDING 查看pend数据

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

key：队列名称
group： 消费者组名称
[IDLE min-idle-time]：:查看过去空闲时间的以上的消息，比如给5000，则查询空闲时间5000ms以上的消息
start end 消息起始范围 “- +”代表所有
count： 获取数量
[consumer]: 获取那个消费者的

测试

XPENDING users g1 - + 10

在这里插入图片描述

参考来源：https://www.bilibili.com/video/BV1cr4y1671t/?spm_id_from=333.788.videopod.episodes&vd_source=97a7d9497f7eb9e537f6b50df8831e27&p=75

Redis - PubSub、Stream流

文章目录

1.基于List的消息队列

2.基于PubSub的消息队列

3.基于Stream的消息队列

1.Redis Streams简介

2.Redis Streams基本命令

1.XADD 添加消息到末尾

2.XLEN 获取消息长度

3.XREAD 读取消息 （单消费模式）

4.XGROUP 消费组操作

5.XREADGROUP GROUP 从消费组读取消息

6.XACK 消息确认

7.XPENDING 查看pend数据

相关文章：

3.XREAD 读取消息（单消费模式）