耳听八方——Stream
数据结构Stream,一个新的强大的支持多播的可持久化消息队列。
它是一个消息链表,将所有加入的消息都串起来,每个消息都有一个唯一的ID和对应的内容。消息是持久化的,Redis重启后,内容还在。
每个Stream都有唯一的名称,它就是Redis的key,在我们首次使用xadd指令追加消息时自动创建。
每个Stream都可以挂多个消费组,每个消费组会有个游标last_delivered_id在Stream数组之上往前移动,表示当前消费组已经消费到哪条消息了。每个消费组都有一个Stream内唯一的名称,消费组不会自动创建,需要单独的指令xgroup create 进行创建,需要指定从Stream的某个消息ID开始消费,这个ID用来初始化last_deliverd_id变量。
每个消费组的状态都是独立的,相互不收影响。同一份Stream内部消息会被每个消费组都消费到。
同一个消费组可以挂接多个消费者,这些消费者之间是竞争关系,任意一个消费者读取了消息都会使游标last_delivered_id往前移动。每个消费者有一个组内唯一名称。
消费者内部会有一个状态变量pending_ids,它记录了当前已经被客户端读取,但是还没有ack的消息。如果客户端没有ack,这个变量里面的消息ID就会越来越多,一旦某个消息被ack,它就开始减少。这个pending_ids变量在Redis被称为PEL(Pending Entries List),这是一个核心的数据结构,用来确保客户端至少消费消息一次,而不会在网络传输的中途丢失而没被处理。
消息ID
消息ID的形式是timestampINMillis-sequence,例如1527846880572-5,它表示当前的消息在毫米时间戳时产生,是该毫秒内产生的第五秒消息。消息ID可以由服务器自动生成,也可以由客户端自己指定,但是形式必须是整数-整数,而且后面加入的消息的ID必须要大于前面的消息ID
消息内容
消息内容就是键值对
增删改查
增删改查等指令说明:
- xadd:向Stream追加消息
- xdel:从Stream中删除消息,这里的删除仅仅是设置标志位,不影响消息总长度
- xrange:获取Stream中的消息列表,会自动过滤已经删除的消息
- xlen:获取Stream消息长度
- del:删除整个Stream消息列表的所有消息
xadd codehole * name laoqian age 30
xadd codehole * name xiaoxu age 29
xlen codehole
xrange codehole - +
xdel codehole 1622789513273-0
del codehole
独立消费
可以在不定义消费组的情况下进行Stream消息的独立,当Stream没有新消息时,可以阻塞等待。Redis设计了一个单独的消费指令xread,可以将Stream当做普通的消息队列来使用。使用xread时,我们可以完全忽略消费组的存在,就好像Stream是一个普通的列表一样。
创建消费组
Stream通过xgroup create指令创建消费组,创建消费组需要提供起始消息ID参数用来初始化last_delivered_id变量
xgroup create codehole cg1 0-0 # 表示从头部开始消费
xgroup create codehole cg2 $ # 表示从尾部开始消费,只接受新消息,当前stream消息会全部忽略
xinfo stream codehole # 获取Stream消息
xinfo groups codehole # 获取Stream的消费组信息
消费
Stream提供了xreadgroup指令可以进行消费组的组内消费,需要提供消费组名称、消费者名称和起始消息ID。它同xread一样,也可以阻塞等待新消息。读到新消息后,对应的消息ID就会进入消费者的PEL结构里,客户端处理完毕后使用xack指令通知服务器,本条消息已经处理完毕,该消息ID就会从PEL中移除。
xreadgroup GROUP cg1 c1 count 1 streams codehole # 每当消费者读取一条消息,last_delivered_id变量就会前进
Stream消息太多怎么办
消息积累太多,Stream链表过长,xdel不会删除消息,只给消息做个标志位。
Redis提供了一个定长Stream功能,在xadd的指令中提供一个定长长度参数maxlen,就可以将老的消息干掉,确保链表不超过指定长度
消息如果忘记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的高可用建立在主从复制基础上,和其他数据结构的复制机制没有区别,也就是说在Sentienl和Cluster集群环境下,Stream支持高可用。不过鉴于Redis的指令复制是异步的,在failover发生时,Redis可能会丢失极少部分数据,这一点Redis的其他数据结构也一样
分区Partition
Redis的服务器没有原生支持分区能力,如果想要使用分区,那就需要分配多个Stream,然后在客户端使用一定的策略来生产消息到不同的Stream。