Kafka延迟/重试/死信队列全链路落地：时间桶与幂等消费实践

做消息中间件这一块的朋友应该没少被问过Kafka 能不能做延迟队列消息消费失败之后能不能自动进重试队列隔一会儿再试重试次数用完了又该怎么把消息送进死信队列这三个问题在我负责的订单场景里全都撞上了当时翻了不少资料但大多数文章只讲了其中一块很少有把延迟、重试、死信串成一条完整链路的。这篇就把我目前在生产环境里跑通的方案完整讲一遍包括为什么这么选型、核心代码骨架、topic 怎么规划以及实测中踩过的几个比较隐蔽的坑给正打算在 Kafka 上自建这套机制的读者一个可以照着搭的样本。1. 三个队列的边界延迟、重试、死信分别解决什么问题很多人刚接触这三个概念时容易混在一起实际上它们在业务里承担的任务完全不同如果边界没有分清后面的 topic 设计、消费者逻辑都会乱成一团。1.1 延迟队列在业务里的真实含义延迟队列的本质是让一条消息在指定的未来时间点变得可见而不是消息本身晚发送。举个例子电商下单后 30 分钟未支付需要自动关单如果在下单那一刻就把关单指令投递到 MQ消费者立刻就会看到这条消息然后立刻把订单关掉这显然不对。我们希望这条消息被雪藏30 分钟到点之后才被消费者看见。很多人会下意识想到 Kafka 的 timestamp 和 retention。但这里必须说清楚Kafka 的retention.ms解决的是消息在 broker 上保留多久后被删除跟消息延迟多久后投递给消费者完全是两码事。保留时间只影响日志清理不会让一条消息在指定时间点突然变得可消费。Kafka 本身也没有类似延迟插件那样的开关所以凡是网上说给 topic 配个 TTL 就能延迟的基本都理解偏了。延迟队列最常见的业务触发场景有三类订单超时关闭、超时确认收货定时任务下发给某个执行节点对账、补单这类需要延后处理的轻量任务。这些场景都存在同一个矛盾事件已经发生了但动作不能立刻执行。延迟队列就是用来承接这个未来动作的。1.2 重试和死信为什么要分开重试队列解决的是这次没办成但下次可能办成的问题。典型场景是消费者调下游 API下游超时或返回 5xx或者数据库出现短时间的锁等待。这类失败是暂时性的给点时间再试一次成功率很高。死信队列则完全不同。它承接的是无论试多少次都不可能成功的消息。比如消息体是 JSON但反序列化直接抛异常或者订单号在业务库里根本不存在再比如用户账号已经被注销下游明确表示这个单没法处理。把这些消息反复重试只会浪费 CPU 和带宽还会拖慢正常消息的处理进度。如果把暂时性失败和永久性失败混在一个队列里你很难判断这条消息失败了很多次到底是网络问题还是数据问题。分开之后逻辑就变得非常清晰重试是过程状态消息还活着还有机会被正常消费死信是终态消息已经被判了死刑要有专门的人工或补偿流程去处理。1.3 Kafka 原生缺什么需要自己补什么Kafka 在消息可靠性和吞吐上的能力没得说但它在队列语义上是相对朴素的消息进 topic 之后所有订阅者立刻就能拉取没有延迟可见的概念也没有失败后重新投递几次的内置机制。所以想在 Kafka 上实现延迟、重试、死信核心思路不是指望 Kafka 原生给你这个能力而是用 topic 的组合和消费者的路由逻辑来模拟。大致套路是延迟把消息先写进一个中间调度 topic由调度服务藏着到期后再转发到真正的业务 topic重试业务消费失败后把消息写回一个重试 topic等退避时间过了再回调死信重试超过阈值或者反序列化失败直接写入专门用于隔离的死信 topic带上完整的失败上下文。这套思路不依赖 Kafka 的任何特殊版本也不用改 broker 配置纯客户端就能实现通用性很强。下面我会按延迟、重试、死信的顺序逐层拆开讲最后再把三者串成一套可落地的链路。2. 延迟队列方案选型为什么我选了中间调度 Topic 时间轮而不是改 TTL延迟队列是三个队列里最容易做错的部分因为网上能搜到的偷懒方案特别多。我这边试过几种之后最终选定了中间调度 topic 时间桶的做法下面把对比和理由讲透。2.1 几种方案的横向对比我把用过的方案整理成了表格方便你对照自己的场景做取舍方案实现思路优点缺点消费者线程直接 sleep消费者拉到消息后在本地线程里 sleep N 毫秒再执行实现最简单改动量最小线程全部被占住吞吐量下降严重服务重启时内存中等待的任务直接丢失修改 Kafka retention 或 TTL让消息在指定时间后过期模拟延迟不需要额外消费者到期是删除消息不是投递消息消息直接丢了外部存储索引 定时扫描把到期时间写入 Redis 或 DB定时任务扫描到期记录后推送实现直观调试方便要维护索引一致性还得处理外部存储和 Kafka 的双写问题中间调度 topic 时间桶消息先进 delay-topic调度消费者按到期时间分桶到点转发目标 topic消息始终在 Kafka 上可靠性好节点重启可恢复和业务消费者完全解耦多了一个 topic 和一段转发逻辑我最终选了最后一种。核心原因是它没有引入额外的存储依赖所有消息生命周期都在 Kafka 内部闭环。延迟消息先落到 Kafka 的 delay-topic就算调度服务挂掉消息也只是暂时没人处理不会丢失重启后消费者从上次的 offset 继续拉依然能把消息转出去。2.2 调度链路的数据流设计先看整条链路长什么样生产者 -- delay-topic(带目标topic和到期时间) -- 调度服务(时间桶) -- 业务topic -- 业务消费者这里的关键是生产者发出去的并不是最终业务消费者要消费的那条消息而是一条带有投递计划的包装消息。调度服务消费 delay-topic 后把消息按到期时间放进不同的时间桶时间到了再原样转发到真正的业务 topic。为了让这条链路通用我会在消息的 header 里携带三个关键字段Header 字段含义示例target-topic到期后要转发到哪个 topicorder-auto-closedelay-ms相对延迟时间单位毫秒1800000origin-time消息产生时的时间戳1690000000000有了这三个字段调度服务就不需要关心具体业务是什么它只做一件事收到一条消息计算到期时间丢进对应的时间桶到点后按target-topic转发。任何新的延迟业务只需要按这个规范投递不需要改调度代码。2.3 简易时间桶的实现骨架我这里的实现叫时间桶本质是时间轮思想的简化版按秒级整点作为桶的 key调度服务内部维护一个 key 为到期时间戳、value 为消息列表的映射另有一个定时任务每秒检查当前秒是否有桶到期。核心逻辑大致是这样public class DelayBucketScheduler { // key: 到期时间戳秒value: 该秒到期的一组消息 private final ConcurrentHashMapLong, ListDelayedRecord buckets new ConcurrentHashMap(); private final KafkaProducerString, String forwardProducer; public void onPoll(ConsumerRecordString, String record) { String targetTopic header(record, target-topic); long delayMs Long.parseLong(header(record, delay-ms)); // 用消息的创建时间计算到期时间避免本地收到消息的时间误差 long expireAtSec (record.timestamp() delayMs) / 1000; DelayedRecord task new DelayedRecord(targetTopic, record.key(), record.value(), record.headers(), record.timestamp()); buckets.computeIfAbsent(expireAtSec, k - new CopyOnWriteArrayList()).add(task); } public void triggerExpired(long currentSec) { ListDelayedRecord tasks buckets.remove(currentSec); if (tasks null) { return; } for (DelayedRecord task : tasks) { ProducerRecordString, String out new ProducerRecord( task.getTargetTopic(), task.getKey(), task.getValue()); // 把原消息的 header 原样带到目标 topic业务消费者可以拿到原始上下文 task.getHeaders().forEach(h - out.headers().add(h)); forwardProducer.send(out, (metadata, exception) - { if (exception ! null) { // 转发失败的补偿策略记录 error-log 并考虑重新延后 log.error(forward delay record failed, exception); } }); } } }定时任务只需要每秒执行一次triggerExpired(nowSec)。这里我有一个经验是千万不要在消息真正被 poll 到的时候才去计算 delayMs 的起点要用 record.timestamp()也就是 broker 写入时间不然调度服务和业务服务之间哪怕有几毫秒的时钟差都会在长延迟场景下被放大。2.4 节点重启时怎么保证消息不丢单机时间桶有个致命问题桶里的数据在内存里节点一重启就没了。我的解决思路不算复杂但很实用——不依赖内存里的桶做持久化而是把消息本身就当成持久化存储。调度服务消费 delay-topic 时不要先删 offset而是让延迟消息始终留在 Kafka 里。时间桶只是内存中的加速索引真正的持久化载体是 delay-topic 自己和消费者 offset。节点重启后消费者从上次提交的 offset 继续拉取那些看似已经处理过的延迟消息会在 poll 中再次出现重新进入时间桶。这样只可能出现重新转发一次的问题不会出现漏转发的问题。而重复转发造成的幂等问题就交给业务消费者去兜底这也是后面第 6 节要专门强调的点。3. 重试队列从暴力重投到带退避的分级重试重试队列的设计直接关系到业务稳定性和下游系统的压力。最糟糕的做法是消费失败就立刻重试重试失败又立刻重试这种风暴式重试会把本来就暂时不可用的下游打到彻底不可用。正确做法是分级退避并且把重试次数和异常信息都管理起来。3.1 第一次失败后不要马上进队列我现在的做法是消费失败后先在当前消费者进程内做快速重试比如间隔 300ms 重试 1 到 2 次。原因是大量的瞬时失败网络抖动、连接池满、短暂的锁冲突往往在毫秒到秒级就能恢复快速重试的成功率很高。直接进 Kafka 重试 topic 反而要经过一次消息序列化、落盘、再消费开销大延迟还高。快速重试的代码骨架大致是这样public void safeProcess(ConsumerRecordString, String record) { Throwable lastError null; for (int attempt 0; attempt 2; attempt) { try { process(record); return; } catch (RetryableException e) { lastError e; try { Thread.sleep(300L * (attempt 1)); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw new RuntimeException(ie); } } } // 快速重试失败之后才进入重试队列的判定 handleRetry(record, lastError); }注意这里我用的是RetryableException这个自定义异常去区分可重试的失败和后面要说的不可重试的失败。如果抛出来的是反序列化异常、业务非法状态异常那根本不该进入这个快速重试逻辑要直接进死信。3.2 重试 Topic 的路由与退避计算快速重试仍然失败后消息会被投递到重试 topic。重试 topic 的消费思路不应该是消费者立刻拉取立即处理而应该复用第二章的延迟转发思想消息带着next-retry-time进入重试 topic由一个调度服务统一管理到时间后重新转发回业务 topic。退避间隔我采用的是指数退避加随机抖动重试次数退避间隔秒11 随机 0~122 随机 0~234 随机 0~448 随机 0~8516 随机 0~16最大间隔封顶 5 分钟避免下游长时间不可用时消息依旧越积越密。加随机抖动的目的很纯粹避免大量同批次失败的消息在同一个时间点同时重试造成下游流量尖峰。3.3 消息头的重试计数与异常快照进入重试 topic 的消息我要求在 header 里携带这么几个字段Header 字段说明retry-count已经重试的次数next-retry-time下次允许被消费的时间戳last-error-class最近一次异常的类名last-error-msg最近一次异常的 messageinitial-offset这条消息最初落业务 topic 时所在的 offsetretry-count和next-retry-time是路由和调度用的last-error-class和last-error-msg则完全是为了人工排查方便。我在生产上遇到过很多次重试好几轮还是失败想查原因为什么结果消息里只有原始 body一点异常线索都没有最后只能靠日志捞非常费劲。所以异常快照我强烈建议保留成本极低收益极高。这里还有一个小技巧投递到重试 topic 时把消息的 key 保持和原消息一致。这样同一个订单的多次重试会落在同一个分区消费端如果需要按 key 做局部顺序控制会容易很多。4. 死信队列把真处理不了的消息单独隔离死信队列这个名字听起来有点吓人实际上它就是业务系统的一个垃圾桶。但垃圾桶也不能乱扔必须设计好什么样的消息该扔进来扔进来之后长什么样谁来清理。4.1 进入死信的触发条件并不是所有处理失败的消息都应该进死信。我一般把进入死信的条件分为三类触发条件示例是否先走重试重试次数超过上限重试 5 次仍然抛RetryableException是先重试后死信消息格式错误JSON 解析失败、字段缺失否直接进死信业务明确拒绝订单状态不允许关单、账号已注销否直接进死信第二类和第三类的共同特点是再重试多少次结果都一样。如果让这种消息反复走重试队列既浪费资源又会把真正需要重试的消息挤到后面。所以一定要在消费者代码里区分异常类型通过自定义异常体系来控制路由方向。4.2 死信 Topic 的命名与字段规范我在项目中死信 topic 的命名统一为原业务 topic 加-dlq后缀比如订单关闭业务 topic 叫order-auto-close死信就是order-auto-close-dlq。这样一眼就能看出这个死信是从哪条业务链路来的。死信消息的 body 我保留原始消息内容不做任何改动方便补偿程序原样重投。额外的诊断信息全部放在 header 里Header 字段说明dlq-reason进入死信的原因如MAX_RETRY_EXCEEDED或DESERIALIZE_FAILEDdlq-time进入死信的时间戳original-topic消息原始业务 topicoriginal-partition原始分区original-offset原始 offseterror-stack最后一次失败的完整堆栈死信 topic 的retention.ms我会配置得比普通 topic 长一般 7 天以上。原因很简单死信不是给马上消费用的它需要给人工留出足够长的处理窗口否则人还没来得及补偿消息就被 broker 清理了那时候才是真丢数据。4.3 死信的处理和人工补偿死信消息不应该有自动消费者去尝试处理它否则就违背了死信设计的初衷。我这边配套的是一个非常轻量的人工运维工具功能就三个按 topic 和关键字查看死信列表单条或批量重新投递到原业务 topic标记某条死信为已处理或确认丢弃。重新投递就是在工具后台起一个 producer把死信消息的 body 原样发回original-topic发之前删掉retry-count等重试 header避免它再次进入那套重试逻辑。这里有个比较容易漏的点如果业务 topic 的 key 是业务唯一键重投时必须保留原来的 key否则消息会落到不同分区可能会带来乱序问题。5. 一整套联动落地Topic、消费者模板与配置参数单看前面几章是散的这一节我把它串成一个完整的落地示例。为了好描述我以一个虚构的订单自动关闭场景作为主线把 topic 规划、消费者模板、核心配置文件全部摆出来。5.1 Topic 规划与创建脚本针对订单自动关闭我会创建四个 topicTopic 名称用途分区数备注order-auto-close业务消费者真正监听的 topic3保留 1 天即可order-auto-close-delay延迟调度中间 topic3保留 1 天order-auto-close-retry重试调度中间 topic3保留 1 天order-auto-close-dlq死信 topic3保留 7 天创建脚本可以用 Kafka 自带的命令行工具kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic order-auto-close \ --partitions 3 --replication-factor 2 \ --config retention.ms86400000 kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic order-auto-close-delay \ --partitions 3 --replication-factor 2 \ --config retention.ms86400000 kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic order-auto-close-retry \ --partitions 3 --replication-factor 2 \ --config retention.ms86400000 kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic order-auto-close-dlq \ --partitions 3 --replication-factor 2 \ --config retention.ms604800000关于分区数我的建议是让这四个 topic 保持一致这样无论是按 key 哈希定位还是做消息流转追踪心智负担都会小很多。业务初期不确定分区数的话宁可设成 3 或 6 的倍数不要设 1因为后面想再扩分区虽然不难但会影响曾经消费过旧分区的消费者端数据再均衡逻辑。5.2 消费者模板与公共代码我把消费者处理模板封装成了一个公共方法业务处理器只需要实现最关键的处理逻辑异常路由和死信判定都在模板里统一完成public abstract class AbstractRetryableConsumer { private static final int MAX_RETRY 5; private final KafkaProducerString, String delayProducer; private final KafkaProducerString, String retryProducer; private final KafkaProducerString, String dlqProducer; protected abstract void doProcess(ConsumerRecordString, String record) throws Exception; protected boolean isRetryable(Throwable t) { return t instanceof RetryableException; } public void handle(ConsumerRecordString, String record) { try { doProcess(record); } catch (Throwable t) { if (!isRetryable(t)) { sendToDlq(record, NON_RETRYABLE, t); return; } int retryCount getHeaderInt(record, retry-count, 0); if (retryCount MAX_RETRY) { sendToDlq(record, MAX_RETRY_EXCEEDED, t); return; } sendToRetry(record, t, retryCount 1); } } private void sendToRetry(ConsumerRecordString, String record, Throwable t, int newCount) { long delaySec Math.min(60L * 5, (long) Math.pow(2, newCount) * 1000); long nextRetryTime System.currentTimeMillis() delaySec; ProducerRecordString, String retryRecord new ProducerRecord( record.topic() -retry, record.key(), record.value()); retryRecord.headers().add(retry-count, String.valueOf(newCount).getBytes()); retryRecord.headers().add(next-retry-time, String.valueOf(nextRetryTime).getBytes()); retryRecord.headers().add(last-error-class, t.getClass().getName().getBytes()); retryRecord.headers().add(last-error-msg, t.getMessage().getBytes()); retryRecord.headers().add(initial-offset, String.valueOf(record.offset()).getBytes()); retryProducer.send(retryRecord); } private void sendToDlq(ConsumerRecordString, String record, String reason, Throwable t) { ProducerRecordString, String dlqRecord new ProducerRecord( record.topic() -dlq, record.key(), record.value()); dlqRecord.headers().add(dlq-reason, reason.getBytes()); dlqRecord.headers().add(dlq-time, String.valueOf(System.currentTimeMillis()).getBytes()); dlqRecord.headers().add(original-topic, record.topic().getBytes()); dlqRecord.headers().add(original-partition, String.valueOf(record.partition()).getBytes()); dlqRecord.headers().add(original-offset, String.valueOf(record.offset()).getBytes()); dlqRecord.headers().add(error-stack, getStackTrace(t).getBytes()); dlqProducer.send(dlqRecord); } }实际业务消费者只需要继承这个模板实现doProcess和isRetryable就能自动获得快速失败进重试、超限进死信的能力。这个模式在团队里推广成本很低新同学接一个新业务 topic只需要写核心处理逻辑不容易把路由搞错。5.3 配置参数里需要重点关注的几项这套链路对 Kafka 生产者和消费者的配置要求比较高我这边有几个参数是特别留意的enable.idempotencetrue延迟转发、重试投递时网络抖动会导致 producer 重发幂等生产者能避免 broker 侧记录重复消息。acksall消息不能丢必须等所有副本确认。max.in.flight.requests.per.connection5配合幂等生产者可以在保证顺序的前提下保留一定的吞吐。auto.offset.resetearliest消费者组新启动时能从最早消息开始避免中间层一启动就漏掉延迟消息。max.poll.interval.ms设置合理值如果业务处理太慢消费者会被踢出组导致 rebalance。这个在最后那一章节还会重点提。另外延迟 topic 和重试 topic 的消费者不要用业务消费线程直接阻塞。原因很简单中间调度层要保证能被快速 poll如果它像业务消费者那样去等一个外部接口返回整个延迟调度链路都会被卡住。6. 实测中的坑延迟精度、重复消费、时钟波动前面讲的是怎么把功能做出来这一节讲的是上线之后才会发现的坑。我在踩这些坑的时候花费的排障时间甚至比写代码还多所以单独拎出来说。6.1 延迟精度别往毫秒上凑第一个要降低预期的是延迟精度。Kafka 的 producer 端有批量发送机制linger.ms哪怕设成 0网络层和 broker 端也可能因为批次原因产生几毫秒到几十毫秒的抖动。消费端poll一次也可能带回一批消息时间桶又是按秒整点触发的。所以整体跑下来秒级延迟是可以接受的毫秒级精度这套方案实现不了。这不只是精度问题更是成本问题。为了更高的精度你得缩小时间桶粒度比如改成 100ms 一个桶那样调度服务和 Kafka 的交互频率会直线上升吞吐反而下降。我通常把时间桶粒度控制在 1 秒业务侧绝大部分需求也都够用。6.2 重复投递与幂等消费延迟转发和重试机制有一个共同特点它们都会导致消息被处理不止一次。比如调度服务在转发时 producer 抛了一个网络异常但 broker 实际已经写入成功了重试发送就会形成重复消息再比如消费者快速重试到一半进程重启业务可能做到一半但消息并没有被 commit。所以在设计这套链路之初我就对业务消费端反复强调必须做幂等消费。最常用的手段是给消息体加一个业务唯一键比如订单号orderId消费者处理前先查一下状态-- 伪代码逻辑 if (order status already closed) { return; // 重复消息直接忽略 } try { update order set status closed where order_id xxx and status ! closed; } catch (DuplicateKeyException e) { return; // 并发重复已经处理过 }这种方式是把至少一次投递变成最终只生效一次的常见解法。Kafka 本身保证不了恰好一次幂等只能靠业务状态来保证。6.3 注意时钟波动和 Kafka 时间戳前面我在延时代码里特意用了record.timestamp()就是因为生产上出过一次事故。调度服务所在的机器时钟比业务服务慢了 5 秒结果所有延迟消息都提前或者延后了几秒触发对账系统因此对出来不少边缘订单。排查了很久最后定位到是因为我一开始用本地System.currentTimeMillis()计算到期时间导致的。如果对时间基准要求更高可以把到期时间计算逻辑改为以 broker 写入消息的 record.timestamp 为基准加上 delayMs。Kafka 的 timestamp 默认是 broker 收到消息时写入的所有消息的时间起点一致跟调度服务本地时钟无关。这样就算调度节点换了机器延迟基准也不会乱。6.4 死信处理时的一个隐蔽坑死信重投还有一个不容易发现的坑如果原业务 topic 已经设置了消息体结构升级旧版本的死信消息重投过去可能又因为字段不合法而再次瞬间进入死信。这样就会形成一个重投 - 立刻进死信的死循环人工看着像处理完了实际死信量一点没降。我现在的做法是死信重投不直接发到业务 topic而是发到业务 topic 的一个预检 topic由预检消费者做兼容性校验能处理就转到业务逻辑处理不了再退回死信。虽然不是所有业务都需要这一步但如果你的消息体是有版本演进的强烈建议加上。6.5 消费者阻塞和 rebalance 的纠缠这套链路里重试 topic 由调度服务统一管理业务消费者本身不应该再去消费 retry-topic。因为业务消费者一旦在处理一条消息时调用了长时间阻塞的外部接口超过max.poll.interval.ms默认 5 分钟还没 poll就会被判定为死亡触发 rebalance。rebalance 期间 broker 会重新分配分区如果刚好消费到一半的消息没有 commitrebalance 结束后又会被重新消费一次。更麻烦的是如果消费者频繁 rebalance整个消费者组的消费进度都会拖慢给人一种Kafka 变慢了的错觉。所以我把所有上游不可控的逻辑都从业务消费线程里剥离出去业务消费者只做两件事尝试处理处理失败就把消息交给重试或死信。任何超过 5 秒的外部调用都建议走异步化。整套链路跑下来我最大的感受是Kafka 确实不是为延迟场景设计的但只要中间调度层设计得足够收敛它反而比很多自带延迟功能的 MQ 更可控因为每个环节都能被监控、被重放、被审计。最后再分享一个小经验所有 topic 的保留时长和消息 key 的规则一定要在上线前一次性想清楚后面改分区数、改 key 策略都会牵连整套消费逻辑代价远比你想象的大。我打算后续把这套延迟、重试、死信的逻辑抽成一个公共组件让新业务通过配置就能接入等组件成型了再回来做一次完整分享。