Redis Streams——有确认、能回溯的消息队列

还记得 List 做队列吗BRPOP一把把消息拿走消费者这时候挂了这条就没了。Streams 不这么干。本次导航Stream 长什么样每条消息一个 ID后面跟一堆字段读写XADD、XRANGE、XREAD消费者组XREADGROUPXACK挂了还能捞回来Redis 8.6 的幂等投递IDMPAUTO/IDMP和 Kafka 比什么时候用 Redis 就够了发车前提醒Redis 核心数据结构二——List 与消息队列 这篇文章已经把 List 当队列的坑写过了这篇接着往下做。先用 docker 启动容器还是之前的容器dockerexec-itredis-demo redis-cli一、Stream 是一条只往后追加的日志别把它想成 ListList 的元素被POP之后就从 Key 里消失了Stream 更像一本流水账新消息往后面贴旧的还在除非你主动裁。每条消息两个部分ID默认是毫秒时间戳-序号比如1789086136869-0。你那边数字肯定不一样规则一样。字段就是一堆 field-value跟 Hash 挺像。一条订单可以带user_id、amount、sku。127.0.0.1:6379XADD orders * user_id1001amount299sku sku_10011789086136869-0127.0.0.1:6379XADD orders * user_id1002amount88sku sku_20021789086136938-0127.0.0.1:6379XLEN orders(integer)2*的意思是ID 让 Redis 自己生成。一般别手写 ID时钟回拨、多实例抢号都是麻烦。二、先当普通日志读按范围翻127.0.0.1:6379XRANGE orders - 1)1)1789086136869-02)1)user_id2)10013)amount4)2995)sku6)sku_10012)1)1789086136938-02)1)user_id2)10023)amount4)885)sku6)sku_2002-是最小是最大。只看某一段就写成XRANGE orders 1789086136869-0 1789086136869-0。从某个 ID 之后接着读用XREAD。0-0表示从最早开始$表示只看比现在更新的127.0.0.1:6379XREAD COUNT1STREAMS orders0-01)1)orders2)1)1)1789086136869-02)1)user_id2)1001...想卡着等新消息加上BLOCK单位毫秒0表示一直等。空流的时候它会挂起有人XADD才醒这点和BRPOP类似XREAD BLOCK10000STREAMS orders $Stream 可以在写入时裁一刀XADD orders MAXLEN1000* user_id1003amount50sku sku_3003大概只留最近 1000 条。精确裁用MAXLEN量大时可以写成MAXLEN ~ 1000允许 Redis 少裁几条换更快的速度。老数据要归档就XTRIM或者定时把XRANGE扫出来丢到别的地方。三、消费者组同一条消息别抢两遍XREAD是广播味道的——每个客户端自己记游标两个进程都能读到同一条。做任务队列通常不是这个需求。你要的是一组工人抢活一条订单只被一个人处理。这就是消费者组。# 从头开始读历史。只想收新消息把 0-0 换成 $127.0.0.1:6379XGROUP CREATE orders packer0-0 OK$和0-0搞反是新手常踩的坑。组已经建了、流还不存在时后面加MKSTREAM让 Redis 顺手建一个空流。两个工人分别叫node-a、node-b127.0.0.1:6379XREADGROUP GROUP packer node-a COUNT1STREAMS orders1)1)orders2)1)1)1789086136869-02)... user_id1001...127.0.0.1:6379XREADGROUP GROUP packer node-b COUNT1STREAMS orders1)1)orders2)1)1)1789086136938-02)... user_id1002...表示给我一条组里还没分过的新消息。node-a拿走第一单node-b拿走第二单不会撞车。消息被读走之后还在 Stream 里。Redis 另外记了一本账谁领了、还没签字。这本账叫 PELPending Entries List。127.0.0.1:6379XPENDING orders packer1)(integer)22)1789086136869-03)1789086136938-04)1)1)node-a2)12)1)node-b2)1处理完了签个字127.0.0.1:6379XACK orders packer1789086136869-0(integer)1XACK只是从 PEL 里划掉不是删消息。别的组还能再读同一条你自己以后用XRANGE也能翻到。这就是 List 做不到的回溯。再建一个组试试127.0.0.1:6379XGROUP CREATE orders stock0-0 OK127.0.0.1:6379XREADGROUP GROUP stock warehouse-1 COUNT1STREAMS orders# 又拿到 1001 那单packer负责打包stock负责扣库存各记各的进度。组与组之间互不打扰。四、工人挂了活怎么转出去node-a领了单还没XACK就进程没了。消息不会丢它还挂在 PEL 里。别人可以用XPENDING看到再用XAUTOCLAIM认领超时的单# 把 packer 组里 idle 超过 60000 毫秒的未确认消息转给 node-bXAUTOCLAIM orders packer node-b600000-0 COUNT10XCLAIM是指定 ID 硬抢XAUTOCLAIM6.2更省事按空闲时间批量捞。线上常见写法工人循环XREADGROUP拿新活隔一会儿再XAUTOCLAIM扫一遍别人留下的残骸。还没 ACK 的时候同一个工人把 ID 写成0再读一次读到的是自己 PEL 里的旧单不是新单。重启之后先用这个把上次没做完的活做掉再去抢。五、生产者重试别写出两条一样的Redis 8.6 给XADD加了幂等。网络抖一下客户端重发以前会在流里留下两条一模一样的订单。完整写法要带生产者 IDID 仍然用*127.0.0.1:6379XADD pay_log IDMPAUTO pay-svc-1 * order_id9001amount2991789086143272-0# 内容一样再发一次ID 还是这条流的长度不变127.0.0.1:6379XADD pay_log IDMPAUTO pay-svc-1 * order_id9001amount2991789086143272-0127.0.0.1:6379XLEN pay_log(integer)1IDMPAUTO按消息内容算一个内部指纹内容相同就当重试服务重启后同一个进程要继续用同一个生产者 ID上面的pay-svc-1换一个就失效了。内容可能真的重复——比如用户连点两下两笔金额一样——就别用IDMPAUTO自己带业务单号XADD pay_log IDMP pay-svc-1 txn-9001 * order_id9001amount299XADD pay_log IDMP pay-svc-1 txn-9001 * order_id9001amount299# 重试还是同一条XADD pay_log IDMP pay-svc-1 txn-9002 * order_id9001amount299# 另一笔会新增幂等记录默认只记大约 100 秒、每个生产者最多 100 个 ID。隔太久再重试Redis 可能已经忘了又会写入一条。业务上重试窗口比较长的话用XCFGSET调大XCFGSET pay_log IDMP-DURATION600IDMP-MAXSIZE1000注意XCFGSET会清掉当前这条流上已经记下的幂等指纹改完配置再跑你的重试逻辑。六、和 Kafka 比Kafka 分区、副本、消费积压、跨机房这些不是 Redis Streams 要做的事。Streams 吃的是 Redis 已经在那、消息量不大、丢了会心疼但还没到要单独养一套 Kafka 的场景下单后发短信、扣积分、刷新库存、服务内部的任务队列。几个硬差别数据在 Redis 内存里堆积能力看你给 Redis 的内存和MAXLEN不是磁盘日志高可用靠 Redis 自己的复制 / 集群不是 Kafka 那套 ISR消费者组够用管理命令也少没有 Broker、Topic、再均衡那一长串运维如果允许偶尔丢、逻辑又简单List 就够了。要确认、要重放、不要消息重复用 Streams。真要日千万级、还得按分区扩去看 Kafka 或 RocketMQ不要逼 Redis 干这活。七、练手XADD三条订单进ordersXRANGE orders - 能看到。XGROUP CREATE orders packer 0-0两个窗口分别用node-a、node-b做XREADGROUP确认两人拿到的 ID 不同。只 ACK 其中一条XPENDING里应该还剩一条。Redis 8.6 的话用IDMPAUTO把同一条支付记录发两遍XLEN仍是 1。能走完这四步List 当年那几个坑基本都有着落了消息还在、能签字、能换人接着干、重试不容易写重。下期讲讲缓存的穿透、雪崩、击穿。本次导航结束欢迎关注、点赞、转发。