22拓展1耳听八方stream天下无鱼

这篇⽂章的主要部分是我在这篇⽂章的主要部分是我在2018年6月1日发表的内容，当时Redis5.0刚刚被Antirez放出来，⽼钱抢先分析了Redis5.0最重要的特性Stream，也被技术圈Redis5.0被作者Antirez突然放了出来，增加了很多新的特色功能。而Redis5.0最大的新特性就是多出了⼀个数据结构Stream，RedisStream狠狠地借鉴了Kafka的设计。RedisStream的结构如上图所⽰，它有⼀个消息链表，将所有加⼊的消息都串起来，每个消息都有⼀个唯⼀的ID和对应的内容。消息是持久化的，Redis重启后，内容还在。每个Stream都有唯⼀的名称，它就是Redis的key，在我们首次每个Stream标last_delivered_id在Stream数组之上往前移动，表⽰当前Stream内唯⼀的名称，消费组不会自动创建，它需要单独的指令xgroupcreate进行创建，需要指定从Stream的某个消息ID开始消费，这个ID每个消费组(ConsumerGroup)响。也就是说同⼀份Stream同⼀个消费组(ConsumerGroup)可以挂接多个消费者(Consumer)，(Consumer内部会有个状态变量pending_ids，它记录了当前已经被客户端读取的消息，但是还没有ack。如果客户端没有ack，这个变量里面的消息ID会越来越多，⼀旦某个消息被ack，它就开始减少。这个pending_ids变量在Redis官方被称之为PEL，也就是PendingEntriesList，这是⼀个很核⼼的数据消息ID的形式是timestampInMillis-sequence戳15278468805725条消息。消息ID可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加⼊的消息的ID要大于前面的消息ID。hash结构的键值对，这没什么特别之del删除*号表⽰服务器自动生成ID，后面顺序跟着⼀堆key/value#名字叫laoqian，年龄30岁xaddcodeholelaoqian生成的消息xaddcodeholexiaoyuxaddcodeholexiaoqian1127.0.0.1:6379>xlen(integer)表⽰最小值:6379>xrangecodehole-:6379>xrangecodehole-1)1)1527849609889-1)2)1)1527849629172-2)1)3)1)1527849637634-2)1)指定最小消息ID:6379>xrangecodehole1527849629172-01)1)1527849629172-1)2)1)1527849637634-2)1)指定最大消息ID:6379>xrangecodehole-1527849629172-1)1)1527849609889-1)2)1)1527849629172-2)1)127.0.0.1:6379>xdelcodehole1527849609889-(integer)1:6379>xlen(integer):6379>xrangecodehole-1)1)1527849629172-1)2)1)1527849637634-2)1)#删除整个Stream127.0.0.1:6379delcodehole(integer)我们可以在不定义消费组的情况下进行Stream消息的独⽴消费，当Stream没有新消息时，甚⾄可以阻塞等待。Redis设计了⼀个单独的消费指令xread，可以将Stream当成普通的消息队列(list)来使用。使用xread时，我们可以完全忽略消费组(ConsumerGroup)的存在，就好⽐Stream就是⼀个普通的列表(list)。从Stream127.0.0.1:6379>xreadcount2streamscodehole01)2)1)1)1527851486781-1)2)1)1527851493405-2)1)从Stream尾部读取⼀条消息，毫⽆疑问，这里不会返回任何127.0.0.1:6379>xreadcount1streams$从尾部阻塞等待新消息到来，下面的指令会堵住，直到新消息到:6379>xreadblock0count1codehole我们从新打开⼀个窗⼝，在这个窗⼝往Stream:6379>xaddcodehole*nameyouming1)客户端如果想要使用xread进行顺序消费，⼀定要记住当前消费到ID。下次继续调用xread时，将上次返回的最后⼀个消息ID作为参数传递进去，就可以继续消费后续的消block0表⽰永远阻塞，直到消息到来，block10001s，如果1s内没有任何消息到来，就返回nil127.0.0.1127.0.0.1:6379>xreadblock1000count1streamscodehole$Stream通过xgroupcreate(ConsumerGroup)，需要传递起始消息ID参数用来初始#127.0.0.1:6379>xgroupcreatecodeholecg10-0$表⽰从尾部开始消费，只接受新消息，当前Stream:6379>xgroupcreatecodehole$获取Stream:6379>xinfostream(integer3共3(integer)(integer)(integer2first-entry10)1)1527851486781-2)1)last-entry1)1527851498956-1)获取Stream127.0.0.1:6379>xinfogroups1)002)#00Stream提供了xreadgroup指令可以进行消费组的组内消费，需。它同也可以阻塞等待新消息。读到新消息后，对应的消息ID就会进⼊消费者的PEL(正在处理的消息)结构里，客户端处理完毕后使用xack指令通知服务器，本条消息已经处理完毕，该消息ID就会从PEL中#>号表⽰从当前消费组的last_delivered_id后面开始读每当消费者读取⼀条消息，last_delivered_id变量就会前127.0.0.1:6379>xreadgroupGROUPcg1c1count1streamscodehole>1)2)1)1)1527851486781-1)127.0.0.1:6379>xreadgroupGROUPcg1c1count1streamscodehole>1)2)1)1)1527851493405-1)127.0.0.1:6379>xreadgroupGROUPcg1c1count2streamscodehole>1)2)1)1)1527851498956-1)2)1)1527852774092-2)1)127.0.0.1:6379>xreadgroupGROUPcg1c1countstreamscodehole>127.0.0.1:6379>xreadgroupGROUPcg1c1blockcount1streamscodehole#开启另⼀个窗⼝，往里塞消息127.0.0.1:6379>xaddcodehole*namelanying127.0.0.1:6379>xreadgroupGROUPcg1c1blockcount1streamscodehole1)2)1)1)1527854062442-1):6379>xinfogroups1)15共5条正在处理的信息还有没有2)(integer0消费组cg2没有任何变化，因为前面我们⼀直在操纵cg1(integer)如果同⼀个消费组有多个消费者，我们可以通过xinfoconsumers指令观察每个消费者的状态127.0.0.1:6379xinfoconsumerscodeholecg1目前还有1个消费者1)556)(integer418715空闲了多⻓时间ms没有读取接下来我们ack127.0.0.1:6379>xackcodeholecg11527851486781-(integer)127.0.0.1:6379>xinfoconsumerscodehole1)4变成了5(integer)ack127.0.0.1:6379>xackcodeholecg11527851493405-(integer)127.0.0.1:6379>xinfoconsumerscodehole1)1)(integer0pel(integer)Stream消息太多怎么办读者很容易想到，要是消息积累太多，Stream的链表岂不是很⻓，内容会不会爆掉?xdel指令⼜不会删除消息，它只是给消息做了个RedisStream功能。在xadd的指令提供⼀个定⻓⻓度maxlen，就可以将⽼的消息干(integer)codeholemaxlen3*xiaoruiage(integer)StreamStream在未来可以提供按消息如果忘记ACK会怎样Stream在每个消费者结构中保存了正在处理中的消息ID列表PEL，如果消费者收到了消息处理完了但是没有回复ack，就会导致PELPEL占用的内存PEL如何避免消息丢失在客户端消费者读取Stream消息时，Redis服务器将消息回复给客户端的过程中，客户端突然断开了连接，消息就丢失了。但是PEL里已经保存了发出去的消息ID。待客户端重新连上之后，可以再次收到PEL中的消息ID列表。不过此时xreadgroup的起始消息ID不能为参数>，而必须是任意有效的消息ID，⼀般将参数设为0-0，表⽰读取所有的PEL消息以及自last_delivered_id之后的新消StreamStream的高可用是建⽴主从复制基础上的，它和其它数据结构的复制机制没有区别，也就是说在Sentinel和Cluster集群环境下Stream是可以⽀持高可用的。不过鉴于Redis的指令复制是异步的，在failover发生时，Redis可能会丢失极小部分数据，这点Redis的其它数据结构也是⼀样的。Redis的服务器没有原生⽀持分区能力，如果想要使用分区，那就需要分配多个Stream，然后在客户端使用⼀定的策略来生产消息到不同的Stream。你也许会认为Kafka要先进很多，它是原生⽀持Partition的。关于这⼀点，我并不认同。记得Kafka的客户端也存在HashStrategy么，因为它也是通过客户端的hash算法来将不同另外,Kafka还⽀持动态增加分区数量的能力，但是这种调整能力也是很蹩脚的，它不会把之前已经存在的内容进行rehash，不会重新分区历史数据。这种简单的动态调整的能力RedisStream通过增加新的Stream就可以做到。Stream的