当前位置：首页 > news >正文

Kafka 消息重试设计：别让失败消息原地打转

news 2026/7/5 14:35:25

Kafka 消息重试设计：别让失败消息原地打转

一、重试不是直接再消费一次

Kafka 常用于微服务解耦。消费失败时，很多代码会直接抛异常，让消息再次被消费。这样简单，但如果下游一直不可用或消息本身有问题，就会原地打转，阻塞后续消息，甚至形成重试风暴。

消息重试要区分临时失败和永久失败。

二、先拆失败类型

flowchart TD A[消费失败] --> B{失败类型} B --> C[下游超时] B --> D[数据格式错误] B --> E[业务状态冲突] C --> F[延迟重试] D --> G[死信] E --> H[人工或补偿]

不是所有失败都该重试。格式错误重试一百次也不会好，下游超时才适合延迟重试。

kafka_retry_policy: transient_error: retry_with_backoff invalid_message: dead_letter business_conflict: manual_review

策略要按错误码判断，不要所有异常一刀切。

三、延迟重试要分级

record RetryMessage(String key, int retryCount, String lastError) {}

可以用多个 retry topic 实现不同延迟，比如 1 分钟、5 分钟、30 分钟。每次失败增加 retryCount，超过上限进入死信。

重试消息要保留原始 key 和上下文，方便幂等和排查。不要只把 payload 丢进新 topic。

四、消费逻辑要幂等

Kafka 默认至少一次投递，重复消费是正常现象。业务写入必须有幂等键，不能因为重试创建重复订单、重复通知、重复扣费。

consumer_idempotency: idempotency_key: message_id unique_index: true record_processed_result: true

如果处理成功但提交 offset 失败，消息还会再次投递。幂等设计就是为这种情况准备的。

死信队列也不是垃圾桶。进入死信后，要有告警、查询、修复和重放流程。否则失败消息只是换个地方沉默。

最后，重试要保护下游。下游故障时，所有消费者疯狂重试，只会让恢复更慢。退避、限流和熔断要一起设计。

重试还要保留顺序要求。某些 topic 的消息必须按业务 key 保序，比如同一个订单的状态变化。如果失败消息被丢到延迟 topic，后续消息可能先执行，状态就乱了。

ordering_policy: preserve_order_by_key: true block_following_messages_on_key_failure: optional

是否阻塞后续消息要按业务决定。强顺序业务宁愿慢，也不能乱；弱顺序业务可以让后续消息先走，提高吞吐。

还要给死信队列加分类字段。格式错误、业务冲突、下游不可用进入死信后的处理人不同，分类清楚才能分派。

最后，重放死信要谨慎。修复代码后批量重放前，最好先小批量验证，避免把历史错误一次性重新打进系统。

消费者还要暴露积压指标。重试 topic、主 topic、死信 topic 的 lag、最老消息年龄和消费速率都要看。只看主 topic lag，可能漏掉大量正在延迟重试的失败消息。

kafka_retry_metrics: main_lag: true retry_lag: true dead_letter_count: true oldest_message_age: true

还要给重试链路设置最大生命周期。消息重试几天后还没成功，通常已经失去业务意义，应该进入人工处理，而不是继续消耗资源。

重试系统的可观测性经常被忽视。除了常规的 lag、消费速率等指标外，重试系统还需要关注"重试环"：某些消息不断在重试 topic 之间流转，但始终无法成功，却也没进入死信队列。这种情况通常发生在错误分类不准确时——消息被判定为"可重试"，但实际上问题根本没修复。为了避免消息在重试环里无限循环，可以在消息头里加入"已尝试的 retry topic 列表"，当检测到消息已经在所有重试级别都失败过时，直接转入死信，而不是继续下一个重试周期。

Kafka 重试还需要考虑"消费者组内重平衡"的影响。当消费者加入或离开组时，分区会重新分配，这可能导致正在处理的重试消息被中断。如果消息处理还没完成就触发了 rebalance，这条消息可能会被另一个消费者再次消费，导致重复处理。虽然幂等设计可以解决数据重复问题，但会带来额外的处理成本。在设计重试机制时，可以通过调整max.poll.interval.ms和session.timeout.ms来减少不必要的重平衡，同时确保消息处理能够在预期的 poll 间隔内完成。