我们团队在某个核心订单系统里被一个问题反复折磨下游支付回调有时候延迟有时候失败还有极端情况下消息根本没法处理。这些问题叠加在一起就是订单状态迟迟不变、用户投诉不断增加、人工后台天天有人翻数据。后来我们决定在现有 Kafka 技术栈上把延迟队列、重试队列、死信队列这三件套一次性补齐彻底把消息治理这件事做了个闭环。这也是我今天想跟你聊的内容基于 Kafka 怎么实现这三个队列为什么这么选以及落地过程中你会踩到哪些坑。这套方案适合有一年以上 Kafka 使用经验但还没系统做过消息治理的团队参考。我先说结论用 Kafka 实现延迟、重试、死信队列完全可行性能和可靠性都有保证但前提是你要理解 Kafka 的模型约束接受它不像某些消息中间件自带延迟等级、死信交换机需要我们自己用代码补上这一层。本文会把三个队列的语义拆清楚、给出可运行的方案然后把我在实际项目中遇到的典型问题一并记录下来。1. 先弄清楚三个队列到底解决什么问题我们做技术方案最忌讳直接跳到怎么实现结果写了一堆代码却没对准业务痛点。延迟、重试、死信这三类队列本质上对应的是消息处理生命周期里的三个典型困境消息不该现在被处理、消息暂时处理不了、消息永远处理不了。1.1 延迟队列解决不该现在执行的问题延迟队列最常见的场景就是订单超时未支付自动关闭。用户下单以后系统期望在比如 30 分钟后检查一次支付状态没支付就自动取消。如果这个消息立刻发给消费者消费者会立刻执行关闭逻辑那所有未支付的订单都会被秒关这显然不对。所以我们需要一个机制让这条消息等到 30 分钟后再被消费者看到。类似的场景还有直播开播前给订阅用户发送提醒提前 5 分钟触发优惠券过期前 24 小时发送即将到期的通知预约类的短信或站内信在指定时间点触达。这些都是延迟队列的典型业务模型。延迟队列的核心价值就是让消息能够按预期的时间点被消费而不是消费方收到就处理。需要注意的是延迟队列不等于定时任务。定时任务通常是遍历数据然后执行批量操作延迟队列则是单个消息级别的调度粒度更细耦合度更低消息本身就是任务单元。1.2 重试队列解决暂时处理不了的问题重试队列服务的场景非常现实消费者从 Kafka 拉取消息后需要调用下游的订单服务、库存服务、风控接口或者第三方支付平台。这些接口不可能永远稳定总会有超时、限流、返回 5xx 的时候。如果一遇到接口异常就丢弃消息结果就是数据丢失订单状态在多个系统之间长期不一致。如果同一线程原地重试几次再放弃又可能阻塞消费速度导致整个消费者组积压。更麻烦的是有些下游失败是短暂的比如网络抖动、数据库连接池满了等 30 秒再试就能恢复但原地阻塞等待又会让消费线程干等。重试队列要解决的核心问题就是把失败后重新尝试这件事从消费主链路中搬出来。消费失败后消息不是被丢弃而是进入一个重试专用的通道等待指定的间隔后被再次提交消费。这样下游抖动不会影响消费主场景的速度消息也不会被轻易放弃。1.3 死信队列解决永远处理不了的问题死信队列是最后一道兜底。消息可能因为各种原因处理失败反序列化失败、消息结构不合法、业务规则已经改变导致无法完成处理、或者重试了 N 次仍然失败。这种情况下如果继续在正常队列或重试队列里死循环只会白白消耗资源、堆积日志、拖延后续消息的处理。这时候就应该把消息判死刑投递到一个单独的死信队列中。死信队列里的消息不应该被常规消费者继续重试而应该由专门的人工补偿程序去读取、分析、修复或者触发告警让值班同学介入。死信队列的价值在于既保留了一条消息的所有上下文又避免它干扰正常链路。我们从死信队列里能查到这条消息来自哪个原始业务、失败原因是什么、重试了多少次、原始 offset 是多少这就是问题排查和系统治理的依据。这三个队列组合在一起才构成一套完整的消息处理可靠方案。延迟队列控制执行时机重试队列容忍瞬时故障死信队列兜底不可恢复的异常三者缺一不可。2. 为什么选 Kafka 而不是专用消息中间件我知道聊到延迟队列、死信队列很多人第一反应是 RabbitMQ 的 TTL 死信交换机或 RocketMQ 的延迟等级。今天拿 Kafka 来做不是因为 Kafka 在这些功能上开箱即用而是技术选型从来不是单项功能对比还要看现有技术栈、运维能力和迁移成本。2.1 几种常见方案的横向对比先把 RabbitMQ、RocketMQ 和 Kafka 在这三个需求上的表现拉出来说说。RabbitMQ 方案RabbitMQ 天然支持消息 TTL 和死信交换机DLX。消息设置过期时间后如果没被消费会进入死信交换机再路由到另一个队列。这在语义上非常贴合延迟和死信场景很多团队早期用它实现延迟队列。缺点是 RabbitMQ 的吞吐量相对 Kafka 低不少消息堆积能力也有限大量消息积压时内存压力明显。而且 RabbitMQ 的延迟基于 TTL同一队列里消息若过期时间不同会出现队头阻塞问题。RocketMQ 方案RocketMQ 原生支持延迟等级比如 1s、5s、10s、30s、1m到最大几个小时的固定等级。它实现得非常优雅生产者指定延迟等级Broker 端会自动做延迟投递。同时也支持事务消息对重试和死信也有一套半内置的方案。但 RocketMQ 和 Kafka 不是同一个生态如果团队基础设施已经全面围绕 Kafka 建立引入 RocketMQ 意味着额外的集群、监听和运维成本。Kafka 方案Kafka 原生并不提供上述任何能力它只有一个追加日志模型生产者写入消费者按 offset 顺序拉取。但 Kafka 有高吞吐、海量堆积、分区顺序、消费组水平扩展这些能力。延迟、重试、死信都可以在上层用代码实现。我们最终选择 Kafka是因为团队不想为了这几个队列特性引入一套新的中间件而是在现有 Kafka 集群上用多 Topic 加代码调度来补全这三个能力。我用一个表格把选型关键点列出来方便你结合自己的处境判断。对比项RabbitMQRocketMQKafka延迟消息TTLDLX存在队头阻塞内置延迟等级需自研调度死信队列内置 DLX半内置需自建 Topic重试机制需手动设计支持需自建 Topic吞吐量中较高最高堆积能力一般较好极强运维成本独立集群独立集群复用现有集群消息回溯有限有限按 offset 灵活回溯如果你所在团队已经有 Kafka 集群并且 Kafka 承载着绝大多数业务消息那在 Kafka 上自研这三个功能远比再引入一套新中间件划算。2.2 自研方案的总体架构思路在 Kafka 上实现延迟、重试、死信整体思路其实非常统一用多个 Topic 分别承担不同的语义然后用生产者或消费者代码控制消息的流转路径。具体来说我会在系统里规划这样几类 Topic延迟专用 Topic存放当前还不该被处理的消息消息体里带上期望执行的时间戳。由一个延迟调度消费者负责扫描到期就转发到真正的业务处理 Topic。业务处理 Topic消费者真正处理业务的地方也是消息的主战场。重试 Topic存放处理失败但还有机会恢复的消息可能按不同的延迟档位拆分成多个 Topic比如 5 秒、30 秒、5 分钟。死信 Topic存放确认无法处理的消息。这里要特别说明一点消息在不同的 Topic 之间流转往往是靠一个小的转发逻辑来完成的。比如延迟 Topic 扫描到一条到期消息不是直接让原本的延迟消费者去执行业务而是把消息投递到业务处理 Topic。这样做的核心好处是职责分离延迟调度器只负责定时业务消费者只负责处理两边的并发配置、消费速率、监控告警可以独立设置互不干扰。我之前见过一个反面案例团队把延迟消息直接发到一个消费者里消费者拉取后判断时间未到就 Thread.sleep结果消费者线程全部阻塞在 sleep 上后续消息全被卡住消费者提交 offset 超时最终整个消费组重平衡消息重复消费一大片。这就是没有把定时和处理解耦造成的灾难。所以我们的设计里延迟到期后一律投递到独立的业务处理 Topic。2.3 Topic 规划与消息结构设计在动手写代码之前先把 Topic 规划好。我建议每个业务按独立前缀建 Topic避免业务之间互相干扰。举个具体例子订单超时关闭场景订单延迟 Topictopic-order-delay。生产者把下单消息扔进来消息体里带目标执行时间。订单处理 Topictopic-order-handler。真正执行订单关闭逻辑的地方。订单重试 Topic比如 topic-order-retry-30stopic-order-retry-5m按重试间隔拆分。订单死信 Topictopic-order-dead-letter。确认无法处理的订单消息最终落在这里。消息体格式上我强烈建议从一开始就设计好不要等到上线后再补。用 JSON 序列化比较通用但一定要预留下几个关键字段。字段含义示例msgId全局唯一消息 ID用于幂等订单号 时间戳生成的 UUIDbizType业务类型ORDER_CLOSEpayload原始业务数据订单 ID、用户 ID 等executeAt期望执行时间戳毫秒1735722600000retryCount已重试次数2sourceTopic原始来源 Topictopic-order-delaysourceOffset原始 offset便于回溯10234errorMsg最近一次失败原因下游库存接口超时字段不在多够用就行。但 retryCount 和 sourceTopic、sourceOffset 这三个字段你一定留好一旦消息走到死信你反查源头就要靠它们。3. 延迟队列的两种落地姿势Kafka 本身不提供延迟消息能力我们无非是通过 Topic 模型、调度扫描、或者中转投递来模拟。这里我把最实用的两种方案展开讲你根据业务体量和延迟精度需求选择。3.1 方案一单一延迟 Topic 加定时轮询这个方案最容易理解生产者把消息发到延迟 Topic消费者启动一个定时任务每隔几秒拉取一批消息检查消息体里的 executeAt 是否已经小于等于当前时间。如果到期就把消息转发到业务处理 Topic如果没到期就丢到下一个轮询周期再处理。实现层面可以使用 KafkaConsumer 的 poll 语义但不要用原生 poll 去逐条 sleep 等待。我的做法是写一个独立的延迟调度消费者它常规拉取消息把消息放入一个本地优先级队列按 executeAt 排序再有一个单独的调度线程触发到期消息的转发。这里要给一个具体的参数建议轮询间隔要根据业务对延迟精度的要求来定。如果只是订单超时关闭这种分钟级精度调度线程每分钟扫描一次就够了根本不需要秒级精确。如果是短信定时发送这类秒级触达要求的场景我建议轮询间隔控制在 1 秒左右不要低于 500 毫秒否则空转成本太高。这个方案的优点是实现简单Topic 数量少运维成本低适合延迟消息量不大、精度要求不高的场景。缺点也很明显如果延迟消息数量特别大调度消费者每次要遍历大量未到期消息无效遍历会占用带宽和 CPU。另外如果消费者重启本地优先级队列里的消息会丢失需要借助 Kafka 自身的 offset 提交机制来确保消息不丢但具体实现相对繁琐。我实际测过这个方案在消息量低于每秒几千条、延迟精度秒级可接受的情况下效果已经很稳。如果量级再大建议直接上方案二。3.2 方案二多 Topic 延迟槽加定时投递方案一的核心问题是大数据量下的无效扫描。方案二换个思路不再把所有延迟消息放在一个 Topic 里统一扫描而是按延迟范围拆分成多个延迟槽每个槽对应一个 Topic 和一台调度器。举例来说我们建这样一组 Topicdelay-5s、delay-30s、delay-1m、delay-30m、delay-1h、delay-24h。生产者发送消息时根据业务需要的延迟时间路由到最接近但又不小于目标时间的那个 Topic。对应每个 Topic有一个独立的调度消费者它专门负责这一档延迟的消息。比如 delay-5s 的调度消费者每 1 秒拉取一次快速投递delay-24h 的调度消费者每 30 分钟甚至每小时拉取一次没必要频繁空转。这种设计的优势是显而易见的每个 Topic 的消息延迟范围收敛调度器空转率大幅降低消息量天然被分片扩展性也好。如果某一档消息量激增可以只针对这一档增加调度消费者实例不影响其他档位。代价是 Topic 数量变多生产端路由逻辑需要维护一个延迟档位表。另外延迟档位是离散的如果业务要求的延迟时间不匹配档位我们需要约定向上匹配策略。举个例子业务要延迟 3 秒就投递到 delay-5s要延迟 10 分钟就投递到 delay-30m。这里会有一点误差但大多数业务场景并不需要精确到毫秒。更接近 RocketMQ 延迟等级的做法就是把档位表做成配置生产端计算目标延迟时间后在配置表里找最小且不小于目标档位的值。RocketMQ 提供的常见档位是 1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h这个表可以直接借鉴。3.3 延迟精度、消息积压与性能取舍关于延迟队列有一个最容易被忽略却极其重要的点你设置的延迟精度直接决定了调度器收益和集群负载之间的平衡。如果业务允许分钟级误差你就不该把调度器的轮询周期压到 1 秒否则大量空 poll 白白消耗 CPU。我们实际线上的做法是分档调优秒级档位的轮询周期是 1 秒分钟级档位的轮询周期是 30 秒小时级档位的轮询周期是 5 分钟。你算一笔账delay-24h 这个 Topic 一天可能只有几千条消息如果调度器每 1 秒空转一次一天的空转次数高达 86400 次一大半是无用功。把扫描周期拉长到 5 分钟收益几乎是零损耗的。消息积压也要单独说。延迟 Topic 的消息本身在 Kafka 中就是长时间不消费的状态这会被 Kafka 记录为消费滞后Lag。这个 Lag 是正常的但你要在监控系统里把它和真正的异常积压区分开。我们的做法是对不同的 Topic 配置各自的 Lag 告警阈值延迟 Topic 的告警阈值显著放宽只监控有没有消费者存活而不是offset 有没有落后。你可以在调度转发时增加一个条件判断消息在延迟 Topic 里已经停留了多久。如果一条消息的停留时间远超预期比如 delay-5s 的消息 30 分钟都没被转发出去多半是调度消费者挂了或者下游处理 Topic 消费积压导致无法写入。这时触发告警比单纯看 Lag 指标更能定位问题。4. 重试队列让失败的消费请求有序重来重试队列的设计核心不是简单地把消息发回原 Topic 再消费一遍而是要设计一套有节奏、有上限、有止损的重试机制。4.1 重试策略设计间隔、次数与动态退避重试策略的设计直接决定系统在下游故障时的表现。固定的失败后立即重试没有意义因为下游刚刚报错大概率还处在故障状态立刻再试反而容易加重下游压力。业界通用的是指数退避策略第一次重试等 5 秒第二次等 30 秒第三次等 5 分钟最多重试 5 次。每一次重试的间隔都在变大给下游充分的恢复时间。具体档位可以根据业务敏感度调整。我提供一个通用的配置模板重试次数延迟间隔对应重试 Topic15 秒topic-retry-5s230 秒topic-retry-30s35 分钟topic-retry-5m430 分钟topic-retry-30m52 小时topic-retry-2h这里有个实践细节与其让消息在同一个重试 Topic 中反复消费、反复失败、反复更新 retryCount不如拆分为多个重试 Topic分别对不同档位的消息进行处理。这样做的直接好处是不同间隔的重试彼此隔离某档重试 Topic 积压了不会影响其他档位而且消费线程可以根据消息所在 Topic 确定性地知道该等多久不用从消息体里解析重试次数再计算 sleep 时间。4.2 基于手动 Ack 加重试 Topic 的消费模型Kafka 默认是自动提交 offset也就是说消费者只要 poll 到消息offset 就会往后移动哪怕业务逻辑抛异常这条消息也不会再被消费到。所以在重试场景中必须关掉自动提交改为手动提交。我的实践模型是这样一个循环流程第一步业务消费者从处理 Topic 拉取消息执行业务逻辑。第二步如果执行成功提交 offset消息生命周期结束。第三步如果执行失败先判断异常类型。如果是参数错误、数据不合法这类不可重试异常直接投递到死信 Topic然后提交 offset。如果是下游超时、限流这类可重试异常读取消息体里的 retryCount如果已经超过最大重试次数投递到死信 Topic否则把 retryCount 加 1投递到对应的重试 Topic然后提交当前 offset。为什么失败后还要提交 offset因为消息已经被处理过了虽然处理失败但它被转移到了重试通道不再属于当前消费流程的处理范围。如果不提交该消息会反复出现在当前消费者里造成无限循环。这一点是新手最容易踩的坑。重试 Topic 的消费者则按各自的间隔策略去消费重新调用业务处理逻辑。注意重试 Topic 消费者的业务处理逻辑和主处理 Topic 消费者的业务处理逻辑应当保持一致不能重写一份有出入的处理方法。我们通常把核心处理方法抽成一个公共类两边都调用同一个方法避免主链路成功、重试链路失败这类判断分叉问题。4.3 幂等性重试之前必须解决重复消费重试机制引入后消息被消费多次是必然事件而且是设计如此。如果业务方法本身不具备幂等性重试就是一个定时炸弹。以订单关闭场景为例原本订单状态是待支付关闭后变成已关闭。如果关闭动作被重复执行第二次执行时订单已经是已关闭再关一次理论上无害。但如果你的处理方法里有金额扣减、库存回滚、发送通知这类副作用操作重复执行就会造成严重问题。幂等的最简做法是为每条消息生成全局唯一 msgId在处理前用 Redis 或数据库记录这个 msgId 的处理状态。处理逻辑先尝试写入状态处理中如果发现已经处理过就直接跳过。如果是数据库操作可以在订单表上加一个 lastProcessMsgId 字段或者用业务表自身的唯一约束来实现幂等。我建议在重试场景中一定要用先记录、后处理的顺序不要先处理再记录否则处理过程中崩溃重试时无法判断是否已执行。5. 死信队列最后的兜底与排查手段很多团队把死信队列当成垃圾桶只管往里扔从来不看。这是非常危险的做法。死信队列的价值在于它是一条蛛丝马迹链能帮你定位线上问题。要让死信队列真正发挥作用必须在设计上做到两点消息内容足够完整、消费和告警链路足够完善。5.1 什么样的消息该进死信不是所有失败消息都应该进死信死信是止损而不是甩锅。我的分类标准是三道闸门。第一道闸门不可重试的异常。消息反序列化失败、JSON 解析失败、业务 ID 缺失、枚举值不合法这类问题不管你重试多少次结果都一样。它们应该在第一次失败时就进死信节省所有后续无意义的消费。第二道闸门重试次数耗尽。消息本身没有问题只是下游一直不可用或者数据依赖不满足。这类消息在走完所有重试档位后如果仍然失败就应该进死信不能无限重试。无限重试的后果是消息永远在系统里流转占用资源而且真正的故障原因反而不容易暴露出来。第三道闸门业务规则不再满足的场景。比如一条优惠券发放消息在重试期间用户已经注销了账号这时候即使重试成功也没有意义。这类过期业务放在死信中由人工或自动补偿程序判断如何处理。5.2 死信消息的字段设计与转储死信消息的内容会直接决定排查效率。我见过很多团队的死信 Topic 里只有原始 payload没有来源、没有原因出了线上问题根本没法查。要解决这个问题投递死信时必须把关键链路信息补全。实际项目中我会在死信消息体里保留原始业务 payload一条消息最核心的数据。sourceTopic 和 sourceOffset记录它从哪个 Topic 的哪个位置来配合 Kafka 的 offset 查询 API 可以找到原始上下文。errorMsg最近一次异常堆栈信息。retryCount累计重试次数用于判断失败是否在合理次数内。deadLetterReason进死信的原因分类比如 RETRY_EXHAUSTED、DESERIALIZE_FAILED、BIZ_RULE_EXPIRED。最后投递时间方便值班人员判断这条消息是什么时候进入死信的。这些字段不是拍脑袋定的而是我在实际排障过程中反推出来的。有一次线上出现一批订单消息进入死信因为没记录 sourceTopic我们花了一个多小时才定位到是某个晚上的发布变更导致消息体字段变更而不是消费逻辑故障。从那以后强制要求所有死信消息必须携带这些字段。5.3 死信 Topic 的监控与人工处理链路死信队列如果只是存着等通知很容易变成无人区。我建议建立这样一套监控和处理机制死信 Topic 的消息数量作为核心告警指标。正常情况下一个健康的业务死信量应该接近于零。如果短时间内大量消息涌入死信 Topic一定是业务链路出了问题比如消息体字段被改坏了、下游接口大面积故障、或者代码里抛了不该抛的异常。告警阈值可以根据业务量设置比如单日超过 100 条就触发 P1 告警。死信消费者只做两件事转储和分析。它把死信消息从 Kafka 中读出写入 ES 或数据库方便按 msgId、业务类型、时间范围检索同时生成死信报表统计死信原因分布。注意死信消费者不应该自动把消息发送回原 Topic 重试这会掩盖问题也可能导致死信队列和重试队列之间无限循环。人工处理方面我建议配置一个简易的补偿管理后台运营或开发人员搜到具体 msgId 后能查看死信详情、标记处理状态、触发一次人工重放。重放时可以直接调用业务公共处理类而不是发送回 Kafka。这能避免重放过程中再次进入死信循环。6. 完整实现从生产者到消费者分别怎么写概念讲得再多不如一段能跑的代码来得实在。我基于 Java 和原生 Kafka 客户端给出一个最小可运行实现你把它套进自己的业务骨架里稍微改改就能用。6.1 生产端代码发送延迟与优先级消息生产者这个环节并不复杂核心是给消息设置一个足够全局唯一的 msgId并带上 executeAt 时间戳。这里我直接展示投递延迟消息的代码。public class DelayMessageProducer { private static final String DELAY_TOPIC topic-order-delay; private final KafkaProducerString, String producer; public DelayMessageProducer() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.1.10:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.ACKS_CONFIG, all); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); this.producer new KafkaProducer(props); } public void sendDelayMessage(String bizType, String payload, long delayMillis) { long executeAt System.currentTimeMillis() delayMillis; String msgId UUID.randomUUID().toString(); DelayMessage message new DelayMessage(); message.setMsgId(msgId); message.setBizType(bizType); message.setPayload(payload); message.setExecuteAt(executeAt); message.setRetryCount(0); ProducerRecordString, String record new ProducerRecord( DELAY_TOPIC, bizType, JSON.toJSONString(message)); producer.send(record, (metadata, exception) - { if (exception ! null) { // 生产失败要记录日志并考虑本地重试或落盘 log.error(延迟消息发送失败: {}, msgId, exception); } }); } }关于参数设置我要多说一句acks 必须设成 allenable.idempotence 必须打开。这不是可选项而是为了保证消息不因网络问题而丢失或者重复。Kafka 的幂等生产者能解决重试导致消息重复的问题加上 msgId 的幂等消费设计两头都守住才能做到端到端的不丢不重。6.2 延迟轮询消费者代码定时扫描与到期投递延迟消费者的代码是这个方案里最关键的一环。我用优先级队列方案实现单一 Topic 轮询逻辑上比较直观也方便做本地调度。public class DelayQueueConsumer { private static final String DELAY_TOPIC topic-order-delay; private static final String HANDLER_TOPIC topic-order-handler; private final KafkaConsumerString, String consumer; private final KafkaProducerString, String producer; private final PriorityBlockingQueueDelayMessage pendingQueue new PriorityBlockingQueue(1000, Comparator.comparingLong(DelayMessage::getExecuteAt)); private final ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); // 是否停止标志 private volatile boolean running true; public DelayQueueConsumer() { Properties props new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.1.10:9092); props.put(ConsumerConfig.GROUP_ID_CONFIG, delay-handler); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); this.consumer new KafkaConsumer(props); this.producer new KafkaProducer(getProducerProps()); } public void start() { Thread pollThread new Thread(this::pollLoop, delay-poll-thread); pollThread.start(); scheduler.scheduleAtFixedRate(this::dispatchExpiredMessages, 0, 1, TimeUnit.SECONDS); } private void pollLoop() { consumer.subscribe(Collections.singletonList(DELAY_TOPIC)); while (running) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(500)); for (ConsumerRecordString, String record : records) { DelayMessage msg JSON.parseObject(record.value(), DelayMessage.class); pendingQueue.offer(msg); // 记录一条待处理的延迟消息 pendingMessageStore.put(msg.getMsgId(), msg); } consumer.commitSync(); } } private void dispatchExpiredMessages() { long now System.currentTimeMillis(); ListDelayMessage readyMessages new ArrayList(); DelayMessage msg; while ((msg pendingQueue.peek()) ! null msg.getExecuteAt() now) { pendingQueue.poll(); readyMessages.add(msg); } for (DelayMessage readyMsg : readyMessages) { // 转发到业务处理 Topic producer.send(new ProducerRecord(HANDLER_TOPIC, readyMsg.getBizType(), JSON.toJSONString(readyMsg)), (metadata, exception) - { if (exception ! null) { log.error(到期消息转发失败: {}, readyMsg.getMsgId(), exception); // 转发失败需要告警不要静默丢弃 } else { pendingMessageStore.remove(readyMsg.getMsgId()); } }); } producer.flush(); } }这里有一个非常关键的代码细节poll 到消息后立即 commitSync还是把消息放入优先级队列并转发成功后再提交我的建议是poll 到消息、放入本地队列后立刻提交 offset。因为此时消息还在我们本地的 pendingQueue 里并未真正投递到下游。即使 Kafka offset 已经提交、进程突然崩溃本地队列中的消息会丢失但我们可以通过配置 pendingMessageStore比如 Redis 或本地磁盘来保存延迟消息快照重启后从快照恢复。如果你不想引入额外的快照存储那就不要过早提交 offset而是等消息转发成功后再提交代价是重复消息可能性增加需要配合 msgId 幂等消费。两种方案各有利弊。如果消息量不大、业务允许偶发重复我推荐等到转发成功再提交 offset如果消息量大、需要精确控制建议引入 Redis 作为消息快照存储。6.3 业务处理与重试逻辑代码判断异常走向业务消费者负责真正的订单关闭逻辑也是重试和死信分流发生的地方。我把核心处理流程抽成一个公共方法主消费和重试消费都调用它。public class OrderHandlerConsumer { private static final int MAX_RETRY_COUNT 5; private final MapInteger, String retryTopicMap buildRetryTopicMap(); private final KafkaConsumerString, String consumer; private final KafkaProducerString, String producer; public void handleMessage(DelayMessage message) { try { // 1. 先做幂等检查 if (idempotentService.isProcessed(message.getMsgId())) { return; } // 2. 执行业务处理关闭订单、更新状态等 orderService.closeOrder(message.getPayload()); // 3. 记录处理成功的消息 ID idempotentService.markProcessed(message.getMsgId()); // 4. 提交 offset 由外层统一管理 } catch (BizException e) { // 不可重试异常直接进死信 sendToDeadLetter(message, e.getMessage(), BIZ_EXCEPTION); } catch (Exception e) { // 可重试异常判断次数 int retryCount message.getRetryCount() 1; if (retryCount MAX_RETRY_COUNT) { sendToDeadLetter(message, e.getMessage(), RETRY_EXHAUSTED); } else { message.setRetryCount(retryCount); sendToRetryTopic(message); } } } private void sendToRetryTopic(DelayMessage message) { // 根据 retryCount 选择对应的重试 Topic String retryTopic retryTopicMap.get(message.getRetryCount()); producer.send(new ProducerRecord(retryTopic, message.getBizType(), JSON.toJSONString(message))); } private void sendToDeadLetter(DelayMessage message, String errorMsg, String reason) { message.setErrorMsg(errorMsg); message.setDeadLetterReason(reason); producer.send(new ProducerRecord(DEAD_LETTER_TOPIC, message.getBizType(), JSON.toJSONString(message))); } }这里我想强调一下幂等检查的位置。很多团队把幂等检查放在外层消费者里这没有错但更重要的是要把检查逻辑和业务本身融合先查数据库中的订单状态如果已经是已关闭就直接跳过如果业务状态和预期不符同样跳过。只靠 Redis 里记 msgId 是不够的因为消息可能来自不同的触发路径比如人工补偿后台触发的任务和原始重试触发的任务携带的 msgId 不同但操作的是同一笔订单。这个时候业务状态本身的幂等才是最终兜底。6.4 关键参数与配置清单代码之外Kafka 的参数配置同样决定成败。我整理一份我常用的参数清单你可以对照检查自己的配置。组件参数推荐值说明Produceracksall必须保证消息写入所有副本Producerenable.idempotencetrue防止生产者重试导致重复Producerlinger.ms5-10适当增加以提升批量吞吐Producermax.in.flight.requests.per.connection5配合幂等确保顺序Consumerenable.auto.commitfalse必须手动管理 offsetConsumermax.poll.records50-200避免单次拉取过多导致处理超时Consumermax.poll.interval.ms300000防止业务耗时过长触发重平衡Brokerretention.ms至少 7 天延迟消息需要长留存便于回溯Brokermessage.max.bytes按最大消息体调整避免大消息写入失败有一个参数我必须单独提醒max.poll.interval.ms。默认值是 5 分钟如果消费者在处理一条消息时耗时超过这个时间消费者就会被判定为挂掉从而触发 Rebalance。在重试场景里业务方法可能调用下游接口接口超时如果设置很长很容易超过这个阈值。我的习惯是把该值设置为 10 分钟以上同时把业务处理的超时时间控制在 5 分钟以内给两者之间留出足够的安全余量。7. 常见问题与排查技巧实录理论推到这你一定想知道实际运行中都会遇到哪些问题。我把过去踩过的坑整理成一份速查表每一类问题后面都附上定位思路和解决方案。7.1 消息延迟不准总是提前或延后延迟不准是这类方案最容易被吐槽的问题原因大概率不在 Kafka 本身而在调度逻辑。一种情况是提前消费。业务要求延迟 30 分钟生产者投递到了 delay-1h 对应的 Topic等于提前了半小时。这种情况要检查生产端的档位匹配逻辑我们约定的是向上匹配即目标延迟小于等于该档位就投递到该档位。如果业务方传进来的延迟值是 45 分钟应该投递到 delay-1h不是 delay-30m。匹配逻辑要是写反了就会导致大量消息提前执行。另一种情况是延后消费。比如 delay-5s 档位的调度器每 30 秒才扫描一次那消息实际延迟可能是 35 秒。解决方法是检查该档位调度器的扫描周期是否和延迟精度匹配。秒级档位必须用秒级扫描周期否则延迟误差会让人抓狂。还有一类隐藏问题调度消费者所在机器和 Kafka 集群之间的时钟同步。如果调度器机器比 Kafka 慢 10 秒那执行时间就会晚 10 秒。建议所有服务统一接入时间同步这问题不大但容易被忽略。7.2 消息积压且在多个 Topic 间反复流转我在线上见过最恐怖的现象是某条消息一直在重试 Topic 和死信 Topic 之间转圈怎么都处理不完还触发了一堆告警。最终定位到的原因是异常分类写错了——把业务规则冲突这种不可重试异常错划分到了可重试的分类里导致每次重试都失败失败再重试直到重试次数耗尽进入死信死信消费者又因为告警配置不当反复触发人工重放。要避免这种问题必须先给异常分类画一条清晰的分界线哪些异常是可恢复的哪些是不可恢复的。不可恢复的异常比如数据校验失败、状态机不合法绝不进入重试队列。可以在公共异常类上增加一个标记接口比如 RetryableException 和 NonRetryableException改造起来比在 catch 块里一长串 if else 要省心得多。积压问题本身也值得警惕。如果重试 Topic 的 Lag 暴涨说明下游服务可能已经故障一段时间。此时要抑制重试生产者的投递速度避免故障下游被打爆。我的做法是给重试消费者增加一个简单的熔断开关连续失败率达到阈值时暂停消费重试 Topic并触发告警人工介入确认下游恢复后再打开开关。7.3 offset 提交异常导致消息丢或重复消息丢失和消息重复是 Kafka 消费场景两个绕不开的极端。丢失通常发生在自动提交下业务异常后没有处理重复通常发生在手动提交 offset 时先处理业务后提交 offset处理结束、offset 未提交时消费者挂掉重启后该消息又被拉到。坦白讲在 Kafka 的至少一次语义下重复消费是不可能从底层根除的我们只能靠幂等来消解后果。真正要谨慎避开的是丢消息处理成功后还没来得及提交 offset这不是问题因为会重复消费但处理失败后你提交了 offset消息就真的丢了。所以在业务逻辑里我的铁律是如果业务处理成功先执行所有必要的副作用操作再提交 offset。如果业务处理失败先把消息转移到重试 Topic 或死信 Topic再提交 offset。只要发生转移动作消息总会在某个 Topic 里存在不会丢失。这个顺序看起来简单但在并发场景里代码一旦写成先 commit 再处理或失败直接报错不转移问题就会立刻出现。7.4 延迟消息量大时 , 单线程调度器顶不住方案二的扩展性比方案一好但即使方案二单个 Topic 的调度器在极端情况下也可能成为瓶颈。比如整点大促时秒级档位会瞬间涌入大量延迟消息。遇到这种情况不要想着把一个消费者做成多线程就完事Kafka 的单分区只能被单线程消费。正确做法是按分区扩展延迟 Topic 设置多个分区调度消费者组设置多个实例每个实例负责一个或多个分区的延迟扫描。分发规则上建议按 bizType 或 msgId 取模保持同一个业务的消息集中在同一个分区方便局部顺序处理。如果消息量实在大到 Kafka 消费者组都无法承载那么延迟调度这一层可以拆出去做成独立服务延迟消息写入 Kafka 后由一个独立的延迟计算引擎把消息分配到内存时间轮到期后再批量回调 Kafka 生产端写入处理 Topic。这个方案是下一档的架构升级日常业务一般用不到但你要知道扩展路径是什么。最后分享一个实操中的小技巧关于整个链路最后想分享一个我们内部一直在用的经验给每一条消息都增加一个系统级标签字段比如 sysTag默认值是空字符串。当消息经历过重试、死信、人工重放之后每经过一跳就在这个字段上追加一跳标识比如 RETRY_1、DLQ、REPLAY。排查问题时只需要看这一个字段整条消息的流转路径就一目了然配合日志链路能快速定位到底是哪一环出了问题。这个技巧没有技术门槛但它在排查复杂跨系统问题时帮了我们大忙。消息治理的本质不是把消息变成一条单行道而是让每一次流转都有迹可循。把这条字段加上之后你会发现自己排查问题时的心态会稳很多。 
长河君 · 2026/10/17 20:50:13 · 阅读 1.2k Kafka消息治理实战：延迟队列、重试队列与死信队列实现
我们团队在某个核心订单系统里被一个问题反复折磨下游支付回调有时候延迟有时候失败还有极端情况下消息根本没法处理。这些问题叠加在一起就是订单状态迟迟不变、用户投诉不断增加、人工后台天天有人翻数据。后来我们决定在现有 Kafka 技术栈上把延迟队列、重试队列、死信队列这三件套一次性补齐彻底把消息治理这件事做了个闭环。这也是我今天想跟你聊的内容基于 Kafka 怎么实现这三个队列为什么这么选以及落地过程中你会踩到哪些坑。这套方案适合有一年以上 Kafka 使用经验但还没系统做过消息治理的团队参考。我先说结论用 Kafka 实现延迟、重试、死信队列完全可行性能和可靠性都有保证但前提是你要理解 Kafka 的模型约束接受它不像某些消息中间件自带延迟等级、死信交换机需要我们自己用代码补上这一层。本文会把三个队列的语义拆清楚、给出可运行的方案然后把我在实际项目中遇到的典型问题一并记录下来。1. 先弄清楚三个队列到底解决什么问题我们做技术方案最忌讳直接跳到怎么实现结果写了一堆代码却没对准业务痛点。延迟、重试、死信这三类队列本质上对应的是消息处理生命周期里的三个典型困境消息不该现在被处理、消息暂时处理不了、消息永远处理不了。1.1 延迟队列解决不该现在执行的问题延迟队列最常见的场景就是订单超时未支付自动关闭。用户下单以后系统期望在比如 30 分钟后检查一次支付状态没支付就自动取消。如果这个消息立刻发给消费者消费者会立刻执行关闭逻辑那所有未支付的订单都会被秒关这显然不对。所以我们需要一个机制让这条消息等到 30 分钟后再被消费者看到。类似的场景还有直播开播前给订阅用户发送提醒提前 5 分钟触发优惠券过期前 24 小时发送即将到期的通知预约类的短信或站内信在指定时间点触达。这些都是延迟队列的典型业务模型。延迟队列的核心价值就是让消息能够按预期的时间点被消费而不是消费方收到就处理。需要注意的是延迟队列不等于定时任务。定时任务通常是遍历数据然后执行批量操作延迟队列则是单个消息级别的调度粒度更细耦合度更低消息本身就是任务单元。1.2 重试队列解决暂时处理不了的问题重试队列服务的场景非常现实消费者从 Kafka 拉取消息后需要调用下游的订单服务、库存服务、风控接口或者第三方支付平台。这些接口不可能永远稳定总会有超时、限流、返回 5xx 的时候。如果一遇到接口异常就丢弃消息结果就是数据丢失订单状态在多个系统之间长期不一致。如果同一线程原地重试几次再放弃又可能阻塞消费速度导致整个消费者组积压。更麻烦的是有些下游失败是短暂的比如网络抖动、数据库连接池满了等 30 秒再试就能恢复但原地阻塞等待又会让消费线程干等。重试队列要解决的核心问题就是把失败后重新尝试这件事从消费主链路中搬出来。消费失败后消息不是被丢弃而是进入一个重试专用的通道等待指定的间隔后被再次提交消费。这样下游抖动不会影响消费主场景的速度消息也不会被轻易放弃。1.3 死信队列解决永远处理不了的问题死信队列是最后一道兜底。消息可能因为各种原因处理失败反序列化失败、消息结构不合法、业务规则已经改变导致无法完成处理、或者重试了 N 次仍然失败。这种情况下如果继续在正常队列或重试队列里死循环只会白白消耗资源、堆积日志、拖延后续消息的处理。这时候就应该把消息判死刑投递到一个单独的死信队列中。死信队列里的消息不应该被常规消费者继续重试而应该由专门的人工补偿程序去读取、分析、修复或者触发告警让值班同学介入。死信队列的价值在于既保留了一条消息的所有上下文又避免它干扰正常链路。我们从死信队列里能查到这条消息来自哪个原始业务、失败原因是什么、重试了多少次、原始 offset 是多少这就是问题排查和系统治理的依据。这三个队列组合在一起才构成一套完整的消息处理可靠方案。延迟队列控制执行时机重试队列容忍瞬时故障死信队列兜底不可恢复的异常三者缺一不可。2. 为什么选 Kafka 而不是专用消息中间件我知道聊到延迟队列、死信队列很多人第一反应是 RabbitMQ 的 TTL 死信交换机或 RocketMQ 的延迟等级。今天拿 Kafka 来做不是因为 Kafka 在这些功能上开箱即用而是技术选型从来不是单项功能对比还要看现有技术栈、运维能力和迁移成本。2.1 几种常见方案的横向对比先把 RabbitMQ、RocketMQ 和 Kafka 在这三个需求上的表现拉出来说说。RabbitMQ 方案RabbitMQ 天然支持消息 TTL 和死信交换机DLX。消息设置过期时间后如果没被消费会进入死信交换机再路由到另一个队列。这在语义上非常贴合延迟和死信场景很多团队早期用它实现延迟队列。缺点是 RabbitMQ 的吞吐量相对 Kafka 低不少消息堆积能力也有限大量消息积压时内存压力明显。而且 RabbitMQ 的延迟基于 TTL同一队列里消息若过期时间不同会出现队头阻塞问题。RocketMQ 方案RocketMQ 原生支持延迟等级比如 1s、5s、10s、30s、1m到最大几个小时的固定等级。它实现得非常优雅生产者指定延迟等级Broker 端会自动做延迟投递。同时也支持事务消息对重试和死信也有一套半内置的方案。但 RocketMQ 和 Kafka 不是同一个生态如果团队基础设施已经全面围绕 Kafka 建立引入 RocketMQ 意味着额外的集群、监听和运维成本。Kafka 方案Kafka 原生并不提供上述任何能力它只有一个追加日志模型生产者写入消费者按 offset 顺序拉取。但 Kafka 有高吞吐、海量堆积、分区顺序、消费组水平扩展这些能力。延迟、重试、死信都可以在上层用代码实现。我们最终选择 Kafka是因为团队不想为了这几个队列特性引入一套新的中间件而是在现有 Kafka 集群上用多 Topic 加代码调度来补全这三个能力。我用一个表格把选型关键点列出来方便你结合自己的处境判断。对比项RabbitMQRocketMQKafka延迟消息TTLDLX存在队头阻塞内置延迟等级需自研调度死信队列内置 DLX半内置需自建 Topic重试机制需手动设计支持需自建 Topic吞吐量中较高最高堆积能力一般较好极强运维成本独立集群独立集群复用现有集群消息回溯有限有限按 offset 灵活回溯如果你所在团队已经有 Kafka 集群并且 Kafka 承载着绝大多数业务消息那在 Kafka 上自研这三个功能远比再引入一套新中间件划算。2.2 自研方案的总体架构思路在 Kafka 上实现延迟、重试、死信整体思路其实非常统一用多个 Topic 分别承担不同的语义然后用生产者或消费者代码控制消息的流转路径。具体来说我会在系统里规划这样几类 Topic延迟专用 Topic存放当前还不该被处理的消息消息体里带上期望执行的时间戳。由一个延迟调度消费者负责扫描到期就转发到真正的业务处理 Topic。业务处理 Topic消费者真正处理业务的地方也是消息的主战场。重试 Topic存放处理失败但还有机会恢复的消息可能按不同的延迟档位拆分成多个 Topic比如 5 秒、30 秒、5 分钟。死信 Topic存放确认无法处理的消息。这里要特别说明一点消息在不同的 Topic 之间流转往往是靠一个小的转发逻辑来完成的。比如延迟 Topic 扫描到一条到期消息不是直接让原本的延迟消费者去执行业务而是把消息投递到业务处理 Topic。这样做的核心好处是职责分离延迟调度器只负责定时业务消费者只负责处理两边的并发配置、消费速率、监控告警可以独立设置互不干扰。我之前见过一个反面案例团队把延迟消息直接发到一个消费者里消费者拉取后判断时间未到就 Thread.sleep结果消费者线程全部阻塞在 sleep 上后续消息全被卡住消费者提交 offset 超时最终整个消费组重平衡消息重复消费一大片。这就是没有把定时和处理解耦造成的灾难。所以我们的设计里延迟到期后一律投递到独立的业务处理 Topic。2.3 Topic 规划与消息结构设计在动手写代码之前先把 Topic 规划好。我建议每个业务按独立前缀建 Topic避免业务之间互相干扰。举个具体例子订单超时关闭场景订单延迟 Topictopic-order-delay。生产者把下单消息扔进来消息体里带目标执行时间。订单处理 Topictopic-order-handler。真正执行订单关闭逻辑的地方。订单重试 Topic比如 topic-order-retry-30stopic-order-retry-5m按重试间隔拆分。订单死信 Topictopic-order-dead-letter。确认无法处理的订单消息最终落在这里。消息体格式上我强烈建议从一开始就设计好不要等到上线后再补。用 JSON 序列化比较通用但一定要预留下几个关键字段。字段含义示例msgId全局唯一消息 ID用于幂等订单号 时间戳生成的 UUIDbizType业务类型ORDER_CLOSEpayload原始业务数据订单 ID、用户 ID 等executeAt期望执行时间戳毫秒1735722600000retryCount已重试次数2sourceTopic原始来源 Topictopic-order-delaysourceOffset原始 offset便于回溯10234errorMsg最近一次失败原因下游库存接口超时字段不在多够用就行。但 retryCount 和 sourceTopic、sourceOffset 这三个字段你一定留好一旦消息走到死信你反查源头就要靠它们。3. 延迟队列的两种落地姿势Kafka 本身不提供延迟消息能力我们无非是通过 Topic 模型、调度扫描、或者中转投递来模拟。这里我把最实用的两种方案展开讲你根据业务体量和延迟精度需求选择。3.1 方案一单一延迟 Topic 加定时轮询这个方案最容易理解生产者把消息发到延迟 Topic消费者启动一个定时任务每隔几秒拉取一批消息检查消息体里的 executeAt 是否已经小于等于当前时间。如果到期就把消息转发到业务处理 Topic如果没到期就丢到下一个轮询周期再处理。实现层面可以使用 KafkaConsumer 的 poll 语义但不要用原生 poll 去逐条 sleep 等待。我的做法是写一个独立的延迟调度消费者它常规拉取消息把消息放入一个本地优先级队列按 executeAt 排序再有一个单独的调度线程触发到期消息的转发。这里要给一个具体的参数建议轮询间隔要根据业务对延迟精度的要求来定。如果只是订单超时关闭这种分钟级精度调度线程每分钟扫描一次就够了根本不需要秒级精确。如果是短信定时发送这类秒级触达要求的场景我建议轮询间隔控制在 1 秒左右不要低于 500 毫秒否则空转成本太高。这个方案的优点是实现简单Topic 数量少运维成本低适合延迟消息量不大、精度要求不高的场景。缺点也很明显如果延迟消息数量特别大调度消费者每次要遍历大量未到期消息无效遍历会占用带宽和 CPU。另外如果消费者重启本地优先级队列里的消息会丢失需要借助 Kafka 自身的 offset 提交机制来确保消息不丢但具体实现相对繁琐。我实际测过这个方案在消息量低于每秒几千条、延迟精度秒级可接受的情况下效果已经很稳。如果量级再大建议直接上方案二。3.2 方案二多 Topic 延迟槽加定时投递方案一的核心问题是大数据量下的无效扫描。方案二换个思路不再把所有延迟消息放在一个 Topic 里统一扫描而是按延迟范围拆分成多个延迟槽每个槽对应一个 Topic 和一台调度器。举例来说我们建这样一组 Topicdelay-5s、delay-30s、delay-1m、delay-30m、delay-1h、delay-24h。生产者发送消息时根据业务需要的延迟时间路由到最接近但又不小于目标时间的那个 Topic。对应每个 Topic有一个独立的调度消费者它专门负责这一档延迟的消息。比如 delay-5s 的调度消费者每 1 秒拉取一次快速投递delay-24h 的调度消费者每 30 分钟甚至每小时拉取一次没必要频繁空转。这种设计的优势是显而易见的每个 Topic 的消息延迟范围收敛调度器空转率大幅降低消息量天然被分片扩展性也好。如果某一档消息量激增可以只针对这一档增加调度消费者实例不影响其他档位。代价是 Topic 数量变多生产端路由逻辑需要维护一个延迟档位表。另外延迟档位是离散的如果业务要求的延迟时间不匹配档位我们需要约定向上匹配策略。举个例子业务要延迟 3 秒就投递到 delay-5s要延迟 10 分钟就投递到 delay-30m。这里会有一点误差但大多数业务场景并不需要精确到毫秒。更接近 RocketMQ 延迟等级的做法就是把档位表做成配置生产端计算目标延迟时间后在配置表里找最小且不小于目标档位的值。RocketMQ 提供的常见档位是 1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h这个表可以直接借鉴。3.3 延迟精度、消息积压与性能取舍关于延迟队列有一个最容易被忽略却极其重要的点你设置的延迟精度直接决定了调度器收益和集群负载之间的平衡。如果业务允许分钟级误差你就不该把调度器的轮询周期压到 1 秒否则大量空 poll 白白消耗 CPU。我们实际线上的做法是分档调优秒级档位的轮询周期是 1 秒分钟级档位的轮询周期是 30 秒小时级档位的轮询周期是 5 分钟。你算一笔账delay-24h 这个 Topic 一天可能只有几千条消息如果调度器每 1 秒空转一次一天的空转次数高达 86400 次一大半是无用功。把扫描周期拉长到 5 分钟收益几乎是零损耗的。消息积压也要单独说。延迟 Topic 的消息本身在 Kafka 中就是长时间不消费的状态这会被 Kafka 记录为消费滞后Lag。这个 Lag 是正常的但你要在监控系统里把它和真正的异常积压区分开。我们的做法是对不同的 Topic 配置各自的 Lag 告警阈值延迟 Topic 的告警阈值显著放宽只监控有没有消费者存活而不是offset 有没有落后。你可以在调度转发时增加一个条件判断消息在延迟 Topic 里已经停留了多久。如果一条消息的停留时间远超预期比如 delay-5s 的消息 30 分钟都没被转发出去多半是调度消费者挂了或者下游处理 Topic 消费积压导致无法写入。这时触发告警比单纯看 Lag 指标更能定位问题。4. 重试队列让失败的消费请求有序重来重试队列的设计核心不是简单地把消息发回原 Topic 再消费一遍而是要设计一套有节奏、有上限、有止损的重试机制。4.1 重试策略设计间隔、次数与动态退避重试策略的设计直接决定系统在下游故障时的表现。固定的失败后立即重试没有意义因为下游刚刚报错大概率还处在故障状态立刻再试反而容易加重下游压力。业界通用的是指数退避策略第一次重试等 5 秒第二次等 30 秒第三次等 5 分钟最多重试 5 次。每一次重试的间隔都在变大给下游充分的恢复时间。具体档位可以根据业务敏感度调整。我提供一个通用的配置模板重试次数延迟间隔对应重试 Topic15 秒topic-retry-5s230 秒topic-retry-30s35 分钟topic-retry-5m430 分钟topic-retry-30m52 小时topic-retry-2h这里有个实践细节与其让消息在同一个重试 Topic 中反复消费、反复失败、反复更新 retryCount不如拆分为多个重试 Topic分别对不同档位的消息进行处理。这样做的直接好处是不同间隔的重试彼此隔离某档重试 Topic 积压了不会影响其他档位而且消费线程可以根据消息所在 Topic 确定性地知道该等多久不用从消息体里解析重试次数再计算 sleep 时间。4.2 基于手动 Ack 加重试 Topic 的消费模型Kafka 默认是自动提交 offset也就是说消费者只要 poll 到消息offset 就会往后移动哪怕业务逻辑抛异常这条消息也不会再被消费到。所以在重试场景中必须关掉自动提交改为手动提交。我的实践模型是这样一个循环流程第一步业务消费者从处理 Topic 拉取消息执行业务逻辑。第二步如果执行成功提交 offset消息生命周期结束。第三步如果执行失败先判断异常类型。如果是参数错误、数据不合法这类不可重试异常直接投递到死信 Topic然后提交 offset。如果是下游超时、限流这类可重试异常读取消息体里的 retryCount如果已经超过最大重试次数投递到死信 Topic否则把 retryCount 加 1投递到对应的重试 Topic然后提交当前 offset。为什么失败后还要提交 offset因为消息已经被处理过了虽然处理失败但它被转移到了重试通道不再属于当前消费流程的处理范围。如果不提交该消息会反复出现在当前消费者里造成无限循环。这一点是新手最容易踩的坑。重试 Topic 的消费者则按各自的间隔策略去消费重新调用业务处理逻辑。注意重试 Topic 消费者的业务处理逻辑和主处理 Topic 消费者的业务处理逻辑应当保持一致不能重写一份有出入的处理方法。我们通常把核心处理方法抽成一个公共类两边都调用同一个方法避免主链路成功、重试链路失败这类判断分叉问题。4.3 幂等性重试之前必须解决重复消费重试机制引入后消息被消费多次是必然事件而且是设计如此。如果业务方法本身不具备幂等性重试就是一个定时炸弹。以订单关闭场景为例原本订单状态是待支付关闭后变成已关闭。如果关闭动作被重复执行第二次执行时订单已经是已关闭再关一次理论上无害。但如果你的处理方法里有金额扣减、库存回滚、发送通知这类副作用操作重复执行就会造成严重问题。幂等的最简做法是为每条消息生成全局唯一 msgId在处理前用 Redis 或数据库记录这个 msgId 的处理状态。处理逻辑先尝试写入状态处理中如果发现已经处理过就直接跳过。如果是数据库操作可以在订单表上加一个 lastProcessMsgId 字段或者用业务表自身的唯一约束来实现幂等。我建议在重试场景中一定要用先记录、后处理的顺序不要先处理再记录否则处理过程中崩溃重试时无法判断是否已执行。5. 死信队列最后的兜底与排查手段很多团队把死信队列当成垃圾桶只管往里扔从来不看。这是非常危险的做法。死信队列的价值在于它是一条蛛丝马迹链能帮你定位线上问题。要让死信队列真正发挥作用必须在设计上做到两点消息内容足够完整、消费和告警链路足够完善。5.1 什么样的消息该进死信不是所有失败消息都应该进死信死信是止损而不是甩锅。我的分类标准是三道闸门。第一道闸门不可重试的异常。消息反序列化失败、JSON 解析失败、业务 ID 缺失、枚举值不合法这类问题不管你重试多少次结果都一样。它们应该在第一次失败时就进死信节省所有后续无意义的消费。第二道闸门重试次数耗尽。消息本身没有问题只是下游一直不可用或者数据依赖不满足。这类消息在走完所有重试档位后如果仍然失败就应该进死信不能无限重试。无限重试的后果是消息永远在系统里流转占用资源而且真正的故障原因反而不容易暴露出来。第三道闸门业务规则不再满足的场景。比如一条优惠券发放消息在重试期间用户已经注销了账号这时候即使重试成功也没有意义。这类过期业务放在死信中由人工或自动补偿程序判断如何处理。5.2 死信消息的字段设计与转储死信消息的内容会直接决定排查效率。我见过很多团队的死信 Topic 里只有原始 payload没有来源、没有原因出了线上问题根本没法查。要解决这个问题投递死信时必须把关键链路信息补全。实际项目中我会在死信消息体里保留原始业务 payload一条消息最核心的数据。sourceTopic 和 sourceOffset记录它从哪个 Topic 的哪个位置来配合 Kafka 的 offset 查询 API 可以找到原始上下文。errorMsg最近一次异常堆栈信息。retryCount累计重试次数用于判断失败是否在合理次数内。deadLetterReason进死信的原因分类比如 RETRY_EXHAUSTED、DESERIALIZE_FAILED、BIZ_RULE_EXPIRED。最后投递时间方便值班人员判断这条消息是什么时候进入死信的。这些字段不是拍脑袋定的而是我在实际排障过程中反推出来的。有一次线上出现一批订单消息进入死信因为没记录 sourceTopic我们花了一个多小时才定位到是某个晚上的发布变更导致消息体字段变更而不是消费逻辑故障。从那以后强制要求所有死信消息必须携带这些字段。5.3 死信 Topic 的监控与人工处理链路死信队列如果只是存着等通知很容易变成无人区。我建议建立这样一套监控和处理机制死信 Topic 的消息数量作为核心告警指标。正常情况下一个健康的业务死信量应该接近于零。如果短时间内大量消息涌入死信 Topic一定是业务链路出了问题比如消息体字段被改坏了、下游接口大面积故障、或者代码里抛了不该抛的异常。告警阈值可以根据业务量设置比如单日超过 100 条就触发 P1 告警。死信消费者只做两件事转储和分析。它把死信消息从 Kafka 中读出写入 ES 或数据库方便按 msgId、业务类型、时间范围检索同时生成死信报表统计死信原因分布。注意死信消费者不应该自动把消息发送回原 Topic 重试这会掩盖问题也可能导致死信队列和重试队列之间无限循环。人工处理方面我建议配置一个简易的补偿管理后台运营或开发人员搜到具体 msgId 后能查看死信详情、标记处理状态、触发一次人工重放。重放时可以直接调用业务公共处理类而不是发送回 Kafka。这能避免重放过程中再次进入死信循环。6. 完整实现从生产者到消费者分别怎么写概念讲得再多不如一段能跑的代码来得实在。我基于 Java 和原生 Kafka 客户端给出一个最小可运行实现你把它套进自己的业务骨架里稍微改改就能用。6.1 生产端代码发送延迟与优先级消息生产者这个环节并不复杂核心是给消息设置一个足够全局唯一的 msgId并带上 executeAt 时间戳。这里我直接展示投递延迟消息的代码。public class DelayMessageProducer { private static final String DELAY_TOPIC topic-order-delay; private final KafkaProducerString, String producer; public DelayMessageProducer() { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.1.10:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.ACKS_CONFIG, all); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); this.producer new KafkaProducer(props); } public void sendDelayMessage(String bizType, String payload, long delayMillis) { long executeAt System.currentTimeMillis() delayMillis; String msgId UUID.randomUUID().toString(); DelayMessage message new DelayMessage(); message.setMsgId(msgId); message.setBizType(bizType); message.setPayload(payload); message.setExecuteAt(executeAt); message.setRetryCount(0); ProducerRecordString, String record new ProducerRecord( DELAY_TOPIC, bizType, JSON.toJSONString(message)); producer.send(record, (metadata, exception) - { if (exception ! null) { // 生产失败要记录日志并考虑本地重试或落盘 log.error(延迟消息发送失败: {}, msgId, exception); } }); } }关于参数设置我要多说一句acks 必须设成 allenable.idempotence 必须打开。这不是可选项而是为了保证消息不因网络问题而丢失或者重复。Kafka 的幂等生产者能解决重试导致消息重复的问题加上 msgId 的幂等消费设计两头都守住才能做到端到端的不丢不重。6.2 延迟轮询消费者代码定时扫描与到期投递延迟消费者的代码是这个方案里最关键的一环。我用优先级队列方案实现单一 Topic 轮询逻辑上比较直观也方便做本地调度。public class DelayQueueConsumer { private static final String DELAY_TOPIC topic-order-delay; private static final String HANDLER_TOPIC topic-order-handler; private final KafkaConsumerString, String consumer; private final KafkaProducerString, String producer; private final PriorityBlockingQueueDelayMessage pendingQueue new PriorityBlockingQueue(1000, Comparator.comparingLong(DelayMessage::getExecuteAt)); private final ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); // 是否停止标志 private volatile boolean running true; public DelayQueueConsumer() { Properties props new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.1.10:9092); props.put(ConsumerConfig.GROUP_ID_CONFIG, delay-handler); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); this.consumer new KafkaConsumer(props); this.producer new KafkaProducer(getProducerProps()); } public void start() { Thread pollThread new Thread(this::pollLoop, delay-poll-thread); pollThread.start(); scheduler.scheduleAtFixedRate(this::dispatchExpiredMessages, 0, 1, TimeUnit.SECONDS); } private void pollLoop() { consumer.subscribe(Collections.singletonList(DELAY_TOPIC)); while (running) { ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(500)); for (ConsumerRecordString, String record : records) { DelayMessage msg JSON.parseObject(record.value(), DelayMessage.class); pendingQueue.offer(msg); // 记录一条待处理的延迟消息 pendingMessageStore.put(msg.getMsgId(), msg); } consumer.commitSync(); } } private void dispatchExpiredMessages() { long now System.currentTimeMillis(); ListDelayMessage readyMessages new ArrayList(); DelayMessage msg; while ((msg pendingQueue.peek()) ! null msg.getExecuteAt() now) { pendingQueue.poll(); readyMessages.add(msg); } for (DelayMessage readyMsg : readyMessages) { // 转发到业务处理 Topic producer.send(new ProducerRecord(HANDLER_TOPIC, readyMsg.getBizType(), JSON.toJSONString(readyMsg)), (metadata, exception) - { if (exception ! null) { log.error(到期消息转发失败: {}, readyMsg.getMsgId(), exception); // 转发失败需要告警不要静默丢弃 } else { pendingMessageStore.remove(readyMsg.getMsgId()); } }); } producer.flush(); } }这里有一个非常关键的代码细节poll 到消息后立即 commitSync还是把消息放入优先级队列并转发成功后再提交我的建议是poll 到消息、放入本地队列后立刻提交 offset。因为此时消息还在我们本地的 pendingQueue 里并未真正投递到下游。即使 Kafka offset 已经提交、进程突然崩溃本地队列中的消息会丢失但我们可以通过配置 pendingMessageStore比如 Redis 或本地磁盘来保存延迟消息快照重启后从快照恢复。如果你不想引入额外的快照存储那就不要过早提交 offset而是等消息转发成功后再提交代价是重复消息可能性增加需要配合 msgId 幂等消费。两种方案各有利弊。如果消息量不大、业务允许偶发重复我推荐等到转发成功再提交 offset如果消息量大、需要精确控制建议引入 Redis 作为消息快照存储。6.3 业务处理与重试逻辑代码判断异常走向业务消费者负责真正的订单关闭逻辑也是重试和死信分流发生的地方。我把核心处理流程抽成一个公共方法主消费和重试消费都调用它。public class OrderHandlerConsumer { private static final int MAX_RETRY_COUNT 5; private final MapInteger, String retryTopicMap buildRetryTopicMap(); private final KafkaConsumerString, String consumer; private final KafkaProducerString, String producer; public void handleMessage(DelayMessage message) { try { // 1. 先做幂等检查 if (idempotentService.isProcessed(message.getMsgId())) { return; } // 2. 执行业务处理关闭订单、更新状态等 orderService.closeOrder(message.getPayload()); // 3. 记录处理成功的消息 ID idempotentService.markProcessed(message.getMsgId()); // 4. 提交 offset 由外层统一管理 } catch (BizException e) { // 不可重试异常直接进死信 sendToDeadLetter(message, e.getMessage(), BIZ_EXCEPTION); } catch (Exception e) { // 可重试异常判断次数 int retryCount message.getRetryCount() 1; if (retryCount MAX_RETRY_COUNT) { sendToDeadLetter(message, e.getMessage(), RETRY_EXHAUSTED); } else { message.setRetryCount(retryCount); sendToRetryTopic(message); } } } private void sendToRetryTopic(DelayMessage message) { // 根据 retryCount 选择对应的重试 Topic String retryTopic retryTopicMap.get(message.getRetryCount()); producer.send(new ProducerRecord(retryTopic, message.getBizType(), JSON.toJSONString(message))); } private void sendToDeadLetter(DelayMessage message, String errorMsg, String reason) { message.setErrorMsg(errorMsg); message.setDeadLetterReason(reason); producer.send(new ProducerRecord(DEAD_LETTER_TOPIC, message.getBizType(), JSON.toJSONString(message))); } }这里我想强调一下幂等检查的位置。很多团队把幂等检查放在外层消费者里这没有错但更重要的是要把检查逻辑和业务本身融合先查数据库中的订单状态如果已经是已关闭就直接跳过如果业务状态和预期不符同样跳过。只靠 Redis 里记 msgId 是不够的因为消息可能来自不同的触发路径比如人工补偿后台触发的任务和原始重试触发的任务携带的 msgId 不同但操作的是同一笔订单。这个时候业务状态本身的幂等才是最终兜底。6.4 关键参数与配置清单代码之外Kafka 的参数配置同样决定成败。我整理一份我常用的参数清单你可以对照检查自己的配置。组件参数推荐值说明Produceracksall必须保证消息写入所有副本Producerenable.idempotencetrue防止生产者重试导致重复Producerlinger.ms5-10适当增加以提升批量吞吐Producermax.in.flight.requests.per.connection5配合幂等确保顺序Consumerenable.auto.commitfalse必须手动管理 offsetConsumermax.poll.records50-200避免单次拉取过多导致处理超时Consumermax.poll.interval.ms300000防止业务耗时过长触发重平衡Brokerretention.ms至少 7 天延迟消息需要长留存便于回溯Brokermessage.max.bytes按最大消息体调整避免大消息写入失败有一个参数我必须单独提醒max.poll.interval.ms。默认值是 5 分钟如果消费者在处理一条消息时耗时超过这个时间消费者就会被判定为挂掉从而触发 Rebalance。在重试场景里业务方法可能调用下游接口接口超时如果设置很长很容易超过这个阈值。我的习惯是把该值设置为 10 分钟以上同时把业务处理的超时时间控制在 5 分钟以内给两者之间留出足够的安全余量。7. 常见问题与排查技巧实录理论推到这你一定想知道实际运行中都会遇到哪些问题。我把过去踩过的坑整理成一份速查表每一类问题后面都附上定位思路和解决方案。7.1 消息延迟不准总是提前或延后延迟不准是这类方案最容易被吐槽的问题原因大概率不在 Kafka 本身而在调度逻辑。一种情况是提前消费。业务要求延迟 30 分钟生产者投递到了 delay-1h 对应的 Topic等于提前了半小时。这种情况要检查生产端的档位匹配逻辑我们约定的是向上匹配即目标延迟小于等于该档位就投递到该档位。如果业务方传进来的延迟值是 45 分钟应该投递到 delay-1h不是 delay-30m。匹配逻辑要是写反了就会导致大量消息提前执行。另一种情况是延后消费。比如 delay-5s 档位的调度器每 30 秒才扫描一次那消息实际延迟可能是 35 秒。解决方法是检查该档位调度器的扫描周期是否和延迟精度匹配。秒级档位必须用秒级扫描周期否则延迟误差会让人抓狂。还有一类隐藏问题调度消费者所在机器和 Kafka 集群之间的时钟同步。如果调度器机器比 Kafka 慢 10 秒那执行时间就会晚 10 秒。建议所有服务统一接入时间同步这问题不大但容易被忽略。7.2 消息积压且在多个 Topic 间反复流转我在线上见过最恐怖的现象是某条消息一直在重试 Topic 和死信 Topic 之间转圈怎么都处理不完还触发了一堆告警。最终定位到的原因是异常分类写错了——把业务规则冲突这种不可重试异常错划分到了可重试的分类里导致每次重试都失败失败再重试直到重试次数耗尽进入死信死信消费者又因为告警配置不当反复触发人工重放。要避免这种问题必须先给异常分类画一条清晰的分界线哪些异常是可恢复的哪些是不可恢复的。不可恢复的异常比如数据校验失败、状态机不合法绝不进入重试队列。可以在公共异常类上增加一个标记接口比如 RetryableException 和 NonRetryableException改造起来比在 catch 块里一长串 if else 要省心得多。积压问题本身也值得警惕。如果重试 Topic 的 Lag 暴涨说明下游服务可能已经故障一段时间。此时要抑制重试生产者的投递速度避免故障下游被打爆。我的做法是给重试消费者增加一个简单的熔断开关连续失败率达到阈值时暂停消费重试 Topic并触发告警人工介入确认下游恢复后再打开开关。7.3 offset 提交异常导致消息丢或重复消息丢失和消息重复是 Kafka 消费场景两个绕不开的极端。丢失通常发生在自动提交下业务异常后没有处理重复通常发生在手动提交 offset 时先处理业务后提交 offset处理结束、offset 未提交时消费者挂掉重启后该消息又被拉到。坦白讲在 Kafka 的至少一次语义下重复消费是不可能从底层根除的我们只能靠幂等来消解后果。真正要谨慎避开的是丢消息处理成功后还没来得及提交 offset这不是问题因为会重复消费但处理失败后你提交了 offset消息就真的丢了。所以在业务逻辑里我的铁律是如果业务处理成功先执行所有必要的副作用操作再提交 offset。如果业务处理失败先把消息转移到重试 Topic 或死信 Topic再提交 offset。只要发生转移动作消息总会在某个 Topic 里存在不会丢失。这个顺序看起来简单但在并发场景里代码一旦写成先 commit 再处理或失败直接报错不转移问题就会立刻出现。7.4 延迟消息量大时 , 单线程调度器顶不住方案二的扩展性比方案一好但即使方案二单个 Topic 的调度器在极端情况下也可能成为瓶颈。比如整点大促时秒级档位会瞬间涌入大量延迟消息。遇到这种情况不要想着把一个消费者做成多线程就完事Kafka 的单分区只能被单线程消费。正确做法是按分区扩展延迟 Topic 设置多个分区调度消费者组设置多个实例每个实例负责一个或多个分区的延迟扫描。分发规则上建议按 bizType 或 msgId 取模保持同一个业务的消息集中在同一个分区方便局部顺序处理。如果消息量实在大到 Kafka 消费者组都无法承载那么延迟调度这一层可以拆出去做成独立服务延迟消息写入 Kafka 后由一个独立的延迟计算引擎把消息分配到内存时间轮到期后再批量回调 Kafka 生产端写入处理 Topic。这个方案是下一档的架构升级日常业务一般用不到但你要知道扩展路径是什么。最后分享一个实操中的小技巧关于整个链路最后想分享一个我们内部一直在用的经验给每一条消息都增加一个系统级标签字段比如 sysTag默认值是空字符串。当消息经历过重试、死信、人工重放之后每经过一跳就在这个字段上追加一跳标识比如 RETRY_1、DLQ、REPLAY。排查问题时只需要看这一个字段整条消息的流转路径就一目了然配合日志链路能快速定位到底是哪一环出了问题。这个技巧没有技术门槛但它在排查复杂跨系统问题时帮了我们大忙。消息治理的本质不是把消息变成一条单行道而是让每一次流转都有迹可循。把这条字段加上之后你会发现自己排查问题时的心态会稳很多。