章目录

Queue)从概念上来理解就是用来存放消息的队列最简单的消息队列模型包括以下三个角色

生产者发送消息到消息队列消息队列存储和管理信息也被称为消息代理(Message

List结构基于List结构模拟消息队列PubSub基本的点对点消息模型Stream比较完善的消息队列模型推荐

这种方式比较简单因为Redis的list数据结构是一个双向链表很容易模拟出队列的效果。

BRPOP或者

注这里使用BRPOP而不是RPOP是因为BRPOP能够实现阻塞的效果而RPOP不能

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

缺点:

PubSub(发布订阅)是Redis2.0版本引入的消息传递模型消费者可以订阅一个或多个channel(频道)生产者向对应channel发送消息后所有订阅者都能收到相关消息

SUBSCRIBE

pattern[pattern]订阅与pattern格式匹配的所有频道

具体操作如下所示

不能支持数据持久化一旦redis宕机数据就会丢失无法避免消息丢失消息堆积有上限超出上限后数据会丢失

Stream是Redis5.0引入的一种新数据类型能够实现功能完善的消息队列因为它本身就是一个消息队列所以我们可以直接通过命令来使用它

3.1

[NOMKSTREAM]如果队列不存在是否自动创建队列默认是自动创建

[MAXLEN|MINID

*|ID消息的唯一id表示由Redis自动生成格式是“时间戳-递增数字”一般推荐使用来自动生成

field

milliseconds]当队列中没有消息时阻塞指定时长单位为秒STREAMS

key

…]要从哪个队列中读取消息key就是队列名可以指定多个队列ID

[ID

…]起始ID只返回大于该ID的消息其中0代表从第一个消息开始$代表从最新的消息开始

举个栗子

注在上述测试中我们只往users队列中添加了一个消息这个时候如果ID使用$来获取最新消息且设置了阻塞等待的话此时读取信息将在阻塞时间过后返回空:

3.3

key队列名称groupName消费者组名称ID起始ID标识0代表队列中第一个消息$代表队列最后一个消息MKSTREAM队列不存在时自动创建队列

注这里要求队列key已经存在才能创建消费者组否则需要开启MKSTREAM让其自动创建新的队列

XGROUP

group消费者组名称consumer消费者名称如果消费者不存在会根据该名称自动创建一个消费者count本次查询的最大数量BLOCK

milliseconds阻塞时间没有消息时会进行等待以毫秒为单位NOACK选择后无需手动ACK获取到消息后自动确认一般不建议设置当我们获取完消息后需要手动确认ackSTREAMS

“”表示从下一个未消费的消息开始其它根据指定id从pending-list中获取消息pending-list用于专门存放那些已消费但未确认的消息例如此时ID为0表示获取pending-list中的第一个消息.

注同一个消费者组中的消费者读取同一个消息队列时若ID使用来读取则下一个读取的消息一定是前面的消费者没有读取到的消息直到消息队列中的消息都被读取过后最后一个读取的消费者返回nil

举个栗子

通过XREADGROUP命令为消费者组g1添加消费者c1、c2、c3来读取list队列消息

可以看到同一个消费者组中的消费者它们都在获取同一个队列中的消息且ID使用来读取下一个读取的消息一定是前面的消费者没有读取到的消息直到消息全部被读完后只返回nil

注每读取完一条消息我们需要对它进行手动确认使其从pending-list中移除使用下述命令可以查看已读取但还未确认的消息

XPENDING

[consumer]]我们查看一下list中还有多少未确认的消息

好的都没被确认所以需要我们手动去确认消息使其从pending-list中移除操作命令如下

XACK

stringRedisTemplate.opsForStream().read(Consumer.from(g1,

c1),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(list,

);//ACK确认

stringRedisTemplate.opsForStream().acknowledge(list,

g1,

stringRedisTemplate.opsForStream().read(Consumer.from(g1,

c1),StreamReadOptions.empty().count(1),StreamOffset.create(list,

ReadOffset.from(0))

上述代码可以配合循环实现被消费者组不断监听的消息队列以上便是对Redis实现消息队列的介绍了如果内容对大家有帮助的话请给这篇文章一个三连关注吧(

)✧✨