《Java 消息中间件·可靠消息与消费治理》6.2 重试主题与退避

为什么重试要离开消费线程

6.1 节的重试是在消费循环里 Thread.sleep 之后重来。它有三个问题：

1. 占住分区。 等待期间这个消费者不再 poll，同分区后面的消息全部排队。退避 5 分钟，就是整个分区停 5 分钟。
2. 进程崩溃就丢。 等待中的记录只存在于内存里。位移没提交，重启后会重新拉到，这还算幸运；如果代码先把记录放进内存队列再提交位移，崩溃就是永久丢失。
3. 无法按失败原因分层。 所有失败共用同一个等待时间，网络抖动和下游维护窗口被同等对待。

重试主题（retry topic）把「等待」从消费线程里挪出去：失败了就把消息写到下一级主题，提交当前位移，当前消费者立刻去处理后面的消息。等待时间由下一级主题的消费者负责。代价是消息多了一次落盘、一次拉取，以及一套需要维护的主题拓扑。

拓扑：每一级都是一个独立主题

本机实验的主题设计：

每一级用独立的消费组，不是同一组的多个订阅。同一组订阅多个主题时，任何一个主题的分区数变化或成员进出都会触发整组再平衡，主链路会被重试主题的抖动牵连；而且整组共享同一份 lag 与告警，出了问题分不清是哪一级。分开之后，每一级的消费速率、lag、告警都可以单独看。

x-attempt 记录已经尝试过几次，x-first-failure-at 从第一次失败起就不再变化。这两个 header 让「重试了多久」不依赖主题名去数。

主题要逐个建齐，漏掉一个，对应的消费线程就只是空转，不会报错：

for t in library.loan.overdue \
         library.loan.overdue.retry.1 \
         library.loan.overdue.retry.2 \
         library.loan.overdue.retry.3 \
         library.loan.overdue.DLT; do
  /opt/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 \
    --create --topic "$t" --partitions 1 --replication-factor 1
done

生产上更常见的做法是由应用启动时用 Admin.createTopics 幂等地建，避免漏建某个主题导致消费线程空转。

退避时间放在哪：四种方案

Kafka 4.3.0 没有原生的「延迟消息」。所谓延迟主题，都是应用层用某种方式实现的。常见四种：

还有一种「按延迟分桶」的变体：不按次数分主题，而按延迟分 retry.5s、retry.1m、retry.10m。同一档内的消息一起等，主题数量可控，但每档的消费并行度独立于主主题，需要单独评估。

本篇选「延迟主题 + not-before」，因为它把等待状态写进了消息 header，进程重启后重新拉到消息就能恢复等待，不依赖内存。实现上有一个必须讲清楚的取舍：消费者读到消息后如果还没到时间，就在线程里等，这个分区在此期间不再拉取。 要做到「等待但不阻塞分区」，需要把消息暂存到内存并 pause() 分区，再用定时器 resume()；暂存区崩溃会丢，又回到了崩溃安全问题。两种做法没有免费午餐，只是把风险放在不同位置。

实现：一个消费者只做一件事

每一级消费者的骨架相同，差别只在「失败后写哪个主题、等多久」：

TopicPartition tp = new TopicPartition(topic, 0);
consumer.assign(List.of(tp));
consumer.seekToBeginning(List.of(tp));   // 实验中从起点读；生产按提交位移

while (!stop) {
    for (ConsumerRecord<String, String> r : consumer.poll(Duration.ofMillis(300))) {
        long notBefore = longHeader(r, "x-not-before");
        long wait = notBefore - System.currentTimeMillis();
        if (wait > 0) {
            Thread.sleep(wait);          // 等待期间不 poll，本分区暂停
        }
        if (handle(r.value())) {         // 业务处理
            consumer.commitSync();
            continue;
        }
        String next = level < 3 ? topic + 1 : DLT;
        long delay = level < 3 ? backoff[level] : 0;
        ProducerRecord<String, String> fwd = new ProducerRecord<>(next, r.key(), r.value());
        fwd.headers()
           .add(header("x-attempt", attempt + 1))
           .add(header("x-first-failure-at", firstFailure))
           .add(header("x-not-before", System.currentTimeMillis() + delay))
           .add(header("x-trace", trace + ">" + next + "@" + now()));
        producer.send(fwd).get();        // 先确认写入
        consumer.commitSync();           // 再提交当前位移
    }
}

三个细节决定了它是否可靠：

• 先 send().get() 再 commitSync()。 反过来会在崩溃时丢消息；当前顺序最坏情况是重复转发，下游按业务键去重。
• x-not-before 用绝对时间戳。 存「还要等多久」在消息积压时会算错，因为消息在主题里排队的时间没有被扣掉。
• DLT 消费者只记录，不再转发。 第一版代码让 DLT 也走同一套转发逻辑，结果它把消息转发给了自己，95 秒内写满了 64 MB 的终端输出上限。这个 bug 很典型：级联逻辑必须有终止条件，不能靠「应该不会走到」。

每一级的消费者配置也要跟着主题走，不能照抄主链路：

bootstrap.servers=127.0.0.1:19592
group.id=libraryd-retry-library.loan.overdue.retry.2
enable.auto.commit=false
auto.offset.reset=earliest
max.poll.records=1
max.poll.interval.ms=300000

max.poll.records=1 是为了让「等待一条消息」不会把后面已经拉下来的记录一起扣在内存里。max.poll.interval.ms 要大于最大退避时间：本篇最大 45 秒，默认 5 分钟足够；如果退避配到 10 分钟，就必须同步调大。用 subscribe 的消费者在等待期间不 poll，会被判为处理超时踢出组，提交位移时抛 CommitFailedException；本篇用 assign 手动分配，没有组心跳，所以 45 秒等待不会触发踢出，但也没有自动再平衡的容错。

实测：一条消息的五跳轨迹

本机实测（Temurin 25.0.5+8 / kafka-clients 4.3.0 / apache/kafka:4.3.0 单节点 KRaft / 2026-10-29）：实验只投一条永远失败的消息，value 含 payload=corrupt，三档退避 5 秒、15 秒、45 秒。以下是原始输出（时间戳为 +08:00）：

[13:37:04.356] attempt library.loan.overdue         off=0 attempt=1 value=overdue loan=loan-0 payload=corrupt
[13:37:04.382] forward library.loan.overdue         off=0 -> library.loan.overdue.retry.1 delay=5000ms attempt=1
[13:37:04.382] hold   library.loan.overdue.retry.1 off=0 wait=4974ms (not-before 13:37:09.356)
[13:37:09.362] attempt library.loan.overdue.retry.1 off=0 attempt=2 value=overdue loan=loan-0 payload=corrupt
[13:37:09.390] forward library.loan.overdue.retry.1 off=0 -> library.loan.overdue.retry.2 delay=15000ms attempt=2
[13:37:09.390] hold   library.loan.overdue.retry.2 off=0 wait=14973ms (not-before 13:37:24.363)
[13:37:24.368] attempt library.loan.overdue.retry.2 off=0 attempt=3 value=overdue loan=loan-0 payload=corrupt
[13:37:24.395] forward library.loan.overdue.retry.2 off=0 -> library.loan.overdue.retry.3 delay=45000ms attempt=3
[13:37:24.395] hold   library.loan.overdue.retry.3 off=0 wait=44974ms (not-before 13:38:09.369)
[13:38:09.375] attempt library.loan.overdue.retry.3 off=0 attempt=4 value=overdue loan=loan-0 payload=corrupt
[13:38:09.434] forward library.loan.overdue.retry.3 off=0 -> library.loan.overdue.DLT delay=0ms attempt=4
[13:38:09.435] dead-letter library.loan.overdue.DLT off=0 attempt=4 key=loan-0

换成表格看得更清楚：

实测的间隔比配置多 6–7 毫秒，来自写入、拉取、线程调度。单次运行，随调度波动，不作基准。

DLT 里的 header 原文：

DLT offset=0 key=loan-0 value=overdue loan=loan-0 payload=corrupt
  x-attempt            = 4
  x-first-failure-at   = 1793252224356
  x-not-before         = 1793252289380
  x-trace              = main@13:37:03.800>library.loan.overdue.retry.1@13:37:04.356>library.loan.overdue.retry.2@13:37:09.363>library.loan.overdue.retry.3@13:37:24.369>library.loan.overdue.DLT@13:38:09.380

x-trace 把五跳串成一行，排查时不用去五个主题里按 key 找。注意它记录的是转发时刻，不是到达下一级后的处理时刻：retry.1@13:37:04.356 表示消息在 13:37:04.356 被写进 retry.1，而它真正被 retry.1 处理是 13:37:09.362。x-attempt=4 表示业务处理一共尝试了 4 次（主链路 1 次 + 三级重试 3 次）。x-first-failure-at=1793252224356 对应 13:37:04.356，和主主题第一次失败的时刻一致。

重试的放大效应

重试不是免费的。每条失败消息在每一级都占一条记录、一次拉取、一次业务调用。下一组实测数字：100 条业务消息，每 10 条有 1 条永远失败，三档退避都设为 0 以便快速跑完（仅用于计数，不模拟真实等待）：

count library.loan.overdue             = 100
count library.loan.overdue.retry.1     = 10
count library.loan.overdue.retry.2     = 10
count library.loan.overdue.retry.3     = 10
count library.loan.overdue.DLT         = 10

100 条业务消息最终产生 140 条记录、140 次处理尝试，放大 1.4 倍。公式是：

总记录数 = N + N × f × L

其中 N 是业务消息数，f 是最终失败率，L 是每条失败消息额外落盘的主题数（本例 3 级重试 + 1 次 DLT，共 4）。f=10%、L=4 时是 1.4 倍；f=30% 时变成 2.2 倍。重试级数越多，放大越接近线性；失败率才是真正的乘数。

放大效应体现在四个地方：

第三项最危险：下游已经过载时，重试会把压力再放大一次。所以退避必须带抖动（jitter），例如在 5 秒基础上加 ±20% 随机，避免同一批失败的消息在同一毫秒一起回来。

顺序、分区与队头阻塞

重试主题会改变顺序语义：

• 同 key 的消息可能乱序。 第一条失败进了 retry.1，第二条成功直接在主主题处理完。等第一条重试成功，它比第二条晚到下游。需要严格顺序时，只能让后一条也停下来等，那就回到了阻塞式重试。
• 分区数要单独决定。 本篇每级 1 分区，重试串行，吞吐低但顺序简单。若主主题 12 分区，重试主题也建 12 分区并按同一个 key 分区，能并行，但跨主题的先后仍然不保证。
• 等待会阻塞该分区。 实测在 retry.3 上 hold 了 44,974 ms；这段时间同一分区的下一条重试消息不会被拉取。本实验只有一条消息，所以只观察到等待本身，没有制造后续消息的延迟数据；机制上是必然的。按延迟分桶（5 秒桶、1 分钟桶）能减少「短延迟被长延迟挡住」的概率。

退避参数怎么定

没有一个通用的「5 秒、15 秒、45 秒」。参数由三件事决定：

1. 下游恢复时间。 下游是数据库主从切换，退避要给到分钟级；是偶发超时，秒级就够。
2. 业务可接受的延迟。 借阅事件晚 1 分钟处理通常没事，库存扣减晚 1 分钟可能已经超卖。容忍度决定总重试窗口的上限。
3. 重试的成功率曲线。 如果第二次重试的成功率已经接近第一次，说明失败与时间无关，多一级只是放大；如果成功率随时间明显上升，说明退避有效。

一种可操作的做法是把「总窗口」定死，再分配各级退避。例如总窗口 5 分钟：5 秒、30 秒、4 分 25 秒，三级之后进 DLT。不要用「每级乘 3、共 5 级」这种没有上限的指数退避，第 5 级的等待会到小时级，业务早已过期。

抖动不要写成「随机 0 到 5 秒」，那会把最短退避变成 0，失去退避的意义。正确做法是在基准值上下浮动一个比例：

static long jitter(long baseMs, double ratio) {
    long delta = (long) (baseMs * ratio);
    return baseMs + java.util.concurrent.ThreadLocalRandom.current()
            .nextLong(-delta, delta + 1);
}

基准 5000 毫秒、比例 0.2，实际值落在 4000–6000 毫秒。测试里要能固定随机种子或直接传 0 比例，否则时间相关的断言会间歇性失败。

每一级要看什么

重试主题把一条消息的旅程拆成了五段，监控也必须跟着拆：

「升级率」比重试成功率更有用：它直接回答「有多少比例的消息没能在这级解决」。如果 retry.1 的升级率从 5% 涨到 40%，不用看应用日志就知道下游出了问题。

什么时候不该用多级重试主题

• 失败率极低且可接受人工处理。 一级重试 + DLT 就够，多两级只增加运维面。
• 下游本身幂等且恢复很快。 消费线程内短退避（几百毫秒）成本更低。
• 消息有时效性。 逾期 10 分钟的消息重试成功也没有业务价值，不如直接进 DLT 并告警。
• 没有独立的消费容量。 重试主题需要额外的消费者实例；如果容量本来就紧张，重试会挤占主链路。
• 失败原因是数据本身。 数据损坏型的消息重试多少次都是同一个异常，应该直接进隔离区，由人修复后再回灌（6.3 节）。

一个折中的起点是「一级重试 + 死信」：5 秒后重试一次，仍失败进 DLT。等观测数据说明「多一级能救回一部分」时再加级，而不是一开始就搭五级拓扑。主题拓扑一旦上线，删级比加级难得多——消费者的位移、监控、告警都绑在主题名上。

回到 libraryd

libraryd 的逾期事件 library.loan.overdue 走三级重试：5 秒处理下游瞬时超时，30 秒处理连接池耗尽，5 分钟处理下游短时维护。超过 5 分钟仍失败，进 DLT 由人工判断是数据问题还是下游长期故障。

关键不是这三个数字，而是三件事同时成立：每一级有独立消费组与独立 lag 告警；x-first-failure-at 让「消息在系统里待了多久」可查；DLT 的 x-trace 能还原完整路径。缺了任何一条，多级重试就只是把失败藏得更深。

阅读导航：上一节 6.1 死信队列的设计 ｜ 下一节 6.3 毒消息隔离与人工干预