Redis中的各种消息队列

基于list结构模拟消息队列

image-20250313210528973

image-20250313210651957

基于PubSub的消息队列

image-20250313211130893

image-20250313211147273

基于Stream的消息队列

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

发送消息的命令：

image-20250313212401765

例如：

image-20250313212445290

读取消息的方式之一：XREAD

image-20250313212511690

例如，使用XREAD读取第一个消息：

image-20250313212533755

XREAD阻塞方式，读取最新的消息：

image-20250313212550739

在业务开发中，我们可以循环的调用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下

image-20250313212624747

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

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

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

基于Stream的消息队列-消费者组

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

image-20250314121637242

创建消费者组：

image-20250314121700920

key：队列名称
groupName：消费者组名称
ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
MKSTREAM：队列不存在时自动创建队列
其它常见命令：

删除指定的消费者组

1
XGROUP DESTORY key groupName

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

1
XGROUP CREATECONSUMER key groupname consumername

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

1
XGROUP DELCONSUMER key groupname consumername

从消费者组读取消息：

1
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • group：消费组名称
  • consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
  • count：本次查询的最大数量
  • BLOCK milliseconds：当没有消息时最长等待时间
  • NOACK：无需手动ACK，获取到消息后自动确认
  • STREAMS key：指定队列名称
  • ID：获取消息的起始ID：

“>”：从下一个未消费的消息开始
其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

消费者监听消息的基本思路：

image-20250314121807451

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

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

最后我们来个小对比：

image-20250314121925725