遂宁网站制作,asp.net旅游网站管理系统代码,济宁seo推广,wordpress管理员登入文章目录 背景stream概述消息 ID消息内容常见操作独立消费创建消费组消费 Stream弊端Stream 消息太多怎么办?消息如果忘记 ACK 会怎样?PEL 如何避免消息丢失?分区 Partition Stream 的高可用总结 背景
为了解决list作为消息队列是无法支持消息多播问题#xff0c;Redis5.0… 文章目录 背景stream概述消息 ID消息内容常见操作独立消费创建消费组消费 Stream弊端Stream 消息太多怎么办?消息如果忘记 ACK 会怎样?PEL 如何避免消息丢失?分区 Partition Stream 的高可用总结 背景
为了解决list作为消息队列是无法支持消息多播问题Redis5.0 多出了一个数据结构 Stream它是一个新的强大的支持多播的可持久化的消息队列作者坦言 Redis Stream 狠狠地借鉴了 Kafka 的设计。
stream概述 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
独立消费
我们可以在不定义消费组的情况下进行 Stream 消息的独立消费当 Stream 没有新消 息时甚至可以阻塞等待。 Redis 设计了一个单独的消费指令 xread可以将 Stream 当成普通的消息队列 (list) 来使用。使用 xread 时我们可以完全忽略消费组 (Consumer Group)的存在就好比 Stream 就是一个普通的列表 (list)。
客户端如果想要使用 xread 进行顺序消费一定要记住当前消费到哪里了也就是返回的消息 ID。下次继续调用 xread 时将上次返回的最后一个消息 ID 作为参数传递进去就可以继续消费后续的消息。block 0 表示永远阻塞直到消息到来 block 1000 表示阻塞 1s如果 1s 内没有任何消息到来就返回 nil。类似kafka广播消费客户端保存offset。
创建消费组 Stream 通过 xgroup create 指令创建消费组 (Consumer Group)需要传递起始消息 ID 参数用来初始化 last_delivered_id 变量。其实就是kafka广播消费下客户端自己保存offset消费者在消费前传递offset告知kafka从哪开始消费果然是借鉴kafka但是没做到kafka的层度kafka客户端会自己保存在内存中不需要使用者自己保存。
消费
Stream 提供了 xreadgroup 指令可以进行消费组的组内消费需要提供消费组名称、消费者名称和起始消息 ID。它同 xread 一样也可以阻塞等待新消息。读到新消息后对应的消息 ID 就会进入消费者的 PEL(正在处理的消息) 结构里客户端处理完毕后使用 xack指令通知服务器本条消息已经处理完毕该消息 ID 就会从 PEL 中移除。
Stream弊端
Stream 消息太多怎么办?
读者很容易想到要是消息积累太多 Stream 的链表岂不是很长内容会不会爆掉?xdel指令又不会删除消息它只是给消息做了个标志位。
Redis 自然考虑到了这一点所以它提供了一个定长 Stream 功能。在 xadd 的指令提供一个定长长度 maxlen就可以将老的消息干掉确保最多不超过指定长度。
这个类似kafka的日志删除只是这里固定为根据大小删除当达到一定量就删除旧数据。
消息如果忘记 ACK 会怎样?
Stream 在每个消费者结构中保存了正在处理中的消息 ID 列表 PEL如果消费者收到 了消息处理完了但是没有回复 ack就会导致 PEL 列表不断增长如果有很多消费组的话那么这个 PEL 占用的内存就会放大。
PEL 如何避免消息丢失?
在客户端消费者读取 Stream 消息时 Redis 服务器将消息回复给客户端的过程中客户端突然断开了连接消息就丢失了。但是 PEL 里已经保存了发出去的消息 ID。待客户端重新连上之后可以再次收到 PEL 中的消息 ID 列表。不过此时 xreadgroup 的起始消息ID 不能比PEL内的大而必须是任意有效的消息 ID一般将参数设为 0-0表示读取所有的PEL 消息以及自 last_delivered_id 之后的新消息。这其实就像kafka提供seek方法用于客户端指定从哪个offset开始消费。
分区 Partition
Redis 的服务器没有原生支持分区能力如果想要使用分区那就需要分配多个Stream然后在客户端使用一定的策略来生产消息到不同的 Stream。所以个人认为还是kafka强大stream只是个对kafka进行大量阉割的消息队列使用上请谨慎。
Stream 的高可用
Stream 的高可用是建立主从复制基础上的它和其它数据结构的复制机制没有区别也就是说在 Sentinel 和 Cluster 集群环境下 Stream 是可以支持高可用的。不过鉴于 Redis 的指令复制是异步的在 failover 发生时 Redis 可能会丢失极小部分数据这点 Redis 的其它数据结构也是一样的。
总结
如果真需要使用消息队列那么还是选择市面上大家认可的消息队列以防在后期迭代时发现大量功能问题和性能问题。
