KafkaProducer超时之谜：消息到底丢没丢？

摘要：线上告警 send() 抛了 TimeoutException，但 broker 端日志显示消息写进去了——到底丢没丢？三个超时参数 max.block.ms、request.timeout.ms、delivery.timeout.ms 分别在哪一层起作用？为什么超时重发会导致重复？这篇从源码把 KafkaProducer 的超时体系讲透。

关键词：KafkaProducer、TimeoutException、超时机制、幂等性、request.timeout.ms、delivery.timeout.ms、消息不丢

一个排查了三天的线上事故

去年有个业务，日志里每隔几分钟就蹦一条：

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for order-events-0: 120001 ms has passed since batch creation

业务同学来找我，说"是不是消息丢了？下游对账差了 300 条"。我第一反应不是查代码，是问他：你开幂等了吗？他说没开。

这就对了——TimeoutException 不等于消息丢了。很多人看到超时就以为消息没发出去，但实际情况可能完全相反。

那天排查了三天才搞清楚根因：broker GC 停顿 40 秒，客户端 request.timeout.ms 等不住，主动断开重发。broker 那边其实已经把消息写盘了，响应没来得及发回来。重发的消息又写了一次——所以下游对账不是少了，是多了。

这篇把 KafkaProducer 的超时体系彻底讲清楚。

01 三个超时参数，分别管哪一段

先记住一句话：三个超时管三段不同的时间。别搞混。

max.block.ms（默认 60 秒）——管的是 send() 方法本身。消息还在客户端这头，连 RecordAccumulator 都没进去。

什么时候会等？BufferPool 内存满了，或者元数据还没拉到。send() 在这里最多等 60 秒，等不到就直接抛异常：

// RecordAccumulator.append() 里的关键逻辑publicRecordAppendResultappend(Stringtopic,intpartition,...){// 申请 ByteBuffer，最多等 max.block.msByteBufferbuffer=free.allocate(size,maxTimeToBlockMs);if(buffer==null){// 等不到内存，直接抛——消息根本没进队列thrownewTimeoutException("Failed to allocate "+size+" bytes within the configured max blocking time "+maxTimeToBlockMs+" ms.");}}

request.timeout.ms（默认 30 秒）——管的是单次请求。消息已经发出去了，在等 broker 回响应。

这个超时是 InFlightRequests 里维护的。Sender 线程每跑一圈就检查一遍：

// InFlightRequests.checkTimeout()publicList<ClientRequest>checkTimeout(longnow,longrequestTimeoutMs){List<ClientRequest>expired=newArrayList<>();for(Map.Entry<String,Deque<ClientRequest>>entry:requests.entrySet()){Deque<ClientRequest>reqs=entry.getValue();Iterator<ClientRequest>it=reqs.iterator();while(it.hasNext()){ClientRequestrequest=it.next();longelapsed=now-request.startTimeMs();if(elapsed>requestTimeoutMs){expired.add(request);it.remove();// 标记这个节点需要断开重连connectionStatus.disconnect(entry.getKey());}}}returnexpired;}

注意这里——超时后不是直接判死刑，而是标记节点断开，然后这个 batch 会被放回 deque 队首，下次重发。消息没丢，只是这次请求超时了。

delivery.timeout.ms（默认 120 秒）——这是总兜底。从消息创建那一刻开始算，到最终成功或失败，不能超过这个时间。

这个检查在 RecordAccumulator 里，每次 poll 待发送 batch 时过一遍：

// RecordAccumulator.maybeExpire()privatebooleanmaybeExpire(ProducerBatchbatch,longnow){longelapsed=now-batch.createdTimeMs();if(elapsed>deliveryTimeoutMs){// 总时间到了，必须放弃——不管重试了多少次batch.done(-1,-1,newTimeoutException("Expiring "+batch.recordCount()+" record(s) for "+batch.topicPartition()+": "+elapsed+" ms has passed since batch creation"));returntrue;}returnfalse;}

线上日志里搜到的那句 “ms has passed since batch creation”，就是从这里打出来的。

02 超时后消息到底去哪了

这是最容易搞混的地方。两种超时，两种结果：

request.timeout.ms 超时——消息大概率还在

broker 可能已经处理了请求，只是响应在路上堵车了。客户端等不住，断开、重连、重发。重发的消息 broker 会再写一次。

这就是重复消息的根源——不是 Kafka 不可靠，是你没开幂等。

delivery.timeout.ms 超时——消息真的没发出去

总时间到了，batch.done() 触发，future 异常完成，callback 里 exception 不为 null。内存立即释放。这种情况下消息确实没到达 broker。

回到开头那个事故：业务日志里的异常消息是 “120001 ms has passed since batch creation”——这是 delivery.timeout.ms 超时。但他下游对账是多了不是少了，为什么？

因为在这 120 秒里，前几次 request.timeout 超时已经重发过了。其中某一次 broker 其实收到并写盘了，但响应没回来。最后 delivery 超时放弃，客户端认为失败，但 broker 那边数据已经在。

一句话总结：request.timeout 超时重发可能导致重复，delivery.timeout 超时才是真失败。

03 为什么开幂等就不会重复

既然超时重发会导致重复，Kafka 怎么解决的？答案是幂等性——enable.idempotence=true。

核心思路：给每个 Producer 分配一个 PID（Producer ID），每个 partition 维护一个单调递增的 sequence number。broker 收到请求后检查 sequence：

PID=abc, partition=0, sequence=5 第一次请求：sequence=5 → 接受，写入 网络超时，重发 第二次请求：sequence=5 → 已经写过了，直接返回成功

broker 端记住最近的 sequence，重复的请求直接返回成功，不重复写盘。这就是幂等性解决重复问题的原理。

但注意——幂等只解决单分区、单会话内的重复。如果你 Producer 重启了，PID 变了，老 sequence 就不认了。跨会话的重复要靠业务方去重（比如用业务唯一键）。

04 线上怎么配这三个参数

别用默认值就上线。根据场景调：

高吞吐核心链路（比如订单事件流）：

• delivery.timeout.ms调到 300000（5 分钟）——给重试留足时间
• request.timeout.ms保持 30000——别太小，大集群网络抖动很正常
• 一定开enable.idempotence=true——别省这个

日志/监控类低价值消息：

• delivery.timeout.ms调小（30 秒）——发不出去就丢，别占着内存
• request.timeout.ms可以 10 秒——快速失败快速降级

排查超时问题的正确姿势：

先看异常消息原文。“ms has passed since batch creation” = delivery 超时，查 broker GC 和 leader 切换。“request timed out” = request 超时，查网络延迟和连接状态。“Failed to allocate … bytes” = max.block 超时，查 buffer.memory 够不够。

写在最后

KafkaProducer 的超时设计不是一个参数，是三层防护：max.block.ms 保护业务线程不被阻塞，request.timeout.ms 给单次请求设上限，delivery.timeout.ms 保证消息最终有定论。

理解了这三层，下次看到 TimeoutException 就不会慌——先判断是哪一层的超时，再决定是查网络、查 broker、还是查内存。