场景
Redis使用list可以实现队列功能，但无法做到广播模式，使用PubSub可以做到广播模式，但是PubSub并不能保证数据的持久化。
Stream就是为了满足上面的需求在Redis 5.0中发布的，它的内部结构是一个链表，将消息都串起来，每个消息拥有一个自己的ID和内容，该结构是持久化保存的。
基本概念
每个Stream都以key作为自己的名字，相当于消息队列中的topic，生产者和消费者对这个key进行发布订阅即可，Stream在第一次使用 xadd 指令时被创建。
消费组
每个Stream可以有多个消费组(Consumer Group)，每个消费组会有个游标(last_delivered_id)在Stream链表上往前移动，表示该消费组消费到了哪条消息。
消费组需要通过 xgroup create 进行创建，需要指定从Stream某个消息ID开始消费，该ID就是游标(last_delivered_id)的初始值。
每个消费者之间是独立的，A消费了数据1，不会影响到B消费数据1，即一份数据可以被多个消费组消费。
消费者
一个消费组可以有多个消费者(Consumer)，这下消费者之间是竞争关系，任意一个消费者消费数据后，都会导致游标(last_delivered_id)向前移动，每个消费者都有一个组内唯一名称。
消费者内部有一个状态变量： pending_ids，该变量记录已经被客户端读取，但是没有ACK的消息，客户端没有ACK的消息越多，这里保存的消息id就越多，ACK后将相应减少，
官方称 pending_ids 为 PEL(Pending Entires List)，用来确保客户端至少消费了一次，防止数据丢失。
消息ID与内容
消息ID格式为：timestamplnMillis-sequence，前者是消息产生的毫秒时间戳，后者是在该毫秒产生的第几条消息，例如 1527846880572-5 ，代表是 1527846880572 毫秒时的第 5 条消息，
消息ID可以由服务器自动生成，也可以是客户端自己指定，但是形式必须是 整数-整数，后来的消息ID必须大于前面的消息ID。
消息内容是一个类似hash结构的键值对。
基本命令
- xadd：向Stream发一条消息
# stream-1 表示stream的名字，* 则是系统该自动生成id，后面的则是我们的数据键值对
127.0.0.1:6379> xadd stream-1 * name zhangsan age 10
"1641454184593-0"
127.0.0.1:6379> xadd stream-1 * name lisi age 10
"1641454195771-0"
127.0.0.1:6379> xadd stream-1 * name wangwu age 10
"1641454208016-0"
- xdel：从Stream中删除一条消息
127.0.0.1:6379> xdel stream-1 1641454195771-0
(integer) 0- xrange：获取Stream中的消息列表
127.0.0.1:6379> xrange stream-1 - +
1) 1) "1641454639427-0"
2) 1) "name"
2) "zhangsan"
3) "age"
4) "10"
2) 1) "1641454646977-0"
2) 1) "name"
2) "lisi"
3) "age"
4) "10"
3) 1) "1641454651750-0"
2) 1) "name"
2) "wangwu"
3) "age"
4) "10"- xlen：获取Stream消息长度
127.0.0.1:6379> xlen stream-1
(integer) 3- del：删除整个Stream消息列表
127.0.0.1:6379> del stream-1
(integer) 1
127.0.0.1:6379> xrange stream-1 - +
(empty array)消费模式
消费模式有两种，独立消费与消费组，前者就是一个消息只能被全局消费一次，后者则是多个消费组可以对同一个消费者消费一次。
独立消费
Stream 提供了 一个 xread 指令，该指令在消费消息的时候会忽略消费组的存在。
- 使用 xread 从头部读取两条消息
127.0.0.1:6379> xread count 2 streams stream-1 0-0
1) 1) "stream-1"
2) 1) 1) "1641455116178-0"
2) 1) "name"
2) "zhangsan"
3) "age"
4) "10"
2) 1) "1641455118418-0"
2) 1) "name"
2) "zhangsan1"
3) "age"
4) "10"- 从尾部接收最新消息，之前的消息全部会被忽略
这种接收最新消息一般配合阻塞使用，如果不使用阻塞大概率读的是空
127.0.0.1:6379> xread count 1 streams stream-1 $
(nil)阻塞读取，当有xadd往该stream-1添加消息时，将立刻返回，block以s为单位，阻塞s表后没有数据将返回，block 0 就是永远阻塞，直到有消息来临。
127.0.0.1:6379> xread block 0 count 1 streams stream-1 $
1) 1) "stream-1"
2) 1) 1) "1641455579648-0"
2) 1) "name"
2) "wangliu"
3) "age"
4) "20"
(38.07s)消费组
通过 xgroup create 可以创建一个消费组，并提供一个消息起始id初始化 last_delivered_id 变量。
- 为 stream-1 添加一个名为 cgroup-1 的消费组，从第一条消息开始消费。
xgroup create stream-1 cgroup-1 0-0- 为 stream-1 添加一个名为 cgroup-2 的消费组，该消费组只监听新消息
xgroup create stream-1 cgroup-2 $- 查看 stream-1 的详细信息
127.0.0.1:6379> xinfo stream stream-1
1) "length"
2) (integer) 4 # 共4个消息
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "last-generated-id"
8) "1641455579648-0"
9) "groups"
10) (integer) 2 # 两个消费组
11) "first-entry" # 第一个消息
12) 1) "1641455116178-0"
2) 1) "name"
2) "zhangsan"
3) "age"
4) "10"
13) "last-entry" # 最后一个消息
14) 1) "1641455579648-0"
2) 1) "name"
2) "wangliu"
3) "age"
4) "20"
消费
xreadgroup 用于消费组进行消费消息，需要提供消费组名称、消费者名称、起始消息ID，与 xread 一样，也可以进行阻塞等待新消息，
读到新消息时，该消息ID将存入消费者的PEL中，处理完毕后客户端使用 xack 指令通知服务器，该消息ID将被从PEL中移除。
- 消费消息
使用消费组 cgroup-1 中的 c1 消费者读取 stream-1 中的 一条消息，> 表示从当前消费组的 last_delivered_id 后面开始读，没读一条消息，该值就会往前进一个。
127.0.0.1:6379> xreadgroup GROUP cgroup-1 c1 count 1 streams stream-1 >
1) 1) "stream-1"
2) 1) 1) "1641455579648-0"
2) 1) "name"
2) "wangliu"
3) "age"
4) "20"
127.0.0.1:6379> xreadgroup GROUP cgroup-1 c1 count 1 streams stream-1 >
(nil) 读完了
- 阻塞等待读取
xreadgroup GROUP cgroup-1 c1 block 0 count 1 streams stream-1 >用另一个客户端发送消息
127.0.0.1:6379> xadd stream-1 * name mn age 30
"1641457442683-0"
读到新消息返回
127.0.0.1:6379> xreadgroup GROUP cgroup-1 c1 block 0 count 1 streams stream-1 >
1) 1) "stream-1"
2) 1) 1) "1641457442683-0"
2) 1) "name"
2) "mn"
3) "age"
4) "30"
(45.20s)- 观察消费组信息
127.0.0.1:6379> xinfo groups stream-1
1) 1) "name"
2) "cgroup-1"
3) "consumers"
4) (integer) 1 # 一个消费者
5) "pending"
6) (integer) 5 # 5 条消息还没有ack
7) "last-delivered-id"
8) "1641457442683-0"
2) 1) "name"
2) "cgroup-2"
3) "consumers"
4) (integer) 0
5) "pending"
6) (integer) 0
7) "last-delivered-id"
8) "1641455579648-0"- 观察消费者信息
# cgroup-1下的消费者
127.0.0.1:6379> xinfo consumers stream-1 cgroup-1
1) 1) "name"
2) "c1" # c1 消费者
3) "pending"
4) (integer) 5 # 5条消息还没有ack
5) "idle"
6) (integer) 504275 # 空闲了多久 ms 没有读取消息
- ack消息
# ack一条消息
xack stream-1 cgroup-1 1641455116178-0# ack多条消息
127.0.0.1:6379> xack stream-1 cgroup-1 1641455118418-0 1641455120461-0 1641455579648-0 1641457442683-0
(integer) 4
5条消息全部ack后，待ack的数量变为0了。
127.0.0.1:6379> xinfo consumers stream-1 cgroup-1
1) 1) "name"
2) "c1"
3) "pending"
4) (integer) 0
5) "idle"
6) (integer) 684848
其他场景
消息定长
如果需要将Stream中的消息保持一定的长度，可以在xadd时，增加一个maxlen参数，该参数可以把老的消息剔除，保证消息长度不超过maxlen。
先增加5条数据
127.0.0.1:6379> xadd stream-1 * name n1
"1641458449368-0"
127.0.0.1:6379> xadd stream-1 * name n2
"1641458451353-0"
127.0.0.1:6379> xadd stream-1 * name n3
"1641458452737-0"
127.0.0.1:6379> xadd stream-1 * name n4
"1641458454740-0"
127.0.0.1:6379> xadd stream-1 * name n5
"1641458456357-0"
使用定长参数，设置长度为3，可以看到，当使用定长参数后，链表中的消息只有3个了。
127.0.0.1:6379> xadd stream-1 maxlen 3 * name n5
"1641458535493-0"
127.0.0.1:6379> xinfo stream stream-1
1) "length"
2) (integer) 3
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "last-generated-id"
8) "1641458535493-0"
9) "groups"
10) (integer) 2
11) "first-entry"
12) 1) "1641458454740-0"
2) 1) "name"
2) "n4"
13) "last-entry"
14) 1) "1641458535493-0"
2) 1) "name"
2) "n5"
忘记ACK
消费者的PEL保存了没有ACK的消息ID，如果没有ACK的消息越来越多，那么这个结构就会占用空间越来越大，所以千万不要忘记ACK。
PEL防止消息丢失
PEL主要的作用是，如果客户端消费了一条消息，此时发生网络断开等原因，这个消息还没被客户端消费就丢失了，因此PEL可以保证，在没有ACK前，一直在PEL中保存该消息ID，等客户端重新连接上后，可以再次消费PEL中的消息ID，一般建议客户端将参数设置成 0-0，表示从laster_delivered_id后开始消费。
Stream高可用
Stream与其他数据结构并没有区别，它的高可用也是通过建立在主从复制上，在 Cluster 和 Sentinel 即可支持高可用，由于Redis指令复制是异步的，因此可能会发生丢失极小部分的数据丢失。
分区Partition
Stream没有提供分区的能力，如果需要实现的话，可以建立多个Stream结构，然后客户端通过对key进行hash的方式，路由到不同的Stream分区中。
