京公网安备11010502063487号stream 队列不是“队列”,而是一个追加日志:消息读完还在,消费者用 offset 决定从哪读。它能做什么、不能做什么、保留策略怎么删?这篇用普通 AMQP 客户端在 RabbitMQ 4.3.6 上把这些都过一遍。
消费者 A(offset=first)读到并 ack: ['m0', 'm1', 'm2', 'm3', 'm4', 'm5', 'm6', 'm7', 'm8', 'm9'];带的 offset 头: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] 消费者 B 随后也从头读: ['m0', 'm1', 'm2', 'm3', 'm4', 'm5', 'm6', 'm7', 'm8', 'm9'](消息还在) rabbitmq-streams stream_status s1: offset=9, role=writer, epoch=1, committed_offset=9, first_offset=0, readers=0, segments=1
stream 队列把消息追加在日志里,每条有一个递增的 offset(从 0 开始),投递时放在 x-stream-offset 头里。ack 不删除任何东西,同一批消息可以被很多消费者重复读取。
offset=first → 收到 30 条,第一条 m0(offset 0),最后一条 m29 offset=last → 收到 1 条,第一条 m29(offset 29),最后一条 m29 offset=next → 收到 0 条,第一条 -(offset -),最后一条 - offset=数字 20 → 收到 10 条,第一条 m20(offset 20),最后一条 m29 offset=时间戳(第 15 条之后) → 收到 15 条,第一条 m15(offset 15),最后一条 m29 offset=不设 → 收到 0 条,第一条 -(offset -),最后一条 - 不设 offset 时的默认行为是 next(只收订阅之后新到的),所以 0 条
消费时用参数 x-stream-offset 指定:first、last(最后一个 chunk)、next(只要新的,也是不设时的默认)、一个数字 offset、或一个时间戳。
先读 5 条(最后一条 offset=4),下次用 offset=5 继续,读到 15 条: 从 m5 到 m19
AMQP 0-9-1 没有消费者位置的存储,消费者要自己记住处理到的 offset,下次用上次 + 1 继续。
basic_get: ConnectionClosedByBroker: (540, "NOT_IMPLEMENTED - basic.get not supported by stream queues queue 's1' in vhost '/'") 不设 prefetch 消费: ChannelClosedByBroker: (406, "PRECONDITION_FAILED - consumer prefetch count is not set for stream queue 's1' in vhost '/'") durable=False: ConnectionClosedByBroker: (541, 'INTERNAL_ERROR - Feature `transient_nonexcl_queues` is deprecated.\nBy default, this feature is not permitted anymore.\nThe feature will be removed from a future m exclusive=True: ChannelClosedByBroker: (406, "PRECONDITION_FAILED - invalid property 'exclusive-owner' for queue 'sx' in vhost '/'") auto_ack 消费: ConnectionClosedByBroker: (540, "NOT_IMPLEMENTED - automatic acknowledgement not supported by stream queue 's1' in vhost '/'")
basic_get、自动确认、非持久、exclusive 队列、不设 prefetch 的消费,都被 broker 拒绝。
发了 60000 条;第一条还能读到的 offset = 44779;stream_status: first_offset=44778 committed_offset=60012 segments=4
数据目录里的 .segment 文件: [('00000000000000000000.segment', 0), ('00000000000000000000.segment', 0), ('00000000000000000000.segment', 0), ('00000000000000044778.segment', 4), ('00000000000000049258.segment', 4), ('00000000000000053729.segment', 4), ('00000000000000058196.segment', 1)](名字是这个段第一条消息的 offset,第二项是 MiB);.index 文件 7 个,每段一个用 x-max-length-bytes 或按时间的 x-max-age 限制,段大小由 x-stream-max-segment-size-bytes 决定;超出限制时最老的段整段删除。
stream 读 100000 条: 924 ms(可重复读 3 次,每次都是同样的 100000 条) classic 读 100000 条: 903 ms(读完就没了,只能读 1 次)
用 AMQP 读 100000 条的耗时与 classic 队列相当(924 对 903 ms);stream 的价值在语义。