做实时数据链路这几年，但凡聊到 Kafka，几乎绕不开两个话题：消息丢失和消息重复。而这两个问题最后都会指向同一个机制——消息重试。很多同学对重试的理解就是"发失败了重发一次""处理失败了再跑一遍"，真到了线上排查才发现，Kafka 的消息重试机制远不是配置一个 retries 参数那么简单。它涉及生产者发送链路、消费者消费链路、offset 提交节奏、幂等设计、死信兜底，是一套需要按角色、按时序、按一致性约束去设计的系统工程。这篇文章我打算把 Kafka 重试机制按生产者和消费者两条链路分别拆开讲，然后重点聊我实际项目里验证过的异步重试主题方案和死信主题设计。过程中会穿插一些调参和排障的经历，比如盲目加大重试次数导致消息堆积、只重试不做幂等导致数据翻倍这些坑。希望看完之后，你在自己做消息中间件选型和消费逻辑建设时，能直接找到可落地的参考方案。1. 重试机制不是单一功能：先分清链路角色和适用边界1.1 一条 Kafka 消息要过四道关口，失败可能发生在任何一道在聊重试之前，我建议先把链路画出来。Kafka 里一条消息从业务方产生到下游系统真正处理完，大致经过四段：生产者客户端把消息发送到 broker，broker 完成分区主副本写入和副本同步，消费者从分区拉取并提交 offset，最后消费者业务逻辑去处理这条消息。重试机制可以在每一段起作用，但每一段的重试目标完全不同。生产端失败常见的是网络闪断、broker 暂时不可达、分区 leader 正在切换，这类重试是为了"保证消息最终提交成功"，重试的代价是可能重复发送。消费者端失败则更复杂，可能是下游数据库抖动、外部接口超时，也可能是消息本身格式不符合预期。消费者侧重试是为了"保证消息最终被正确处理"，但代价可能是 offset 提交混乱、消费线程阻塞、甚至触发再平衡。还有一种失败几乎没法用重试解决，比如消息体超过了 broker 限制、目标主题不存在、消费者反序列化失败，这类属于配置或数据问题，重试多少次都是同样的结果。所以设计重试方案的第一步，不是去调参数，而是先明确你现在要解决的是哪一段链路的失败。同一个集群里，生产端重试和消费端重试要分开设计，混在一套配置里很容易出现"生产端重试次数很多，消费端却没有任何防护"这种失衡状态。1.2 可重试错误和不可重试错误的分界线我习惯把 Kafka 相关的异常分成两类：可重试和不可重试。这个划分非常关键，因为重试机制本身有意义的前提是"这次失败是暂时的，下一次大概率能成功"。可重试的典型情况：网络连接超时、UnknownTopicOrPartition（分区 leader 选举中）、NotLeaderOrPartition、LeaderNotAvailable、操作超时、broker 返回压力过大。这类异常通常是基础设施抖动造成的，等网络恢复或者 leader 选举完成后重试即可。不可重试的典型情况：消息尺寸超限（RecordTooLargeException）、序列化或反序列化失败、认证授权失败、业务校验失败。这类异常无论重试 3 次还是 30 次，结果完全一样。把这些错误也丢进重试循环，不仅浪费资源，还会把真正的问题掩盖在大量重试日志里。附一张我平时参考的分类表：失败类型典型异常是否重试处理方式网络抖动TimeoutException、NetworkException是短退避重试broker 状态切换LeaderNotAvailableException、ReplicaNotAvailable是中等退避重试配置类错误RecordTooLargeException、SerializationException否记录日志、人工排查业务逻辑失败下游系统校验失败视业务而定重试或进入死信1.3 任何重试方案都绕不开的两个前提：幂等和顺序重试的本质是让同一条消息被重新发送或重新处理，这天然带来了两个副作用：重复和乱序。如果业务系统没有针对重复做好幂等，那么重试次数越多，脏数据越多；如果处理逻辑对顺序敏感，那么异步重试很可能先处理失败消息的后续消息，导致状态回退或数据覆盖。我以前做过一个订单状态同步的任务，下游系统处理支付回调。第一次版本没有业务幂等键，消费者处理失败后把消息重新塞回队列重试，结果一次网络抖动导致十几条消息被重复处理，订单状态被写了两遍。所以我的一个铁律是：设计重试机制之前，先强制回答两个问题——这个消费逻辑是不是幂等的？如果不做任何缓存，同一业务键的消息重复处理两次结果是否一致？这个分区的消费顺序被打破后，下游能否接受短暂的不一致？如果这两个问题都没有明确答案，最好的做法不是继续设计重试架构，而是先跟业务方确认一致性要求。明确之后，再根据业务特点选择阻塞重试、异步重试或者干脆做死信兜底。2. 生产者端重试：三个核心参数和幂等性搭配2.1 retries、retry.backoff.ms 和 delivery.timeout.ms 的配合逻辑生产者端配置重试，最核心的参数有三个：retries、retry.backoff.ms、delivery.timeout.ms。很多资料单独解释三个参数，但实际生效的时候是叠加关系。retries 是最大重试次数，retry.backoff.ms 是每次重试之间的等待时间，delivery.timeout.ms 是整个发送行为的"总预算时间"，意思是生产者在这个时间内如果还没等来消息提交确认，就会放弃。注意，delivery.timeout.ms 是从消息准备发送开始计时，到 broker 确认提交结束，它包含了网络请求时间、退避间隔时间，也包括在缓冲里等待的时间。举一个简单例子：假设 retries=5，retry.backoff.ms=300，如果不设 delivery.timeout.ms，消息最多会经历 5 次重试，每次之间间隔 300ms。但如果同时设置了 delivery.timeout.ms=120000，那么无论 retries 设了多少，只要 120 秒内消息还没被确认提交，一样会超时失败。这两个控制条件谁先触发，谁就生效。这里有一个需要重点提醒的版本差异：从 Kafka 客户端 2.1 开始，生产者的 retries 默认值变成了 Integer.MAX_VALUE，也就是"无限重试"。官方做出这个改动的意图是让消息发送永远不会因为重试次数用完而失败，真正决定成败的是 delivery.timeout.ms。所以在新版本客户端里，如果你只调了 retries 没调 delivery.timeout，重试实际上是被 timeout 限制的，而不是被 retries 限制。我遇到过有同事在旧资料影响下把 retries 从 2147483647 改成 3，结果业务上出现了偶发发送失败，就是因为他没有理解这层改动。2.2 enable.idempotence=true 之后，重试不再是简单重发如果生产者在重试时只是简单地把同一消息重新发送一遍，在 acks=0 或 acks=1 的场景下是有可能产生重复消息的。好在 Kafka 提供了幂等生产者，通过 enable.idempotence=true 开启。它的原理是每个生产者分配一个 PID，并且给每个分区维护一个递增的序列号，broker 端会校验并保留最近的序列号。如果重试时发送的批次序列号与已接收的序列号重复，broker 会直接拒绝或忽略，避免重复写入。开启幂等生产者的一个额外收益是它限制了发送请求的并发数：Kafka 会要求 max.in.flight.requests.per.connection 不超过 5，这样重试时消息批次不会出现大面积乱序。这个限制保证了即使客户端同时有多个未确认的批次，重试过程也不至于彻底打乱分区顺序。我个人的线上配置通常会这样写：props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); props.put(ProducerConfig.RETRIES_CONFIG, 10); props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 200); props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 60000); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);这样做的理由是：acks=all 配合 min.insync.replicas，保证消息主副本和同步副本都写入成功；幂等开启后即使网络闪断导致重试，消息也不会在 broker 端重复落盘；delivery.timeout 设成 60 秒，预留了足够的重试窗口，又不至于让生产者无限等待而阻塞后续消息。2.3 生产端重试期间，最容易被忽略的元数据状态重试期间如果遇到分区 leader 变更，生产者需要重新拉取元数据并连接新的 leader。旧版本的元数据刷新间隔默认是 5 分钟，但在异常场景下这个间隔太长了。所以线上环境我会把 metadata.max.age.ms 调小一些，比如 30000，并且把 connections.max.idle.ms 设置为一个合理值。这样重试过程中出现 leader 切换，生产者能较快感知新节点。另外，生产端重试本质上是在给下游制造一个"消息可能会晚一点点到"的窗口。这个窗口内的数据系统仍然处于最终一致性的过渡态，业务侧如果出现读不到刚才写入的数据，不要误认为是 bug。我自己在排查问题时会先看生产端指标里的 record-error-rate 和 request-latency-avg，如果这两个指标在重试期间有波动，说明大概率是网络或 broker 侧的临时问题；如果死等 retry 也解决不了，就要去查 broker 端的 isr 状态了。3. 消费者本地重试：实现简单，但处理节奏必须严谨3.1 同步重试的最大代价不是慢，是拖垮整个分区很多应用在消费者代码里写的是最简单的同步重试：try-catch 捕获异常后，在循环里 sleep 一段时间再次处理，直到成功或者达到最大次数。这种方案在低流量、单条消息处理快的场景下没问题，但流量稍微上来就开始暴露问题。Runnable 在跑 org.apache.kafka.clients.consumer.KafkaConsumer 时，poll 循环是单线程的。如果某一条消息进入同步重试，整个消费线程就被占住了，该消费者负责的所有分区停止拉取。处理时间一长，消费者的 max.poll.interval.ms 会超时，broker 认为消费者失联，触发新一轮 rebalance。更麻烦的是，如果处理逻辑涉及外部调用，每次重试的 sleep 时间加上外部调用的超时时间可能远超过你预期的数字。所以我给同步重试定了一条红线：单条消息的重试总耗时，必须明显小于 max.poll.interval.ms，同时把 max.poll.records 调小。比如 max.poll.interval.ms=300000 时，一次 poll 拉取 500 条消息，其中一条失败重试 3 次，每次间隔 2 秒，这 4 秒的额外耗时还勉强可控；但如果一条消息的重试时间超过 30 秒，并且这种情况频繁发生，整个消费者的 lag 就会快速飙升。3.2 手动 ACK 和重试的配合：千万不要在重试期间提交 offset重试期间最危险的操作，是 KafkaConsumer 自动提交 offset 的机制还在后台跑。如果 enable.auto.commit=true，消费者会定期自动提交已拉取消息的 offset，那么这条正在重试的消息，其 offset 可能已经被提交了。这时候即使重试成功，也不会造成太大问题；但重试最终失败，这条消息就变成了"假装已消费"，真正丢失了。正确做法是在关键业务上关闭自动提交，改成手动 ack。如果一条消息处理失败，不提交它的 offset，消费者下一次 poll 时会再次拉取同一批或同一条消息。不过这里有个很实际的细节：Kafka 的 offset 是单调递增的，你不能在一个批次里只提交成功消息的 offset。也就是说，如果拉取了 100 条消息，第 50 条失败，你即使把前 49 条处理成功，也不能单独确认到第 49 条，因为 offset 提交是针对整个分区的。最终要么这个批次整体回退重新消费，要么你"跳过失败消息"并提交 offset，让失败消息进入其他通道。实践下来更稳妥的配合是：主流程里每处理一条消息就手动 commitSync，收到异常先记录失败消息的 topic、partition、offset，继续处理下一条，最后统一把失败消息发到重试或死信通道。这样不阻塞分区，也不丢数据，代价是需要额外维护补偿队列。3.3 一个简单的业务内指数退避重试范式如果确认要使用本地阻塞重试，我会按这个逻辑去写：int maxAttempts = 3; long baseBackoffMs = 1000L; long maxBackoffMs = 5000L; for (int attempt = 1; attempt <= maxAttempts; attempt++) { try { handleMessage(record); consumer.commitSync(); break; } catch (Exception e) { if (attempt == maxAttempts) { log.error("消息处理失败，进入死信通道, topic={}, partition={}, offset={}", record.topic(), record.partition(), record.offset(), e); sendToDlt(record); consumer.commitSync(); break; } long backoff = Math.min(baseBackoffMs * (1L << (attempt - 1)), maxBackoffMs); Thread.sleep(backoff); } }注意这里我在第 3 次失败后直接走死信通道并提交了 offset，避免失败消息无限阻塞分区。退避时间按 1 秒、2 秒、5 秒递增，加了一些随机抖动会更好，避免同一时刻大量消息同时触发重试对下游造成冲击。我一般会在这里加一个前提：如果 handleMessage 抛出的是不可重试异常（比如反序列化失败），应该直接走死信，不用消耗重试次数。4. 把重试交给 Kafka Topic：异步重试架构与实现细节4.1 为什么本地重试撑不住之后，要向"重试主题"迁移当消费流量上来之后，本地阻塞重试带来的问题会越来越明显：消费线程被占住，lag 上升，rebalance 频繁，失败消息还会拖累同一批次里其他正常消息。与其在消费端的线程里和失败消息死磕，不如把重试的职责从"消费进程内"移出去。这就是异步重试主题方案。异步重试的基本思路是：主消费者拉取业务主题消息后，如果处理失败，不直接放弃，而是把消息写入一个专门的重试主题（比如 topic-business-retry），同时记录重试次数和下次可重试的时间。另一个重试消费者专门消费这个重试主题，到了规定时间再把消息重新塞回业务主题，或者直接在重试消费者里调用业务逻辑。这样主消费者的 poll 循环永远不会因为单条消息的失败而阻塞，重试的节奏完全由重试主题控制。4.2 重试主题里的延迟消费：时间戳、暂停分区和轮询实现重试主题延迟消费的时候，我常用的是一个基于时间戳的简化方案：每条消息发送到重试主题时，在消息 header 里写入 nextRetryTime（下次允许处理的 Unix 时间戳）和 retryCount。重试消费者 poll 到消息后，先读 header 判断当前时间是否已经达到 nextRetryTime。如果还没到，就暂停这个 consumer 对应的分区，设置 pause 的集合只包含当前正在被延迟的分区，然后继续 poll 其他到了时间的消息。等待时间用 timeout 控制，比如每次 poll 设置 1000ms。如果只有一个分区需要延迟，暂停这个分区即可；如果多个分区都需要延迟，注意不要让暂停逻辑把所有分区都暂停掉，否则消费者会一直空转拉不到消息。这里我贴一段简化的实现逻辑：Map<TopicPartition, Long> nextRetryMap = new HashMap<>(); while (isRunning) { ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500)); for (ConsumerRecord<String, byte[]> record : records) { long nextTime = readHeader(record, "nextRetryTime"); if (System.currentTimeMillis() < nextTime) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); consumer.pause(singleton(tp)); nextRetryMap.put(tp, nextTime); continue; } process(record); } // 检查是否已到达可恢复消费的时间 for (Map.Entry<TopicPartition, Long> entry : nextRetryMap.entrySet()) { if (System.currentTimeMillis() >= entry.getValue()) { consumer.resume(singleton(entry.getKey())); nextRetryMap.remove(entry.getKey()); } } }这种方案最大的优势是不依赖外部延迟队列，纯 Kafka 内部就能完成。缺点也很明显：如果使用公共消费组和自动分区分配，暂停分区会受 rebalance 干扰，所以重试主题的分区数建议设小一些，并且重试消费者最好稳定运行，不要频繁扩容缩容。4.3 重试次数记录在 header 里还是外部存储里？消息重试次数要放在哪？我首选的是消息 header。Kafka 消息 header 本身支持二进制字段，可以存储 int、string 等简单元数据。每次失败时，从 header 取出 retryCount 加一，如果超过配置的最大次数就转死信。放在 header 里的好处是消息在主题之间流转时，只要原样发送，次数信息不会丢，整个重试体系可以只依赖 Kafka 本身。外部存储（比如 Redis）的竞速方案通常是为了更精准地管理延迟队列，但在高并发下还要处理一致性，复杂度高不少。如果只是做重试计数和退避计算，不建议引入额外组件。需要同时存"消息内容"和"重试上下文"时，我会把消息 key 设置成业务幂等键，而 value 用原始消息序列化格式，header 里放重试信息。退避算法上，我常用指数退避加抖动：long nextDelay = Math.min(initialDelay * Math.pow(2, retryCount), maxDelay); long jitter = ThreadLocalRandom.current().nextLong(0, 1000); nextRetryTime = System.currentTimeMillis() + nextDelay + jitter;4.4 异步重试到底会不会打乱顺序？如何保住顺序这是异步重试方案里最尖锐的问题。本地阻塞重试虽然慢，但它是严格的串行处理，消息顺序天然可靠。异步重试一旦把消息发回重试主题，再重新塞回业务主题，重试消息和后续新消息之间的相对顺序就完全被打破了。比如消息 A 处理失败，消息 B 跟着来了，B 可能先被处理，A 重试成功后才处理，顺序就颠倒了。如果业务对顺序要求很高，我在实践中是这样处理的：重试消息回发时，用原消息的 key 作为分区键，保证同一条业务 key 的消息始终进入同一个分区。同时，重试消费者在调起业务逻辑之前，先对上游是否还有同 key 消息做一次判断。如果同 key 的消息还在原主题未被消费，就把重试消息再次压回重试队列并延后一段时间。后端的可用状态判断有点像"攒批"，虽然会增加重试次数，但能把乱序窗口尽量缩小。如果业务接受最终一致性，顺序要求是"同一批内有序"而不是全局严格有序，那异步重试完全够用。反过来，如果业务要求绝对强顺序，我建议直接放弃异步重试方案，要么加大 max.poll.interval 和本地重试的容忍度，要么上死信人工处理，而不是在技术上生造一个顺序保证锁。5. 死信主题：重试尽头必须有一个合法出口5.1 死信主题该存什么，不该存什么无论重试参数怎么调，总有那么一些消息会屡屡失败。如果消费者处理失败的逻辑没有出口，就会陷入重试循环，把消费线程拖到崩溃。所以死信主题不是可选项，而是重试链路的必选项。死信主题里应该放的是：原始消息的完整内容、原始主题名、分区号、offset、失败原因摘要、重试次数。原始内容用来做补偿和分析，分区号和 offset 用来做原始链路定位。格式上建议直接复用上游的消息格式，比如 JSON 或 Avro，但最好在 value 外面再加一层可追溯的封装，这样死信表可以直接被消费然后写搜索引擎或数据库。我的实践是：死信消息的 key 保持原业务 key，value 用统一包装，例如{"original": {...}, "error": "...", "retryCount": 5, "timestamp": ...}。重试消费者发现重试次数超过阈值后，把这个包装后的消息发送到死信主题。注意不要把死信主题和业务主题混用，否则下游消费逻辑会因为数据格式不同而报错。5.2 重放死信消息：手动回灌的正确姿势死信主题里的消息不是用来躺着睡觉的。绝大多数场景下，我们会在修复完下游问题之后，把死信消息重新回放到业务主题，让消费者再跑一遍。回放时最怕的就是死信消息回到原主题后再次失败，然后再次进入重试、再次进入死信，形成一个无限循环。所以在回放前我会做两件事：一是检查死信消息的失败原因。如果是下游依赖恢复后能成功的暂时性问题，直接重建一个用户变化并重新推送；如果原始消息本身就不合法（比如字段缺失），先进行补偿修正，修好后再推送。二是给回放消息打上特殊标记。比如在 header 里加入replay-from-dlt=true，消费者在头部做分支判断：如果是死信回放的消息且再次失败，不再走重试逻辑，而是直接进入另一条"人工待办"通道。这样相当于把无限循环切断在最外层。回放操作我通常交给重试消费者完成，而不是另外写一个独立脚本。原因很简单，重试消费者已经具备从死信主题读取消息的能力，同时它知道原业务主题如何处理。差别只在 rebalance 风险和处理时效上。5.3 观察死信主题：监控和告警是重试链路的最后一环重试链路如果没有任何监控，出了问题会非常难看。我见过一个系统，重试主题里积压了几十万条消息，消费 lag 持续走高，但因为没人关注，直到业务方反馈数据延迟才发现问题。所以我会在接入重试死信方案时同步做三个指标：第一个是重试主题消费 lag。一旦 lag 保持增长，说明主消费者或重试消费者积压了处理不完的消息，需要扩容或检查下游性能。第二个是死信主题增量。死信消息在短时间内大幅增加，说明有系统性故障，比如数据库批量不可用、消息格式整体变更。第三个是单条消息的平均重试次数和重试时长。这两个指标能反映下游系统的健康度，如果平均重试次数长期偏高，下游显然已经处于亚健康状态。告警方面，我的经验是死信增量告警要及时并且可执行，不要流于表面。每次告警都应该附带一条样例死信消息和它的失败原因，至少要能自动完成"哪个分区、哪个业务键、什么异常"的定位。否则运维人员收到告警后还要手动去查原始日志，告警价值就大打折扣了。6. 实战排障记录：我在重试机制上踩过的几个典型坑6.1 无脑调大重试次数，恢复瞬间把下游打垮曾经有一次线上网络抖动，生产端大量消息进入重试，broker 恢复后所有重试消息几乎在同一时刻重新发送，瞬间打爆了下游数据库的连接数。当时重试次数设得很大，退避时间却很短，而且没有加随机抖动。后来我重新设计了生产端重试策略：退避时间从 200ms 改为 500ms 起步，每次退避乘以 1.5，并加上 0 到 300ms 的随机抖动，保证重试请求不会形成刺猬效应。这个案例给我最大的教训是：重试参数不是越大越好。重试是对消息的保护，但也可能是对下游的二次伤害。任何重试方案的退避时间都要考虑下游的真实吞吐能力，宁可少重试几次，多走死信补偿，也不要让重试风暴把整个系统打挂。6.2 只重试不幂等，一条订单回调消息被处理了三遍另一个项目里，消费端处理失败后会把消息重新塞回业务主题。当时大家认为"重试会复制一条消息"，忘记了 Kafka 的 at-least-once 语义叠加重试之后，可能变成 at-most-once 或者 exactly-once 的模拟失败。结果有一次下游业务数据库短暂超时，订单回调消息被重试了三次，下游表被插入了三条重复记录。排查到最后，发现是消费逻辑里没有用订单号做唯一约束。从那以后，我在所有消费逻辑里都强制设计幂等键。在数据库层面，业务表上有一个唯一索引；在缓存层面，用 Redis 的 SETNX 做消费去重标记。即使重试逻辑出现重复投递，下游也能正确去重。如果你的场景是写文件或调外部接口，至少也在业务代码里做一个基于业务键的去重检查，不要指望 Kafka 客户端帮你做掉一切。6.3 异步重试消费者 sleep 时间过长，把自己睡出了 rebalance异步重试方案的初期版本里，我用 Thread.sleep 实现延迟消费，失败后让消费者线程睡几十秒再重试。结果每次延迟期间消费者的心跳链路都在，但 poll 循环长时间不触发，触发了 max.poll.interval.ms 超时，消费者被判死，随后引发 rebalance。每次 rebalance 都会重新分配分区，前面的延迟状态经常丢失，消息被重复消费的次数成倍增加。后来我把延迟等待改成了前面介绍的"时间戳 + 暂停分区 + 轮询"方案，并且把重试消费者的 poll 循环保持在较短周期，保证心跳和 poll 始终活跃。这个改动之后，rebalance 问题才真正消失。如果你的场景轮询方案实现成本较高，另一个替代思路是引入额外的外部延迟队列，但这是后话，至少不要用裸 Thread.sleep 做长时间延迟。6.4 我的最终决策参考表项目做多了之后，我现在接手任何 Kafka 消费任务，会在设计文档里明确一条决策链：先确认业务是否幂等，再确认是否强顺序。如果两者都是"是"，就选阻塞重试，并在主线程内严格控制退避时间；如果顺序要求弱但幂等要求强，优先异步重试主题加死信兜底；如果既不幂等也不强顺序，那就简单很多，本地重试加严格死信即可。场景特征推荐方案关键约束强顺序、低吞吐消费线程内阻塞重试单条重试耗时 < max.poll.interval，max.poll.records 调小高吞吐、弱顺序异步重试主题 + 退避重试次数记录在 header，分区键用原业务键顺序和吞吐都高分区暂停 + 业务键攒批重试消费者手动 assign 分区，暂停分区避免 rebalance无法修复的失败死信主题 + 人工补偿重放前做有效性校验，添加 replay 标记在实际项目中，越是复杂的重试架构，越需要清晰的幂等设计和监控指标。Kafka 的消息重试机制不是一个开关，而是一条从生产端到消费端，再延伸到死信补偿的完整链路。希望这篇内容能帮你在设计自己的方案时少走一些弯路。
做实时数据链路这几年，但凡聊到 Kafka，几乎绕不开两个话题：消息丢失和消息重复。而这两个问题最后都会指向同一个机制——消息重试。很多同学对重试的理解就是"发失败了重发一次""处理失败了再跑一遍"，真到了线上排查才发现，Kafka 的消息重试机制远不是配置一个 retries 参数那么简单。它涉及生产者发送链路、消费者消费链路、offset 提交节奏、幂等设计、死信兜底，是一套需要按角色、按时序、按一致性约束去设计的系统工程。这篇文章我打算把 Kafka 重试机制按生产者和消费者两条链路分别拆开讲，然后重点聊我实际项目里验证过的异步重试主题方案和死信主题设计。过程中会穿插一些调参和排障的经历，比如盲目加大重试次数导致消息堆积、只重试不做幂等导致数据翻倍这些坑。希望看完之后，你在自己做消息中间件选型和消费逻辑建设时，能直接找到可落地的参考方案。1. 重试机制不是单一功能：先分清链路角色和适用边界1.1 一条 Kafka 消息要过四道关口，失败可能发生在任何一道在聊重试之前，我建议先把链路画出来。Kafka 里一条消息从业务方产生到下游系统真正处理完，大致经过四段：生产者客户端把消息发送到 broker，broker 完成分区主副本写入和副本同步，消费者从分区拉取并提交 offset，最后消费者业务逻辑去处理这条消息。重试机制可以在每一段起作用，但每一段的重试目标完全不同。生产端失败常见的是网络闪断、broker 暂时不可达、分区 leader 正在切换，这类重试是为了"保证消息最终提交成功"，重试的代价是可能重复发送。消费者端失败则更复杂，可能是下游数据库抖动、外部接口超时，也可能是消息本身格式不符合预期。消费者侧重试是为了"保证消息最终被正确处理"，但代价可能是 offset 提交混乱、消费线程阻塞、甚至触发再平衡。还有一种失败几乎没法用重试解决，比如消息体超过了 broker 限制、目标主题不存在、消费者反序列化失败，这类属于配置或数据问题，重试多少次都是同样的结果。所以设计重试方案的第一步，不是去调参数，而是先明确你现在要解决的是哪一段链路的失败。同一个集群里，生产端重试和消费端重试要分开设计，混在一套配置里很容易出现"生产端重试次数很多，消费端却没有任何防护"这种失衡状态。1.2 可重试错误和不可重试错误的分界线我习惯把 Kafka 相关的异常分成两类：可重试和不可重试。这个划分非常关键，因为重试机制本身有意义的前提是"这次失败是暂时的，下一次大概率能成功"。可重试的典型情况：网络连接超时、UnknownTopicOrPartition（分区 leader 选举中）、NotLeaderOrPartition、LeaderNotAvailable、操作超时、broker 返回压力过大。这类异常通常是基础设施抖动造成的，等网络恢复或者 leader 选举完成后重试即可。不可重试的典型情况：消息尺寸超限（RecordTooLargeException）、序列化或反序列化失败、认证授权失败、业务校验失败。这类异常无论重试 3 次还是 30 次，结果完全一样。把这些错误也丢进重试循环，不仅浪费资源，还会把真正的问题掩盖在大量重试日志里。附一张我平时参考的分类表：失败类型典型异常是否重试处理方式网络抖动TimeoutException、NetworkException是短退避重试broker 状态切换LeaderNotAvailableException、ReplicaNotAvailable是中等退避重试配置类错误RecordTooLargeException、SerializationException否记录日志、人工排查业务逻辑失败下游系统校验失败视业务而定重试或进入死信1.3 任何重试方案都绕不开的两个前提：幂等和顺序重试的本质是让同一条消息被重新发送或重新处理，这天然带来了两个副作用：重复和乱序。如果业务系统没有针对重复做好幂等，那么重试次数越多，脏数据越多；如果处理逻辑对顺序敏感，那么异步重试很可能先处理失败消息的后续消息，导致状态回退或数据覆盖。我以前做过一个订单状态同步的任务，下游系统处理支付回调。第一次版本没有业务幂等键，消费者处理失败后把消息重新塞回队列重试，结果一次网络抖动导致十几条消息被重复处理，订单状态被写了两遍。所以我的一个铁律是：设计重试机制之前，先强制回答两个问题——这个消费逻辑是不是幂等的？如果不做任何缓存，同一业务键的消息重复处理两次结果是否一致？这个分区的消费顺序被打破后，下游能否接受短暂的不一致？如果这两个问题都没有明确答案，最好的做法不是继续设计重试架构，而是先跟业务方确认一致性要求。明确之后，再根据业务特点选择阻塞重试、异步重试或者干脆做死信兜底。2. 生产者端重试：三个核心参数和幂等性搭配2.1 retries、retry.backoff.ms 和 delivery.timeout.ms 的配合逻辑生产者端配置重试，最核心的参数有三个：retries、retry.backoff.ms、delivery.timeout.ms。很多资料单独解释三个参数，但实际生效的时候是叠加关系。retries 是最大重试次数，retry.backoff.ms 是每次重试之间的等待时间，delivery.timeout.ms 是整个发送行为的"总预算时间"，意思是生产者在这个时间内如果还没等来消息提交确认，就会放弃。注意，delivery.timeout.ms 是从消息准备发送开始计时，到 broker 确认提交结束，它包含了网络请求时间、退避间隔时间，也包括在缓冲里等待的时间。举一个简单例子：假设 retries=5，retry.backoff.ms=300，如果不设 delivery.timeout.ms，消息最多会经历 5 次重试，每次之间间隔 300ms。但如果同时设置了 delivery.timeout.ms=120000，那么无论 retries 设了多少，只要 120 秒内消息还没被确认提交，一样会超时失败。这两个控制条件谁先触发，谁就生效。这里有一个需要重点提醒的版本差异：从 Kafka 客户端 2.1 开始，生产者的 retries 默认值变成了 Integer.MAX_VALUE，也就是"无限重试"。官方做出这个改动的意图是让消息发送永远不会因为重试次数用完而失败，真正决定成败的是 delivery.timeout.ms。所以在新版本客户端里，如果你只调了 retries 没调 delivery.timeout，重试实际上是被 timeout 限制的，而不是被 retries 限制。我遇到过有同事在旧资料影响下把 retries 从 2147483647 改成 3，结果业务上出现了偶发发送失败，就是因为他没有理解这层改动。2.2 enable.idempotence=true 之后，重试不再是简单重发如果生产者在重试时只是简单地把同一消息重新发送一遍，在 acks=0 或 acks=1 的场景下是有可能产生重复消息的。好在 Kafka 提供了幂等生产者，通过 enable.idempotence=true 开启。它的原理是每个生产者分配一个 PID，并且给每个分区维护一个递增的序列号，broker 端会校验并保留最近的序列号。如果重试时发送的批次序列号与已接收的序列号重复，broker 会直接拒绝或忽略，避免重复写入。开启幂等生产者的一个额外收益是它限制了发送请求的并发数：Kafka 会要求 max.in.flight.requests.per.connection 不超过 5，这样重试时消息批次不会出现大面积乱序。这个限制保证了即使客户端同时有多个未确认的批次，重试过程也不至于彻底打乱分区顺序。我个人的线上配置通常会这样写：props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); props.put(ProducerConfig.RETRIES_CONFIG, 10); props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 200); props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 60000); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);这样做的理由是：acks=all 配合 min.insync.replicas，保证消息主副本和同步副本都写入成功；幂等开启后即使网络闪断导致重试，消息也不会在 broker 端重复落盘；delivery.timeout 设成 60 秒，预留了足够的重试窗口，又不至于让生产者无限等待而阻塞后续消息。2.3 生产端重试期间，最容易被忽略的元数据状态重试期间如果遇到分区 leader 变更，生产者需要重新拉取元数据并连接新的 leader。旧版本的元数据刷新间隔默认是 5 分钟，但在异常场景下这个间隔太长了。所以线上环境我会把 metadata.max.age.ms 调小一些，比如 30000，并且把 connections.max.idle.ms 设置为一个合理值。这样重试过程中出现 leader 切换，生产者能较快感知新节点。另外，生产端重试本质上是在给下游制造一个"消息可能会晚一点点到"的窗口。这个窗口内的数据系统仍然处于最终一致性的过渡态，业务侧如果出现读不到刚才写入的数据，不要误认为是 bug。我自己在排查问题时会先看生产端指标里的 record-error-rate 和 request-latency-avg，如果这两个指标在重试期间有波动，说明大概率是网络或 broker 侧的临时问题；如果死等 retry 也解决不了，就要去查 broker 端的 isr 状态了。3. 消费者本地重试：实现简单，但处理节奏必须严谨3.1 同步重试的最大代价不是慢，是拖垮整个分区很多应用在消费者代码里写的是最简单的同步重试：try-catch 捕获异常后，在循环里 sleep 一段时间再次处理，直到成功或者达到最大次数。这种方案在低流量、单条消息处理快的场景下没问题，但流量稍微上来就开始暴露问题。Runnable 在跑 org.apache.kafka.clients.consumer.KafkaConsumer 时，poll 循环是单线程的。如果某一条消息进入同步重试，整个消费线程就被占住了，该消费者负责的所有分区停止拉取。处理时间一长，消费者的 max.poll.interval.ms 会超时，broker 认为消费者失联，触发新一轮 rebalance。更麻烦的是，如果处理逻辑涉及外部调用，每次重试的 sleep 时间加上外部调用的超时时间可能远超过你预期的数字。所以我给同步重试定了一条红线：单条消息的重试总耗时，必须明显小于 max.poll.interval.ms，同时把 max.poll.records 调小。比如 max.poll.interval.ms=300000 时，一次 poll 拉取 500 条消息，其中一条失败重试 3 次，每次间隔 2 秒，这 4 秒的额外耗时还勉强可控；但如果一条消息的重试时间超过 30 秒，并且这种情况频繁发生，整个消费者的 lag 就会快速飙升。3.2 手动 ACK 和重试的配合：千万不要在重试期间提交 offset重试期间最危险的操作，是 KafkaConsumer 自动提交 offset 的机制还在后台跑。如果 enable.auto.commit=true，消费者会定期自动提交已拉取消息的 offset，那么这条正在重试的消息，其 offset 可能已经被提交了。这时候即使重试成功，也不会造成太大问题；但重试最终失败，这条消息就变成了"假装已消费"，真正丢失了。正确做法是在关键业务上关闭自动提交，改成手动 ack。如果一条消息处理失败，不提交它的 offset，消费者下一次 poll 时会再次拉取同一批或同一条消息。不过这里有个很实际的细节：Kafka 的 offset 是单调递增的，你不能在一个批次里只提交成功消息的 offset。也就是说，如果拉取了 100 条消息，第 50 条失败，你即使把前 49 条处理成功，也不能单独确认到第 49 条，因为 offset 提交是针对整个分区的。最终要么这个批次整体回退重新消费，要么你"跳过失败消息"并提交 offset，让失败消息进入其他通道。实践下来更稳妥的配合是：主流程里每处理一条消息就手动 commitSync，收到异常先记录失败消息的 topic、partition、offset，继续处理下一条，最后统一把失败消息发到重试或死信通道。这样不阻塞分区，也不丢数据，代价是需要额外维护补偿队列。3.3 一个简单的业务内指数退避重试范式如果确认要使用本地阻塞重试，我会按这个逻辑去写：int maxAttempts = 3; long baseBackoffMs = 1000L; long maxBackoffMs = 5000L; for (int attempt = 1; attempt <= maxAttempts; attempt++) { try { handleMessage(record); consumer.commitSync(); break; } catch (Exception e) { if (attempt == maxAttempts) { log.error("消息处理失败，进入死信通道, topic={}, partition={}, offset={}", record.topic(), record.partition(), record.offset(), e); sendToDlt(record); consumer.commitSync(); break; } long backoff = Math.min(baseBackoffMs * (1L << (attempt - 1)), maxBackoffMs); Thread.sleep(backoff); } }注意这里我在第 3 次失败后直接走死信通道并提交了 offset，避免失败消息无限阻塞分区。退避时间按 1 秒、2 秒、5 秒递增，加了一些随机抖动会更好，避免同一时刻大量消息同时触发重试对下游造成冲击。我一般会在这里加一个前提：如果 handleMessage 抛出的是不可重试异常（比如反序列化失败），应该直接走死信，不用消耗重试次数。4. 把重试交给 Kafka Topic：异步重试架构与实现细节4.1 为什么本地重试撑不住之后，要向"重试主题"迁移当消费流量上来之后，本地阻塞重试带来的问题会越来越明显：消费线程被占住，lag 上升，rebalance 频繁，失败消息还会拖累同一批次里其他正常消息。与其在消费端的线程里和失败消息死磕，不如把重试的职责从"消费进程内"移出去。这就是异步重试主题方案。异步重试的基本思路是：主消费者拉取业务主题消息后，如果处理失败，不直接放弃，而是把消息写入一个专门的重试主题（比如 topic-business-retry），同时记录重试次数和下次可重试的时间。另一个重试消费者专门消费这个重试主题，到了规定时间再把消息重新塞回业务主题，或者直接在重试消费者里调用业务逻辑。这样主消费者的 poll 循环永远不会因为单条消息的失败而阻塞，重试的节奏完全由重试主题控制。4.2 重试主题里的延迟消费：时间戳、暂停分区和轮询实现重试主题延迟消费的时候，我常用的是一个基于时间戳的简化方案：每条消息发送到重试主题时，在消息 header 里写入 nextRetryTime（下次允许处理的 Unix 时间戳）和 retryCount。重试消费者 poll 到消息后，先读 header 判断当前时间是否已经达到 nextRetryTime。如果还没到，就暂停这个 consumer 对应的分区，设置 pause 的集合只包含当前正在被延迟的分区，然后继续 poll 其他到了时间的消息。等待时间用 timeout 控制，比如每次 poll 设置 1000ms。如果只有一个分区需要延迟，暂停这个分区即可；如果多个分区都需要延迟，注意不要让暂停逻辑把所有分区都暂停掉，否则消费者会一直空转拉不到消息。这里我贴一段简化的实现逻辑：Map<TopicPartition, Long> nextRetryMap = new HashMap<>(); while (isRunning) { ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500)); for (ConsumerRecord<String, byte[]> record : records) { long nextTime = readHeader(record, "nextRetryTime"); if (System.currentTimeMillis() < nextTime) { TopicPartition tp = new TopicPartition(record.topic(), record.partition()); consumer.pause(singleton(tp)); nextRetryMap.put(tp, nextTime); continue; } process(record); } // 检查是否已到达可恢复消费的时间 for (Map.Entry<TopicPartition, Long> entry : nextRetryMap.entrySet()) { if (System.currentTimeMillis() >= entry.getValue()) { consumer.resume(singleton(entry.getKey())); nextRetryMap.remove(entry.getKey()); } } }这种方案最大的优势是不依赖外部延迟队列，纯 Kafka 内部就能完成。缺点也很明显：如果使用公共消费组和自动分区分配，暂停分区会受 rebalance 干扰，所以重试主题的分区数建议设小一些，并且重试消费者最好稳定运行，不要频繁扩容缩容。4.3 重试次数记录在 header 里还是外部存储里？消息重试次数要放在哪？我首选的是消息 header。Kafka 消息 header 本身支持二进制字段，可以存储 int、string 等简单元数据。每次失败时，从 header 取出 retryCount 加一，如果超过配置的最大次数就转死信。放在 header 里的好处是消息在主题之间流转时，只要原样发送，次数信息不会丢，整个重试体系可以只依赖 Kafka 本身。外部存储（比如 Redis）的竞速方案通常是为了更精准地管理延迟队列，但在高并发下还要处理一致性，复杂度高不少。如果只是做重试计数和退避计算，不建议引入额外组件。需要同时存"消息内容"和"重试上下文"时，我会把消息 key 设置成业务幂等键，而 value 用原始消息序列化格式，header 里放重试信息。退避算法上，我常用指数退避加抖动：long nextDelay = Math.min(initialDelay * Math.pow(2, retryCount), maxDelay); long jitter = ThreadLocalRandom.current().nextLong(0, 1000); nextRetryTime = System.currentTimeMillis() + nextDelay + jitter;4.4 异步重试到底会不会打乱顺序？如何保住顺序这是异步重试方案里最尖锐的问题。本地阻塞重试虽然慢，但它是严格的串行处理，消息顺序天然可靠。异步重试一旦把消息发回重试主题，再重新塞回业务主题，重试消息和后续新消息之间的相对顺序就完全被打破了。比如消息 A 处理失败，消息 B 跟着来了，B 可能先被处理，A 重试成功后才处理，顺序就颠倒了。如果业务对顺序要求很高，我在实践中是这样处理的：重试消息回发时，用原消息的 key 作为分区键，保证同一条业务 key 的消息始终进入同一个分区。同时，重试消费者在调起业务逻辑之前，先对上游是否还有同 key 消息做一次判断。如果同 key 的消息还在原主题未被消费，就把重试消息再次压回重试队列并延后一段时间。后端的可用状态判断有点像"攒批"，虽然会增加重试次数，但能把乱序窗口尽量缩小。如果业务接受最终一致性，顺序要求是"同一批内有序"而不是全局严格有序，那异步重试完全够用。反过来，如果业务要求绝对强顺序，我建议直接放弃异步重试方案，要么加大 max.poll.interval 和本地重试的容忍度，要么上死信人工处理，而不是在技术上生造一个顺序保证锁。5. 死信主题：重试尽头必须有一个合法出口5.1 死信主题该存什么，不该存什么无论重试参数怎么调，总有那么一些消息会屡屡失败。如果消费者处理失败的逻辑没有出口，就会陷入重试循环，把消费线程拖到崩溃。所以死信主题不是可选项，而是重试链路的必选项。死信主题里应该放的是：原始消息的完整内容、原始主题名、分区号、offset、失败原因摘要、重试次数。原始内容用来做补偿和分析，分区号和 offset 用来做原始链路定位。格式上建议直接复用上游的消息格式，比如 JSON 或 Avro，但最好在 value 外面再加一层可追溯的封装，这样死信表可以直接被消费然后写搜索引擎或数据库。我的实践是：死信消息的 key 保持原业务 key，value 用统一包装，例如{"original": {...}, "error": "...", "retryCount": 5, "timestamp": ...}。重试消费者发现重试次数超过阈值后，把这个包装后的消息发送到死信主题。注意不要把死信主题和业务主题混用，否则下游消费逻辑会因为数据格式不同而报错。5.2 重放死信消息：手动回灌的正确姿势死信主题里的消息不是用来躺着睡觉的。绝大多数场景下，我们会在修复完下游问题之后，把死信消息重新回放到业务主题，让消费者再跑一遍。回放时最怕的就是死信消息回到原主题后再次失败，然后再次进入重试、再次进入死信，形成一个无限循环。所以在回放前我会做两件事：一是检查死信消息的失败原因。如果是下游依赖恢复后能成功的暂时性问题，直接重建一个用户变化并重新推送；如果原始消息本身就不合法（比如字段缺失），先进行补偿修正，修好后再推送。二是给回放消息打上特殊标记。比如在 header 里加入replay-from-dlt=true，消费者在头部做分支判断：如果是死信回放的消息且再次失败，不再走重试逻辑，而是直接进入另一条"人工待办"通道。这样相当于把无限循环切断在最外层。回放操作我通常交给重试消费者完成，而不是另外写一个独立脚本。原因很简单，重试消费者已经具备从死信主题读取消息的能力，同时它知道原业务主题如何处理。差别只在 rebalance 风险和处理时效上。5.3 观察死信主题：监控和告警是重试链路的最后一环重试链路如果没有任何监控，出了问题会非常难看。我见过一个系统，重试主题里积压了几十万条消息，消费 lag 持续走高，但因为没人关注，直到业务方反馈数据延迟才发现问题。所以我会在接入重试死信方案时同步做三个指标：第一个是重试主题消费 lag。一旦 lag 保持增长，说明主消费者或重试消费者积压了处理不完的消息，需要扩容或检查下游性能。第二个是死信主题增量。死信消息在短时间内大幅增加，说明有系统性故障，比如数据库批量不可用、消息格式整体变更。第三个是单条消息的平均重试次数和重试时长。这两个指标能反映下游系统的健康度，如果平均重试次数长期偏高，下游显然已经处于亚健康状态。告警方面，我的经验是死信增量告警要及时并且可执行，不要流于表面。每次告警都应该附带一条样例死信消息和它的失败原因，至少要能自动完成"哪个分区、哪个业务键、什么异常"的定位。否则运维人员收到告警后还要手动去查原始日志，告警价值就大打折扣了。6. 实战排障记录：我在重试机制上踩过的几个典型坑6.1 无脑调大重试次数，恢复瞬间把下游打垮曾经有一次线上网络抖动，生产端大量消息进入重试，broker 恢复后所有重试消息几乎在同一时刻重新发送，瞬间打爆了下游数据库的连接数。当时重试次数设得很大，退避时间却很短，而且没有加随机抖动。后来我重新设计了生产端重试策略：退避时间从 200ms 改为 500ms 起步，每次退避乘以 1.5，并加上 0 到 300ms 的随机抖动，保证重试请求不会形成刺猬效应。这个案例给我最大的教训是：重试参数不是越大越好。重试是对消息的保护，但也可能是对下游的二次伤害。任何重试方案的退避时间都要考虑下游的真实吞吐能力，宁可少重试几次，多走死信补偿，也不要让重试风暴把整个系统打挂。6.2 只重试不幂等，一条订单回调消息被处理了三遍另一个项目里，消费端处理失败后会把消息重新塞回业务主题。当时大家认为"重试会复制一条消息"，忘记了 Kafka 的 at-least-once 语义叠加重试之后，可能变成 at-most-once 或者 exactly-once 的模拟失败。结果有一次下游业务数据库短暂超时，订单回调消息被重试了三次，下游表被插入了三条重复记录。排查到最后，发现是消费逻辑里没有用订单号做唯一约束。从那以后，我在所有消费逻辑里都强制设计幂等键。在数据库层面，业务表上有一个唯一索引；在缓存层面，用 Redis 的 SETNX 做消费去重标记。即使重试逻辑出现重复投递，下游也能正确去重。如果你的场景是写文件或调外部接口，至少也在业务代码里做一个基于业务键的去重检查，不要指望 Kafka 客户端帮你做掉一切。6.3 异步重试消费者 sleep 时间过长，把自己睡出了 rebalance异步重试方案的初期版本里，我用 Thread.sleep 实现延迟消费，失败后让消费者线程睡几十秒再重试。结果每次延迟期间消费者的心跳链路都在，但 poll 循环长时间不触发，触发了 max.poll.interval.ms 超时，消费者被判死，随后引发 rebalance。每次 rebalance 都会重新分配分区，前面的延迟状态经常丢失，消息被重复消费的次数成倍增加。后来我把延迟等待改成了前面介绍的"时间戳 + 暂停分区 + 轮询"方案，并且把重试消费者的 poll 循环保持在较短周期，保证心跳和 poll 始终活跃。这个改动之后，rebalance 问题才真正消失。如果你的场景轮询方案实现成本较高，另一个替代思路是引入额外的外部延迟队列，但这是后话，至少不要用裸 Thread.sleep 做长时间延迟。6.4 我的最终决策参考表项目做多了之后，我现在接手任何 Kafka 消费任务，会在设计文档里明确一条决策链：先确认业务是否幂等，再确认是否强顺序。如果两者都是"是"，就选阻塞重试，并在主线程内严格控制退避时间；如果顺序要求弱但幂等要求强，优先异步重试主题加死信兜底；如果既不幂等也不强顺序，那就简单很多，本地重试加严格死信即可。场景特征推荐方案关键约束强顺序、低吞吐消费线程内阻塞重试单条重试耗时 < max.poll.interval，max.poll.records 调小高吞吐、弱顺序异步重试主题 + 退避重试次数记录在 header，分区键用原业务键顺序和吞吐都高分区暂停 + 业务键攒批重试消费者手动 assign 分区，暂停分区避免 rebalance无法修复的失败死信主题 + 人工补偿重放前做有效性校验，添加 replay 标记在实际项目中，越是复杂的重试架构，越需要清晰的幂等设计和监控指标。Kafka 的消息重试机制不是一个开关，而是一条从生产端到消费端，再延伸到死信补偿的完整链路。希望这篇内容能帮你在设计自己的方案时少走一些弯路。