REDIS 5.0+ · STREAM · XADD · XREADGROUP · XACK · PEL · XAUTOCLAIM

一本只能往后写的流水账，每个班组夹着自己的书签

Stream 是 Redis 5.0 加的消息队列类型。生产者只往末尾追加，消息不会因为被读走就消失；每个消费组记着自己读到哪了，组里的每个人还要记着“我领了哪些还没处理完”。这样消费者挂了、重启了，消息都还找得回来。

1001-0下单 #10011002-0下单 #10021003-0下单 #10031004-0下单 #10041005-0下单 #10051006-0下单 #10061007-0下单 #1007XADD →只能往这追加
↑ 积分服务组读到这↑ 发货服务组读到这

两个消费组读的是同一本账，各读各的进度，互不影响。真实 ID 前半段是毫秒时间戳，这里缩短了。

流水账Stream

只追加的消息日志，读了不删，每条有唯一递增的 ID。

要自己控制长度
班组consumer group

一个组夹一个书签，组内分活，组间各拿一份。

一般一个服务一个组
组员consumer

组里的一个实例，领到的消息只归它处理。

名字要固定
待确认清单PEL

领了但还没 XACK 的消息，都记在这里。

消息不丢的关键

Redis 做消息队列的三代

在 Stream 之前，大家用 List 或 Pub/Sub 凑合。它们各缺一块：要么没有确认，要么不保存，要么不能多个服务各收一份。

ListLPUSH + BRPOP

LPUSH queue msg
BRPOP queue 0
消费者取走msg处理到一半崩溃 → 丢了

简单好用，消息会持久化。但一取走就从列表里没了，处理失败就丢；一条消息只能给一个消费者，没法让多个服务各收一份。可以用 BLMOVE 先挪到“处理中”列表来补救。

Pub/SubPUBLISH / SUBSCRIBE

SUBSCRIBE order-events
PUBLISH order-events msg
订阅者重启的那几秒 → 期间的消息全丢

一条消息发给所有在线的订阅者，但不保存：不在线就收不到，也没有确认。订阅者处理太慢，缓冲区超限会被断开（client-output-buffer-limit pubsub 32mb 8mb 60）。

Stream5.0 起

XADD orders * id 1001
XREADGROUP GROUP g c1 STREAMS orders >
c1领走后崩溃 → 消息还在 PEL 里 → 别人认领 不丢

消息持久化、有 ID、能回溯；有消费组和确认机制；多个组各收一份，组内多个实例分着处理。

遇到这种情况ListPub/SubStream
消费者暂时下线消息留在列表里这段时间的消息丢了消息还在，回来接着读
处理到一半崩溃已经取走，丢了丢了还在 PEL 里，可以重新处理
两个服务都要收到只能一个收到都收到两个消费组各收一份
一个服务多个实例分活可以每个实例都收到，重复处理同一组内自动分
回头再看历史消息取走就没了不保存按 ID 范围读

Stream 长什么样

每条消息是一组“字段-值”，外加一个由 Redis 生成的 ID。ID 一定递增，所以消息天然有序，也能按时间范围查。

消息 ID
1730793600123毫秒时间戳-0同一毫秒内的序号

XADD 时传 *，Redis 自动生成。如果机器时钟往回调了，Redis 会沿用上一个 ID 的时间戳、只加序号，保证 ID 不会变小。

底下怎么存
基数树节点 1730793600… 基数树节点 1730793601… 基数树节点 1730793602…

一棵基数树（rax）按 ID 前缀索引，每个节点里是一个 listpack，紧凑地装着一批消息（Redis 第 04 篇）。按 ID 查找和范围读都很快。

基本命令redis-cli
XADD orders MAXLEN ~ 100000 * orderId 1001 amount 99   # 追加，并把长度大约控制在 10 万条
→ "1730793600123-0"
XLEN orders                                            # 一共多少条
XRANGE orders - + COUNT 10                             # 从头读 10 条；- 最小 + 最大
XREAD COUNT 10 BLOCK 5000 STREAMS orders $             # 不用消费组：阻塞等 5 秒，只要新来的（$）

消费组：组内分活，组间各拿一份

每个消费组有一个书签（last_delivered_id），记着组里最后发出去的是哪条。组里谁来要消息，就从书签后面拿下一条给它，并记进它的待确认清单（PEL），直到它 XACK。

发货服务组 ship
c1 拿 1001、1003c2 拿 1002、1004

同一组的两个实例分着处理，一条消息只给其中一个。

积分服务组 points
c9 拿 1001～1004

另一个组从头读，所有消息它也都能收到一份，和发货组互不影响。

消费组命令redis-cli
XGROUP CREATE orders ship $ MKSTREAM                     # 建组：$ 只要以后的新消息，0 从头开始
XREADGROUP GROUP ship c1 COUNT 10 BLOCK 2000 STREAMS orders >   # >：要组里还没发给任何人的新消息
XACK orders ship 1730793600123-0                         # 处理完了，从 PEL 里划掉
XREADGROUP GROUP ship c1 STREAMS orders 0                # 0：不要新的，把我 PEL 里没 ACK 的再给我一遍
XPENDING orders ship - + 10                              # 看谁手上压着哪些、压了多久、投递过几次
XAUTOCLAIM orders ship c1 60000 0-0 COUNT 10             # 6.2 起：把闲置超过 60 秒的消息认领到 c1 名下

消费组实验台

一个 Stream orders，消费组 g1 里有 c1、c2 两个消费者。生产、读取、确认、让 c2 崩溃、重启、让 c1 认领超时消息，再建一个组 g2 从头读。下面会显示每条消息在各组里的状态、每个人的待确认清单，以及实际执行的命令。

生产者时钟 0 秒
c1
c2
组 g2

    怎么做到消息不丢

    Stream 的保证是“至少一次”：只要没 ACK，消息就一直在某个人的待确认清单里，迟早会被再处理一次。代价是可能重复，所以消费逻辑必须幂等。

    1 · 处理完再 ACK先干活，后确认业务成功之后再 XACK。崩在中间，消息还在 PEL 里。
    2 · 重启先补旧账读 ID 0，处理自己的 PEL消费者名字要固定（比如主机名），重启后先用 0 读回没 ACK 的，再用 > 读新的。
    3 · 别人挂了就认领定时 XAUTOCLAIM实例永久下线了，它 PEL 里的消息由别的实例认领，投递次数加 1。
    4 · 坏消息进死信投递次数太多就隔离同一条投递超过 3 次还失败，写进 orders:dlq 再 ACK，别让它无限重试。
    Spring Data Redis 消费手动 ACK
    var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
        .builder().pollTimeout(Duration.ofSeconds(2)).batchSize(10).build();
var container = StreamMessageListenerContainer.create(connectionFactory, options);

container.receive(
    Consumer.from("ship", hostName),                         // 消费者名字固定
    StreamOffset.create("orders", ReadOffset.lastConsumed()),  // 等于 >
    msg -> {
        shipService.handle(msg.getValue());                     // 必须幂等
        redis.opsForStream().acknowledge("orders", "ship", msg.getId());
    });
container.start();
    生产和建组StringRedisTemplate
    // 发消息
RecordId id = redis.opsForStream()
        .add("orders", Map.of("orderId", "1001", "amount", "99"));

// 控制长度：大约保留最近 10 万条
redis.opsForStream().trim("orders", 100_000, true);

// 启动时建组；组已存在会报 BUSYGROUP，忽略即可
try {
    redis.opsForStream().createGroup("orders", ReadOffset.from("0"), "ship");
} catch (RedisSystemException e) {
    if (!String.valueOf(e.getMessage()).contains("BUSYGROUP")) throw e;
}
    Redis 本身也会丢Stream 的持久化靠 RDB / AOF，主从复制是异步的。AOF 每秒刷盘时断电丢约 1 秒，主从切换时也可能丢最后一点（Redis 第 02 篇、第 03 篇）。对账、支付这类一条都不能丢的消息，用 Kafka、RocketMQ 这类专门的消息队列。

    延迟消息：Stream 不支持，用 ZSet 补

    “30 分钟后取消未支付订单”这种需求，Stream 没有现成的办法。常见做法是用 ZSet 当延迟区：分数是到期时间，定时把到期的挪进 Stream 或直接处理。

    ZSet delay:orderscancel:1001 → 10:00:00 已到期cancel:1002 → 10:00:02 已到期cancel:1003 → 10:30:00
    每秒用一段 Lua 取出到期的、删掉、XADD 进 orders:cancel
    搬运到期消息的 Lua取出和删除必须原子，否则多个实例会搬重复
    local due = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, 100)
for _, m in ipairs(due) do
    redis.call('ZREM', KEYS[1], m)
    redis.call('XADD', KEYS[2], '*', 'msg', m)
end
return #due

    集群下 KEYS[1] 和 KEYS[2] 要在同一个槽，可以都加上同一个 hash tag。也可以直接用 Redisson 的 RDelayedQueue，它就是这个思路。

    和专门的消息队列比

    已经有 Redis、消息量不大、能接受极端情况下丢一点，Stream 很省事。否则用专门的消息队列。

    Redis StreamKafkaRabbitMQRocketMQ
    存在哪内存，靠 RDB / AOF 落盘磁盘日志，多副本磁盘，可配镜像 / 仲裁队列磁盘，多副本
    能堆多少受内存限制很多，按时间或大小保留中等，堆多了变慢很多
    扩展一个 key 只在一个节点，要自己拆多个 key分区，天然水平扩展按队列扩按队列扩
    回溯历史按 ID按 offset 或时间消费后删除按时间
    死信 / 重试自己实现自己实现死信交换机自带重试和死信
    延迟消息不支持，用 ZSet 补不支持插件或 TTL + 死信支持
    运维成本已有 Redis 就几乎为零高中中
    适合轻量异步任务、站内通知、小流量事件日志、埋点、海量事件流复杂路由、业务消息订单、交易类业务消息

    Redis 版本，以及 JDK 8 和 JDK 17

    Stream 在 5.0 加入后一直在补功能。Java 这边 Spring Data Redis 2.2 起支持 Stream，Boot 2 和 3 用法一样。

    Redis 5.0 ～ 6.0 · Boot 2

    JDK 8+
    • 5.0：Stream、消费组、XCLAIM（要先 XPENDING 查出来再一个个认领）
    • Spring Data Redis 2.2 起有 StreamOperations 和 StreamMessageListenerContainer
    • 配置前缀 spring.redis.*

    Redis 6.2 ～ 7 · Boot 3

    JDK 17+
    • 6.2：XAUTOCLAIM 一条命令扫描并认领；XADD 支持 MINID 按 ID 裁剪、NOMKSTREAM；XPENDING 能按闲置时间过滤
    • 7.0：XINFO GROUPS 多了 lag，直接看到这个组还有多少没读
    • 监听容器可以传自己的 Executor；JDK 21 下可以给它虚拟线程
    没变的：只追加的日志消费组 + PEL至少一次，要幂等内存装不下就要裁剪

    常踩的坑

    Stream 的坑大多出在“没 ACK”和“没裁剪”上。

    Stream 越来越大，把内存吃光了

    读过的消息不会自动删。XADD 时带 MAXLEN ~，或定时 XTRIM。但裁剪会删掉还没被慢消费者读到的消息，要盯着各组的 lag。

    PEL 越来越长，重启后一大堆消息被重新处理

    处理完忘了 XACK，或者用了自动 ACK 又在别处手动 ACK。用 XPENDING 看看谁手上压着消息。

    一个实例下线后，它领的消息再也没人处理

    消息卡在它的 PEL 里。其他实例定时用 XAUTOCLAIM 认领闲置太久的消息。

    每次发版，PEL 里都多出一堆孤儿消息

    消费者名字用了随机 UUID，重启后是个“新人”，旧名字下的消息没人管。用固定名字，比如主机名或 Pod 名。

    同一笔订单发了两次货

    至少一次投递，消息可能重复：处理完还没来得及 ACK 就崩了，或者被别人认领后又处理一遍。业务用订单号做幂等。

    一条格式错误的消息，被反复投递了上千次

    每次都失败、都不 ACK，又被认领回来。看 XPENDING 里的投递次数，超过阈值就转进死信 Stream 并 ACK。

    BUSYGROUP Consumer Group name already exists

    每次启动都 XGROUP CREATE，组已经存在就报这个错。捕获后忽略，或者先用 XINFO GROUPS 查一下。

    用 Pub/Sub 发业务消息，服务重启就丢

    Pub/Sub 不保存消息。要可靠送达的用 Stream；Pub/Sub 只适合“丢了也无所谓”的通知，比如本地缓存失效广播。

    集群里一个 Stream 的吞吐上不去

    一个 key 只在一个节点上。按业务拆成多个 Stream，比如 orders:0～orders:7，生产者按订单号取模。

    同一个用户的消息处理顺序乱了

    组里多个消费者并行处理，先领的不一定先完成。需要按用户有序，就按用户 ID 把消息分到固定的 Stream，每个 Stream 只让一个消费者处理。