关注“数据和云”，精彩不容错过

老钱

钱文品（老钱），互联网分布式高并发技术十年老兵，目前任掌阅科技资深后端工程师。熟练使用 Java、Python、Golang 等多种计算机语言，开发过游戏，制作过网站，写过消息推送系统和MySQL 中间件，实现过开源的 ORM 框架、Web 框架、RPC 框架等

摘要

Redis 的 Stream 有哪些特别的功能？跟 kafka 有哪些异同？怎么更好的使用它呢？本文作者老钱对此调研颇多，小编读后觉得受益很大，大家也不妨详细了解下。

内容概述

Redis5.0最近被作者突然放出来了，增加了很多新的特色功能。而Redis5.0最大的新特性就是多出了一个数据结构Stream，它是一个新的强大的支持多播的可持久化的消息队列，作者坦言Redis Stream狠狠地借鉴了Kafka的设计。

Redis Stream的结构如上图所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的ID和对应的内容。消息是持久化的，Redis重启后，内容还在。

每个Stream都有唯一的名称，它就是Redis的key，在我们首次使用xadd指令追加消息时自动创建。

每个Stream都可以挂多个消费组，每个消费组会有个游标last_delivered_id在Stream数组之上往前移动，表示当前消费组已经消费到哪条消息了。每个消费组都有一个Stream内唯一的名称，消费组不会自动创建，它需要单独的指令xgroup create进行创建，需要指定从Stream的某个消息ID开始消费，这个ID用来初始化last_delivered_id变量。

每个消费组(Consumer Group)的状态都是独立的，相互不受影响。也就是说同一份Stream内部的消息会被每个消费组都消费到。

同一个消费组(Consumer Group)可以挂接多个消费者(Consumer)，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标last_delivered_id往前移动。每个消费者者有一个组内唯一名称。

消费者(Consumer)内部会有个状态变量pending_ids，它记录了当前已经被客户端读取的消息，但是还没有ack。如果客户端没有ack，这个变量里面的消息ID会越来越多，一旦某个消息被ack，它就开始减少。这个pending_ids变量在Redis官方被称之为PEL，也就是Pending Entries List，这是一个很核心的数据结构，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了没处理。

消息ID

消息ID的形式是timestampInMillis-sequence，例如1527846880572-5，它表示当前的消息在毫米时间戳1527846880572时产生，并且是该毫秒内产生的第5条消息。消息ID可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加入的消息的ID要大于前面的消息ID。

消息内容

消息内容就是键值对，形如hash结构的键值对，这没什么特别之处。

增删改查

1. xadd 追加消息

2. xdel 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度

3. xrange 获取消息列表，会自动过滤已经删除的消息

4. xlen 消息长度

5. del 删除Stream

# *号表示服务器自动生成ID，后面顺序跟着一堆key/value127.0.0.1:6379> xadd codehole * name laoqian age 30  #  名字叫laoqian，年龄30岁1527849609889-0  # 生成的消息ID127.0.0.1:6379> xadd codehole * name xiaoyu age 291527849629172-0127.0.0.1:6379> xadd codehole * name xiaoqian age 11527849637634-0127.0.0.1:6379> xlen codehole(integer) 3127.0.0.1:6379> xrange codehole - +  # -表示最小值, +表示最大值127.0.0.1:6379> xrange codehole - +1) 1) 1527849609889-0   2) 1) "name"      2) "laoqian"      3) "age"      4) "30"2) 1) 1527849629172-0   2) 1) "name"      2) "xiaoyu"      3) "age"      4) "29"3) 1) 1527849637634-0   2) 1) "name"      2) "xiaoqian"      3) "age"      4) "1"127.0.0.1:6379> xrange codehole 1527849629172-0 +  # 指定最小消息ID的列表1) 1) 1527849629172-0   2) 1) "name"      2) "xiaoyu"      3) "age"      4) "29"2) 1) 1527849637634-0   2) 1) "name"      2) "xiaoqian"      3) "age"      4) "1"127.0.0.1:6379> xrange codehole - 1527849629172-0  # 指定最大消息ID的列表1) 1) 1527849609889-0   2) 1) "name"      2) "laoqian"      3) "age"      4) "30"2) 1) 1527849629172-0   2) 1) "name"      2) "xiaoyu"      3) "age"      4) "29"127.0.0.1:6379> xdel codehole 1527849609889-0(integer) 1127.0.0.1:6379> xlen codehole  # 长度不受影响(integer) 3127.0.0.1:6379> xrange codehole - +  # 被删除的消息没了1) 1) 1527849629172-0   2) 1) "name"      2) "xiaoyu"      3) "age"      4) "29"2) 1) 1527849637634-0   2) 1) "name"      2) "xiaoqian"      3) "age"      4) "1"127.0.0.1:6379> del codehole  # 删除整个Stream(integer) 1

将将将将~

surprise

重磅活动

OCM联盟线下大咖开讲了！三位重磅导师强势来袭，干货满满。有这么好的学习机会你还在等什么？扫描图片上方二维码，名额有限先到先得！

独立消费

我们可以在不定义消费组的情况下进行Stream消息的独立消费，当Stream没有新消息时，甚至可以阻塞等待。Redis设计了一个单独的消费指令xread，可以将Stream当成普通的消息队列(list)来使用。使用xread时，我们可以完全忽略消费组(Consumer Group)的存在，就好比Stream就是一个普通的列表(list)。

# 从Stream头部读取两条消息127.0.0.1:6379> xread count 2 streams codehole 0-01) 1) "codehole"   2) 1) 1) 1527851486781-0         2) 1) "name"            2) "laoqian"            3) "age"            4) "30"      2) 1) 1527851493405-0         2) 1) "name"            2) "yurui"            3) "age"            4) "29"# 从Stream尾部读取一条消息，毫无疑问，这里不会返回任何消息127.0.0.1:6379> xread count 1 streams codehole $(nil)# 从尾部阻塞等待新消息到来，下面的指令会堵住，直到新消息到来127.0.0.1:6379> xread block 0 count 1 streams codehole $# 我们从新打开一个窗口，在这个窗口往Stream里塞消息127.0.0.1:6379> xadd codehole * name youming age 601527852774092-0# 再切换到前面的窗口，我们可以看到阻塞解除了，返回了新的消息内容# 而且还显示了一个等待时间，这里我们等待了93s127.0.0.1:6379> xread block 0 count 1 streams codehole $1) 1) "codehole"   2) 1) 1) 1527852774092-0         2) 1) "name"            2) "youming"            3) "age"            4) "60"(93.11s)

客户端如果想要使用xread进行顺序消费，一定要记住当前消费到哪里了，也就是返回的消息ID。下次继续调用xread时，将上次返回的最后一个消息ID作为参数传递进去，就可以继续消费后续的消息。

block 0表示永远阻塞，直到消息到来，block 1000表示阻塞1s，如果1s内没有任何消息到来，就返回nil

127.0.0.1:6379> xread block 1000 count 1 streams codehole $(nil)(1.07s)

创建消费组

Stream通过xgroup create指令创建消费组(Consumer Group)，需要传递起始消息ID参数用来初始化last_delivered_id变量。

127.0.0.1:6379> xgroup create codehole cg1 0-0  #  表示从头开始消费OK# $表示从尾部开始消费，只接受新消息，当前Stream消息会全部忽略127.0.0.1:6379> xgroup create codehole cg2 $OK127.0.0.1:6379> xinfo codehole  # 获取Stream信息1) length2) (integer) 3  # 共3个消息3) radix-tree-keys4) (integer) 15) radix-tree-nodes6) (integer) 27) groups8) (integer) 2  # 两个消费组9) first-entry  # 第一个消息10) 1) 1527851486781-0    2) 1) "name"       2) "laoqian"       3) "age"       4) "30"11) last-entry  # 最后一个消息12) 1) 1527851498956-0    2) 1) "name"       2) "xiaoqian"       3) "age"       4) "1"127.0.0.1:6379> xinfo groups codehole  # 获取Stream的消费组信息1) 1) name   2) "cg1"   3) consumers   4) (integer) 0  # 该消费组还没有消费者   5) pending   6) (integer) 0  # 该消费组没有正在处理的消息2) 1) name   2) "cg2"   3) consumers  # 该消费组还没有消费者   4) (integer) 0   5) pending   6) (integer) 0  # 该消费组没有正在处理的消息

消费

Stream提供了xreadgroup指令可以进行消费组的组内消费，需要提供消费组名称、消费者名称和起始消息ID。它同xread一样，也可以阻塞等待新消息。读到新消息后，对应的消息ID就会进入消费者的PEL(正在处理的消息)结构里，客户端处理完毕后使用xack指令通知服务器，本条消息已经处理完毕，该消息ID就会从PEL中移除。

# >号表示从当前消费组的last_delivered_id后面开始读# 每当消费者读取一条消息，last_delivered_id变量就会前进127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >1) 1) "codehole"   2) 1) 1) 1527851486781-0         2) 1) "name"            2) "laoqian"            3) "age"            4) "30"127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >1) 1) "codehole"   2) 1) 1) 1527851493405-0         2) 1) "name"            2) "yurui"            3) "age"            4) "29"127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 2 streams codehole >1) 1) "codehole"   2) 1) 1) 1527851498956-0         2) 1) "name"            2) "xiaoqian"            3) "age"            4) "1"      2) 1) 1527852774092-0         2) 1) "name"            2) "youming"            3) "age"            4) "60"# 再继续读取，就没有新消息了127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >(nil)# 那就阻塞等待吧127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole ># 开启另一个窗口，往里塞消息127.0.0.1:6379> xadd codehole * name lanying age 611527854062442-0# 回到前一个窗口，发现阻塞解除，收到新消息了127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >1) 1) "codehole"   2) 1) 1) 1527854062442-0         2) 1) "name"            2) "lanying"            3) "age"            4) "61"(36.54s)127.0.0.1:6379> xinfo groups codehole  # 观察消费组信息1) 1) name   2) "cg1"   3) consumers   4) (integer) 1  # 一个消费者   5) pending   6) (integer) 5  # 共5条正在处理的信息还有没有ack2) 1) name   2) "cg2"   3) consumers   4) (integer) 0  # 消费组cg2没有任何变化，因为前面我们一直在操纵cg1   5) pending   6) (integer) 0# 如果同一个消费组有多个消费者，我们可以通过xinfo consumers指令观察每个消费者的状态127.0.0.1:6379> xinfo consumers codehole cg1  # 目前还有1个消费者1) 1) name   2) "c1"   3) pending   4) (integer) 5  # 共5条待处理消息   5) idle   6) (integer) 418715  # 空闲了多长时间ms没有读取消息了# 接下来我们ack一条消息127.0.0.1:6379> xack codehole cg1 1527851486781-0(integer) 1127.0.0.1:6379> xinfo consumers codehole cg11) 1) name   2) "c1"   3) pending   4) (integer) 4  # 变成了5条   5) idle   6) (integer) 668504# 下面ack所有消息127.0.0.1:6379> xack codehole cg1 1527851493405-0 1527851498956-0 1527852774092-0 1527854062442-0(integer) 4127.0.0.1:6379> xinfo consumers codehole cg11) 1) name   2) "c1"   3) pending   4) (integer) 0  # pel空了   5) idle   6) (integer) 745505

Stream消息太多怎么办

读者很容易想到，要是消息积累太多，Stream的链表岂不是很长，内容会不会爆掉就是个问题了。xdel指令又不会删除消息，它只是给消息做了个标志位。

Redis自然考虑到了这一点，所以它提供了一个定长Stream功能。在xadd的指令提供一个定长长度maxlen，就可以将老的消息干掉，确保最多不超过指定长度。

127.0.0.1:6379> xlen codehole(integer) 5127.0.0.1:6379> xadd codehole maxlen 3 * name xiaorui age 11527855160273-0127.0.0.1:6379> xlen codehole(integer) 3

我们看到Stream的长度被砍掉了。

消息如果忘记ACK会怎样

Stream在每个消费者结构中保存了正在处理中的消息ID列表PEL，如果消费者收到了消息处理完了但是没有回复ack，就会导致PEL列表不断增长，如果有很多消费组的话，那么这个PEL占用的内存就会放大。

PEL如何避免消息丢失

在客户端消费者读取Stream消息时，Redis服务器将消息回复给客户端的过程中，客户端突然断开了连接，消息就丢失了。但是PEL里已经保存了发出去的消息ID。待客户端重新连上之后，可以再次收到PEL中的消息ID列表。不过此时xreadgroup的起始消息ID不能为参数>，而必须是任意有效的消息ID，一般将参数设为0-0，表示读取所有的PEL消息以及自last_delivered_id之后的新消息。

结论

Stream的消费模型借鉴了kafka的消费分组的概念，它弥补了Redis Pub/Sub不能持久化消息的缺陷。但是它又不同于kafka，kafka的消息可以分partition，而Stream不行。如果非要分parition的话，得在客户端做，提供不同的Stream名称，对消息进行hash取模来选择往哪个Stream里塞。如果读者稍微研究过Redis作者的另一个开源项目Disque的话，这极可能是作者意识到Disque项目的活跃程度不够，所以将Disque的内容移植到了Redis里面。这只是本人的猜测，未必是作者的初衷。如果读者有什么不同的想法，可以在评论区一起参与讨论。

转载自：码洞。

投稿：有投稿意向技术人请在公众号对话框留言。

转载：意向文章下方留言。

点击 阅读原文 获得更多精彩。

更多精彩请关注 “数据和云” 公众号

资源下载

关注公众号：数据和云（OraNews）回复关键字获取

2018DTCC , 数据库大会PPT

2017DTC，2017 DTC 大会 PPT

DBALIFE ，“DBA 的一天”海报

DBA04 ，DBA 手记4 电子书

122ARCH ，Oracle 12.2体系结构图

2017OOW ，Oracle OpenWorld 资料

PRELECTION ，大讲堂讲师课程资料

近期文章

仅仅使用AWR做报告? 性能优化还未入门

实战课堂：一则CPU 100%的故障分析

杨廷琨:如何编写高效SQL(含PPT)

一份高达555页的技术PPT会是什么样子？

大象起舞：用PostgreSQL解海盗分金问题

ProxySQL！像C罗一样的强大

高手过招:用SQL解决环环相扣刑侦推理问题