前言
Redis5.0增加了很多新的特色功能。而Redis5.0最大的新特性就是多出了一个数据结构Stream，它是一个新的强大的支持多播的可持久化的消息队列，RedisStream狠狠地借鉴了Kafka的设计。
今天，咱们就来深度剖析Redis的拓展应用——Stream
基本介绍
RedisStream的结构如上图所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的ID和对应的内容。消息是持久化的，Redis重启后，内容还在。
每个Stream都有唯一的名称，它就是Redis的key，在我们首次使用xadd指令追加消息时自动创建。
每个Stream都可以挂多个消费组，每个消费组会有个游标last_delivered_id在Stream数组之上往前移动，表示当前消费组已经消费到哪条消息了。每个消费组都有一个Stream内唯一的名称，消费组不会自动创建，它需要单独的指令xgroupcreate进行创建，需要指定从Stream的某个消息ID开始消费，这个ID用来初始化last_delivered_id变量。
每个消费组(ConsumerGroup)的状态都是独立的，相互不受影响。也就是说同一份Stream内部的消息会被每个消费组都消费到。
同一个消费组(ConsumerGroup)可以挂接多个消费者(Consumer)，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标last_delivered_id往前移动。每个消费者有一个组内唯一名称。
消费者(Consumer)内部会有个状态变量pending_ids，它记录了当前已经被客户端读取的消息，但是还没有ack。如果客户端没有ack，这个变量里面的消息ID会越来越多，一旦某个消息被ack，它就开始减少。这个pending_ids变量在Redis官方被称之为PEL，也就是PendingEntriesList，这是一个很核心的数据结构，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了没处理。
消息ID
消息ID的形式是timestampInMillis-sequence，例如2-5，它表示当前的消息在毫米时间戳2时产生，并且是该毫秒内产生的第5条消息。消息ID可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加入的消息的ID要大于前面的消息ID。
消息内容
消息内容就是键值对，形如hash结构的键值对，这没什么特别之处。
增删改查
1、xadd追加消息2、xdel删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度3、xrange获取消息列表，会自动过滤已经删除的消息4、xlen消息长度5、del删除Stream
独立消费
我们可以在不定义消费组的情况下进行Stream消息的独立消费，当Stream没有新消息时，甚至可以阻塞等待。Redis设计了一个单独的消费指令xread，可以将Stream当成普通的消息队列(list)来使用。使用xread时，我们可以完全忽略消费组(ConsumerGroup)的存在，就好比Stream就是一个普通的列表(list)。
客户端如果想要使用xread进行顺序消费，一定要记住当前消费到哪里了，也就是返回的消息ID。下次继续调用xread时，将上次返回的最后一个消息ID作为参数传递进去，就可以继续消费后续的消息。
block0表示永远阻塞，直到消息到来，block表示阻塞1s，如果1s内没有任何消息到来，就返回nil。
创建消费组
Stream通过xgroupcreate指令创建消费组(ConsumerGroup)，需要传递起始消息ID参数用来初始化last_delivered_id变量。
消费
Stream提供了xreadgroup指令可以进行消费组的组内消费，需要提供消费组名称、消费者名称和起始消息ID。它同xread一样，也可以阻塞等待新消息。读到新消息后，对应的消息ID就会进入消费者的PEL(正在处理的消息)结构里，客户端处理完毕后使用xack指令通知服务器，本条消息已经处理完毕，该消息ID就会从PEL中移除。
Stream消息太多怎么办?
读者很容易想到，要是消息积累太多，Stream的链表岂不是很长，内容会不会爆掉?xdel指令又不会删除消息，它只是给消息做了个标志位。
Redis自然考虑到了这一点，所以它提供了一个定长Stream功能。在xadd的指令提供一个定长长度maxlen，就可以将老的消息干掉，确保最多不超过指定长度。
我们看到Stream的长度被砍掉了。如果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的高可用
Stream的高可用是建立主从复制基础上的，它和其它数据结构的复制机制没有区别，也就是说在Sentinel和Cluster集群环境下Stream是可以支持高可用的。不过鉴于Redis的指令复制是异步的，在failover发生时，Redis可能会丢失极小部分数据，这点Redis的其它数据结构也是一样的。
分区Partition
Redis的服务器没有原生支持分区能力，如果想要使用分区，那就需要分配多个Stream，然后在客户端使用一定的策略来生产消息到不同的Stream。你也许会认为Kafka要先进很多，它是原生支持Partition的。关于这一点，我并不认同。记得Kafka的客户端也存在HashStrategy么，因为它也是通过客户端的hash算法来将不同的消息塞入不同分区的。另外,Kafka还支持动态增加分区数量的能力，但是这种调整能力也是很蹩脚的，它不会把之前已经存在的内容进行rehash，不会重新分区历史数据。这种简单的动态调整的能力RedisStream通过增加新的Stream就可以做到。
后记
Stream的消费模型借鉴了Kafka的消费分组的概念，它弥补了RedisPub/Sub不能持久化消息的缺陷。但是它又不同于kafka，Kafka的消息可以分partition，而Stream不行。如果非要分parition的话，得在客户端做，提供不同的Stream名称，对消息进行hash取模来选择往哪个Stream里塞。
以上就是小编整理的Redis拓展应用之Stream详解，只是小编自己的看法，有哪里不对的地方，请各位大佬指出，咱们共同进步。
喜欢小编请多多点赞评论分享，
