【redis-06】redis的stream流实现消息中间件

redis系列整体栏目

---

内容	链接地址
【一】redis基本数据类型和使用场景	https://zhenghuisheng.blog.csdn.net/article/details/142406325
【二】redis的持久化机制和原理	https://zhenghuisheng.blog.csdn.net/article/details/142441756
【三】redis缓存穿透、缓存击穿、缓存雪崩	https://zhenghuisheng.blog.csdn.net/article/details/142577507
【四】redisson实现分布式锁实战和源码剖析	https://zhenghuisheng.blog.csdn.net/article/details/142646301
【五】redis保证和mysql数据一致性	https://zhenghuisheng.blog.csdn.net/article/details/142687101
【六】redis的stream流实现消息中间件	https://zhenghuisheng.blog.csdn.net/article/details/142721269

---

如需转载，请输入：https://blog.csdn.net/zhenghuishengq/article/details/142721269

一，redis的新特性-队列stream

在了解这个redis的新特性之前，可以先查看一下官网的详细文档：stream流官方文档

redis的stream流是从5.0版本才开始提出，本人这里安装的是 6.2.6 版本。它的底层原理是借鉴于kafka的底层实现，因此可以参考本人前面的写的kafka的文章。redis 的stream流队列其主要组件有：消息队列、生产者、消费者、消费者组、消息及消息id、偏移量等

建立这个stream的主要原因，是作者想通过这个redis来取代mq那些中间件，redis在项目中时必不可少的，但是引入mq就会多引入一个第三方的中间件，让系统稳定性没那么高，mq一挂就有可能导致整个系统瘫痪

1，redis的stream流基本使用

创建一个stream队列的命令如下，通过xadd的方式实现往队列中添加消息。如下创建一个商品的队列，然后设置商品type类型为小米手机，商品名称name为小米8，得到的结果如下图

xadd product_queue * type xiaomi name xiaomi8

product_quque 表示队列的名称，***** 表示由服务器自动生成一个id，其id通过时间戳+序号(毫秒时间内第n条消息)

可以直接通过 xlen 命令查看队列的长度，可以发现已经队列的长度为4

xlen product_queue

也可以直接通过 xrange 命令将全部的消息展示出来，在后面需要加上 - + 两个命令操作符，也可以在后面加一个id来设置范围 。

• - 表示在这个队列中的最小的id，
• + 表示在这个队列里面最大的id

xrange product_queue - +
xrange product_queue - 1728139368101-0   //获取前两个
xrange product_queue 1728139374509-0 +   //获取后两个

删除命令也比较简单，可以直接通过 xdel 命令实现删除，执行完命令之后，可以发现队列中的数据已被删除。但是这个xdel使用的是逻辑删除消息，而不是物理删除。

xdel product_queue 1728139368101-0

也可以查看整个stream队列的详细信息，可以直接通过 xinfo 命令来实现。其返回信息如下，length表示返回4条数据，

xinfo stream product_queue

返回的消息如下，会将整个队列的信息详细的返回，并且根据不同的redis版本返回一些不同的额外参数

127.0.0.1:6379> XINFO STREAM product_queue1) "length"	2) (integer) 5					//表示5条数据3) "radix-tree-keys"	4) (integer) 1					//用于存储 Stream 元素的 Radix Tree 中的键数量5) "radix-tree-nodes"6) (integer) 2					//Radix Tree 中的节点数量，反映了树的复杂度7) "groups"8) (integer) 2					//与此 Stream 相关的消费组（Consumer Group）数量9) "last-generated-id"
10) "1608049761947-0"			//Stream 中最后一个生成的消息的 ID
11) "first-entry"
12) 1) "1608049732151-0"2) 1) "name"2) "item1"
13) "last-entry"
14) 1) "1608049761947-0"2) 1) "name"2) "item5"

2，stream队列消息消费

由于redis的stream流主要是借鉴于kafka，因此其内部消费方式和kafka一样，主要有单消费者和消费者组 。由于上面已经往 product_queue 队列中投递了消息，因此接下来主要讲解消息如何被消费

2.1，单消费者

单消费者也比较好理解，就是此时不属于任何一个消费者组中的消费者。其消费方式如下，可以直接通过 xread 的方式进行消息的读取

xread count 1 streams product_queue 0-0

• count表示读取消息的条数，比如后面接1表示只读取一条数据
• streams表示一个关键字，需要配合xread使用
• 0-0前面这个0表示读取队列最开始的消息，后面这个0表示只读取一条数据

除了从前面读取消息之外，也可以直接从后面开始读取数据，可以直接通过 $ 解决

xread count 1 streams product_queue $				//直接读取
xread block 0 count 1 streams product_queue $		//阻塞式读取    

但是直接通过这种单消费者方式实现消息消费的话，也存在着一定的缺陷，因为单消费者消费消息，其消费完成的偏移量是需要手动实现提交的，因此单消费者实现消息消费会比较的复杂。

2.2，消费者组

2.2.1，订阅消费者组

上面提到了单消费者实现消息消费需要手动的提交偏移量，这样下次才知道当前消费者的消息消费到了哪里，在redis内部中，已经提供好了一个可以自动实现消息消费后记录偏移量的功能，不需要开发者自行的去实现。

创建消费者组的命令如下，可以通过xgroup 实现，如为 product_queue 的队列创建一个consumer1的消费者组，设置从头开始读取消息

xgroup create product_queue consumer1 0-0		//从前面开始消费消息

也可以创建一个名称为consumer2的消费者组，从后面开始读取消息

xgroup create product_queue consumer2 $    		//从后面开始消费消息

可以直接通过 xinfo groups 命令来查看该队列对应的全部的消费者组的信息，可以发现此时已经有两个消费者组

xinfo groups product_queue

2.2.2，消息消费

在实现完消息订阅之后，由于kafka设计的理念是，一个分区下的消息只能被消费者组中的一条消息消费，因此redis中stream流的设计理念也一样。

其消息消费的命令如下，通过 xreadgroup 实现消费者组消费，GROUP表示一个关键字，需要和xreadGroup结合使用，consumer1表示一个消费者组，c1表示消费者中的任意一个消费者，count 1表示只消费一条消息，最后面的 > 表示获取的消息通过 last_delivered_id 后一条开始消费

xreadgroup GROUP consumer1 c1 count 1 streams product_queue >

在消息被消费完成之后，再来查看一下消费者组的详细消息，在上面执行这个命令时此时的consumer1消费者组对应的value值是0-0，当有消息被消费之后，这个消费者组对应的 last_delivered_id 就发生了改变，其指针往后移动了一位

其偏移量主要就是通过这个 last_delivered_id 来解决的，每次消费一条消息，偏移量就会往后移动一位，这样就能解决消息重复消费的问题，也不需要像单消费者一样需要手动去记录消息消费完后偏移量的记录。

也可以直接通过命令查看消费者本身的消息，通过 xinfo comsumers 结合使用，查看哪个队列下面的那个消费者组，可以发现此时有一个c1的消费者进行消费，并且有一条消息处于pending未确认的状态

xinfo consumers product_queue consumer1

当然也可以通过命令的方式手动的进行消息消费的确认机制，通过xack的机制进行手动的确认，再次查询这个消费者详细信息之后，可以发现此时处于pending未确认的状态的消息已经被确认了，此时的值为0

xack product_queue consumer1 1728141022160-0

3，stream出现之前如何实现消息中间件

3.1，list类型实现

在list的数据类型中，可以通过Lpush+Rprop的方式实现消息中间件，其原理也比较简单，生产者从列表的左边加入消息，消费者从列表的右边消费消息， 这样保证了消息先进先出（FIFO）原则，适用于简单的消息队列系统 。

• 生产者：使用 LPUSH 向列表的左边插入消息
• 消费者：使用 RPOP 从列表的右边消费消息。

但是这种数据类型也存在缺陷，只能适用于小型轻量级、快速开发的场景，如果遇到了高并发场景，或者消息需要手动确认机制等场景，那么这种list方式就不太合适

3.2， Pub/Sub 发布订阅模式

redis内部也提供了一种发布订阅的模式，其简单使用如下，就是通过publish发送消息，subscribe接收消息

PUBLISH channel message: 发布者通过该命令向 channel 发布 message。

SUBSCRIBE channel: 订阅者通过该命令订阅 channel，并接收其发布的消息。

在redis中，这种发布订阅一般比较的适用于实时场景，如实时消息推送，聊天等场景。但是缺陷也明显：

• 首先内部并没有提供持久化机制，意味着数据会丢失
• 其次内部也没有提供消息小人机制，某些高可靠场景不适合
• 消息堆积很可能造成redis宕机问题

4，stream底层设计及优化

再讲解上面的基本使用之后，再来看这幅图就比较简单。

4.1，队列设置最大值

在redis中一个队列的大小也可以设置最大值，防止因为队列太长导致内存不足而而宕机。再创建队列时可以直接通过 MAXLEN 设置最大值，并且可以通过一个 ~ 设置成一个近似值，也可以不加这个近似值成为一个精确值

//最大值设置成1000，并且是一个近似值
xadd product_queue MAXLEN ~ 1000 * type xiaomi name xiaomi8//最大值设置成1000，并且是一个精确值
xadd product_queue MAXLEN 1000 * type xiaomi name xiaomi8

近似值往往可以更加灵活，在性能上高于精准值。当达到或者超过这个设置的值的时候，redis就会触发内存淘汰策略将数据淘汰。

4.2，使用消费者组

在消费者端应该直接考虑使用消费者组而不是单消费者，每个消费者组内部有一个 last_delivered_id ，可以通过这个字段记录对应的偏移量，这样如果出现宕机或者重启等情况，就能知道消费者消费到了哪个偏移量上面，从而从根本上解决一些消息重复消费等情况

4.3，消息的应答机制

在每个消费者组中，都会有一个pendding ids的数组，这个数组会记录所有未应答的消息id，可以通过确认数组中的id来保证消息确实被消费。当消息长时间不被ack应答时，也会被触发内存淘汰策略被淘汰

4.4，优化点

如果设计的队列太多，可以考虑部署一些cluster集群、哨兵+主从集群来保证整个系统的高可用和高性能。如果一个队列中的消息被大量的生产和消费，可以考虑 写热点分散 的方式将数据多分布在几个队列里面，然后通过hash或者轮询等方式进行消息的消费