1. 消息发送的“薛定谔状态”为什么代码不报错消息却没了刚接触 Kafka 那会儿我踩过一个特别典型的坑。本地写了个 Producer 程序producer.send(record)调用完控制台干干净净没有异常堆栈没有红色日志程序正常退出。我满心欢喜地去 Consumer 那边拉数据结果一条都没有。反复检查 Topic 名称、Broker 地址、序列化器全都没问题。代码不报错消息却发送失败——这个现象困扰了我整整一个下午。后来把 Producer 的日志级别调到 DEBUG才在茫茫日志里看到一行不起眼的RecordTooLargeException。原来消息体超过了 Broker 端message.max.bytes的限制但 Producer 默认是异步发送异常被“吞”在了回调里主线程根本感知不到。这就是 Kafka Producer 最容易被忽视的设计发送动作和发送结果在时间上是解耦的。这个现象背后牵扯出的是一整套关于 Kafka Producer 发送机制、异常处理模型、配置参数体系的知识。它不是一个孤立的 bug而是理解 Kafka 客户端设计哲学的入口。这篇文章我会从源码层面把这件事彻底讲透消息从send()调用到真正落盘的完整链路是什么哪些环节可能“静默失败”acks、retries、delivery.timeout.ms这些参数到底怎么配合以及怎么写出一个“消息发不出去就一定会知道”的可靠 Producer。适合已经用过 Kafka 但被类似问题坑过、想深入理解客户端机制的开发者。2. 发送链路全景拆解一条消息的奇幻漂流2.1 从 send() 到 RecordAccumulator主线程只做了三件事很多人以为producer.send()就是“把消息发出去”实际上它做的事情远比想象中少。在 Kafka 的源码里KafkaProducer.send()方法的核心逻辑可以概括为三步拦截器处理、序列化、追加到累加器。第一步ProducerInterceptors会依次调用所有拦截器的onSend()方法。这一步是同步的如果拦截器抛异常send()会直接抛出。但大多数情况下拦截器只是做埋点或修改消息头不会阻断。第二步waitOnMetadata()会检查目标 Topic 的元数据是否可用。如果元数据还没拉取到这里会阻塞等待最长等待时间由max.block.ms控制默认 60 秒。超过这个时间会抛TimeoutException。注意这是少数几个会同步抛异常的地方。第三步RecordAccumulator.append()把消息追加到对应分区的DequeProducerBatch中。如果累加器的内存池BufferPool已经耗尽这里也会阻塞同样受max.block.ms约束。走完这三步send()方法就返回了一个FutureRecordMetadata。消息此时还躺在客户端的内存缓冲区里根本没有离开 JVM。真正的网络发送是由一个独立的Sender线程在后台完成的。这就是为什么主线程感知不到发送失败——失败发生在另一个线程里。2.2 Sender 线程与 NetworkClient异步发送的真正执行者Sender线程是 KafkaProducer 启动时创建的一个守护线程它在一个while (running)循环里不断执行runOnce()。每一轮循环做这几件事从RecordAccumulator里取出已经满足发送条件的批次ready()方法判断把这些批次按照目标 Broker 节点分组然后通过NetworkClient发送出去。NetworkClient底层是 Java NIO 的Selector它负责维护与各个 Broker 的 TCP 连接、注册读写事件、处理网络 IO。当某个批次发送完成成功或失败Sender会调用completeBatch()来处理结果成功则调用accumulator.deallocate()释放内存并触发用户的 Callback失败则判断是否可重试可重试就重新入队不可重试就标记为失败。关键点在于所有这些失败处理最终都只体现在 Callback 的onCompletion()方法里或者记录在RecordMetadata中。如果你调用send()时没有传 Callback也没有对返回的 Future 调用get()那么失败信息就真的“石沉大海”了。这就是“代码不报错但消息发送失败”最根本的原因。2.3 三种发送模式fire-and-forget、同步、异步回调Kafka Producer 提供了三种使用模式它们的可靠性依次递增但很多人默认用的是最不可靠的那种。Fire-and-forget 模式就是直接producer.send(record)不关心返回值。这种模式吞吐量最高但消息丢失时你完全不知道。它只适合那些允许丢数据的场景比如日志采集、埋点统计。同步发送模式是producer.send(record).get()。get()会阻塞当前线程直到发送完成如果失败会抛出ExecutionException里面包裹着真正的异常。这种模式可靠性高但吞吐量会大幅下降因为每条消息都要等一个网络往返。异步回调模式是producer.send(record, callback)。Callback 的onCompletion(RecordMetadata metadata, Exception exception)方法会在发送完成时被 Sender 线程调用。如果exception不为 null说明发送失败。这是生产环境最推荐的模式兼顾吞吐量和可靠性。我个人的经验是任何对数据可靠性有要求的场景都必须用异步回调模式并且在 Callback 里做失败记录或重试。Fire-and-forget 只应该出现在压测或者明确允许丢数据的场景。3. 静默失败的六大元凶从源码定位问题根因3.1 元凶一消息体超过大小限制这是最常见的静默失败原因。Kafka 有三个地方限制消息大小任何一个超限都会导致发送失败。参数所在端默认值作用max.request.sizeProducer1MB单个请求的最大字节数message.max.bytesBroker1MBBroker 能接收的单条消息最大值max.message.bytesTopic继承 BrokerTopic 级别的消息大小限制当消息体超过max.request.size时RecordAccumulator.append()会直接抛出RecordTooLargeException这个异常是同步抛出的send()调用处能捕获到。但如果消息体没超过 Producer 的限制却超过了 Broker 的message.max.bytes那么消息会被发出去但 Broker 会返回错误这个错误在 Callback 里以RecordTooLargeException的形式出现。如果你没传 Callback就完全感知不到。排查方法很简单在 Callback 里打印异常或者临时把max.request.size和 Broker 的message.max.bytes都调大验证。生产环境调整时要注意Broker 端参数修改后需要重启或动态更新Topic 级别可以用kafka-configs.sh动态调整。3.2 元凶二acks 设置为 0 导致的“假成功”acks参数控制的是 Producer 认为“发送成功”的标准。它有三个取值acks0Producer 把消息发出去就认为成功不等 Broker 确认。这种情况下即使 Broker 挂了、消息根本没写入Producer 也会返回成功的RecordMetadata。acks1Leader 副本写入本地日志后就返回确认。如果 Leader 在返回确认后、Follower 同步前宕机消息可能丢失。acksall或-1所有 ISR 副本都写入后才返回确认。可靠性最高。很多人为了追求吞吐量把acks设成 0然后发现消息“发送成功”但 Consumer 拉不到。这不是 bug是设计如此。acks0 时Producer 的“成功”只代表消息进入了网络缓冲区不代表 Broker 收到了。源码层面Sender.completeBatch()在处理响应时如果acks0NetworkClient根本不会等待响应直接调用completeBatch()并传入 null 异常。所以 Callback 里的exception永远是 nullRecordMetadata里的 offset 也是 -1。3.3 元凶三重试耗尽后的最终失败Kafka Producer 有一套重试机制由retries新版本用delivery.timeout.ms统一控制和retry.backoff.ms配合。当 Broker 返回可重试的错误如NotLeaderForPartitionException、NetworkExceptionProducer 会把批次重新入队等待retry.backoff.ms后再次发送。问题在于重试是在 Sender 线程里异步进行的重试最终失败时异常同样只出现在 Callback 里。如果重试次数耗尽或者delivery.timeout.ms到期批次会被标记为失败Callback 收到TimeoutException或最后一次的具体异常。delivery.timeout.ms这个参数很关键它默认是 1200002 分钟并且必须大于等于request.timeout.ms linger.ms。它的含义是从send()调用开始到最终成功或失败的总时间上限。超过这个时间无论重试多少次都会强制失败。很多“消息莫名消失”的案例其实是消息在重试队列里待了 2 分钟最终超时失败了而调用方早就继续往下走了。3.4 元凶四序列化异常被延迟抛出序列化是在send()的同步阶段完成的按理说序列化异常应该同步抛出。但这里有个细节KafkaProducer.send()里调用doSend()doSend()里先调用serialize()。如果序列化失败会抛出SerializationException这个异常确实会同步抛出。但是如果你用的是自定义 Serializer并且在serialize()方法里捕获了异常返回了 null 或者空字节数组那么消息会被正常发送但内容可能是错的。Consumer 端反序列化时才会报错。这种情况比较隐蔽排查时要检查 Serializer 的实现。还有一种情况Key 的序列化和 Value 的序列化是分开的。如果 Key 序列化失败异常会同步抛出但如果 Key 为 null 且 Serializer 不支持 null也会抛异常。这些都属于同步异常相对容易发现。3.5 元凶五分区器返回无效分区默认的DefaultPartitioner会根据 Key 的哈希值选择分区。如果 Key 为 null则使用粘性分区策略Sticky Partitioning在一段时间内固定发往同一个分区。但如果自定义了 Partitioner并且partition()方法返回了一个超出范围的分区号或者返回了 -1就会出问题。返回超出范围的分区号时RecordAccumulator.append()会抛出IllegalArgumentException这是同步异常。但如果 Partitioner 返回 -1Kafka 会认为需要重新计算分区在某些版本里可能导致消息被丢弃或行为异常。排查时可以在 Partitioner 里加日志确认返回的分区号在[0, numPartitions)范围内。3.6 元凶六缓冲区满导致的阻塞超时RecordAccumulator的内存由buffer.memory控制默认 32MB。当所有缓冲内存都被未发送的批次占用时新的send()调用会阻塞等待max.block.ms默认 60 秒。如果 60 秒内还没有内存释放会抛出TimeoutException。这个异常是同步抛出的send()调用处能捕获。但很多人把send()放在 try-catch 里却只 catch 了KafkaException而TimeoutException是InterruptedException的子类可能被漏掉。另外如果max.block.ms设得很大调用线程会长时间阻塞看起来像“卡死”而不是报错。4. 构建可靠 Producer 的实操方案4.1 参数配置清单一份可以直接抄的模板基于上面的分析我整理了一份生产环境可用的 Producer 配置模板。这份配置的核心思路是宁可发送慢一点也要保证每条消息的去向是明确的。# 基础连接配置 bootstrap.serverskafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092 key.serializerorg.apache.kafka.common.serialization.StringSerializer value.serializerorg.apache.kafka.common.serialization.StringSerializer # 可靠性核心配置 acksall enable.idempotencetrue retries2147483647 delivery.timeout.ms120000 request.timeout.ms30000 max.in.flight.requests.per.connection5 # 性能与缓冲配置 linger.ms20 batch.size32768 buffer.memory67108864 compression.typelz4 # 大小限制 max.request.size5242880这里重点解释几个参数的选择理由。acksall是可靠性的基础配合enable.idempotencetrue可以保证消息不重复、不丢失在单分区内。retries设为Integer.MAX_VALUE意味着无限重试但不用担心无限阻塞因为delivery.timeout.ms会在 2 分钟后强制失败。max.in.flight.requests.per.connection5是幂等性开启时的最大值再大就会破坏顺序保证。linger.ms20是我个人的经验值。默认是 0意味着消息来了立刻发送但这样会导致大量小批次吞吐量上不去。设成 20 毫秒让消息在缓冲区里攒一攒形成更大的批次吞吐量能提升好几倍而 20 毫秒的延迟对大多数业务来说完全可以接受。max.request.size52428805MB是为了容纳较大的消息体。但要注意Broker 端的message.max.bytes也要相应调整否则消息发出去也会被拒绝。4.2 异步回调的标准写法让失败无处可逃配置只是基础代码层面的处理才是关键。下面是一个标准的异步发送方法包含了完整的异常处理和日志记录。public void sendMessage(String topic, String key, String value) { ProducerRecordString, String record new ProducerRecord(topic, key, value); producer.send(record, (metadata, exception) - { if (exception ! null) { // 发送失败记录详细上下文 log.error(消息发送失败, topic{}, key{}, value{}, error{}, topic, key, value, exception.getMessage(), exception); // 根据异常类型决定后续动作 if (exception instanceof RecordTooLargeException) { // 消息太大重试也没用记录到死信表 saveToDeadLetterQueue(record, exception); } else if (exception instanceof TimeoutException) { // 超时可以尝试降级或告警 alertService.sendAlert(Kafka发送超时, topic); } else { // 其他异常记录到重试表 saveToRetryQueue(record, exception); } } else { // 发送成功记录元数据 log.debug(消息发送成功, topic{}, partition{}, offset{}, metadata.topic(), metadata.partition(), metadata.offset()); } }); }这段代码有几个关键点。第一Callback 里必须判断exception是否为 null这是区分成功和失败的唯一依据。第二日志要记录完整的上下文topic、key、value、异常堆栈否则事后排查时根本不知道哪条消息出了问题。第三要根据异常类型做不同的处理RecordTooLargeException重试无意义TimeoutException可能需要告警其他异常可以进入重试队列。注意Callback 是在 Sender 线程里执行的不要在 Callback 里做耗时操作比如同步写数据库否则会阻塞 Sender 线程影响整体吞吐量。正确的做法是把失败消息投递到一个内部队列由独立的线程池异步处理。4.3 消息大小的事前校验把问题挡在发送之前与其等发送失败再处理不如在发送前就校验消息大小。下面是一个简单的校验方法private static final int MAX_MESSAGE_BYTES 5 * 1024 * 1024; public boolean validateMessageSize(String value) { if (value null) { return true; } byte[] bytes value.getBytes(StandardCharsets.UTF_8); if (bytes.length MAX_MESSAGE_BYTES) { log.warn(消息体过大, size{}, max{}, bytes.length, MAX_MESSAGE_BYTES); return false; } return true; }这个校验的阈值应该和max.request.size以及 Broker 的message.max.bytes保持一致。如果消息确实需要超过这个大小可以考虑压缩compression.typegzip或lz4或者把大消息拆分成多条小消息或者把消息体存到对象存储Kafka 里只传引用。我踩过的一个坑是压缩是在 Producer 端做的但max.request.size检查的是压缩前的字节数。所以即使开了压缩如果原始消息超过max.request.size还是会被拒绝。这一点在配置时要特别注意。4.4 监控与告警让静默失败变得可见再好的代码也挡不住所有问题所以监控是最后一道防线。Kafka Producer 暴露了很多 JMX 指标其中最关键的是这几个指标名含义告警阈值建议record-error-rate发送失败速率大于 0 持续 1 分钟record-retry-rate重试速率持续高于正常水平request-latency-avg平均请求延迟超过 500msbuffer-available-bytes可用缓冲字节数低于总缓冲的 20%waiting-threads阻塞在 send() 的线程数大于 0record-error-rate是最重要的指标它直接反映了有多少消息发送失败。如果这个值大于 0说明有消息没发出去必须立即排查。buffer-available-bytes持续走低说明生产速度超过了发送速度可能需要增加buffer.memory或者增加分区数。这些指标可以通过 JMX 导出到 Prometheus再配合 Grafana 做可视化。如果不想搭这么重的监控至少要在应用层做一个简单的计数器定期打印发送成功和失败的数量。5. 常见问题速查与排查技巧实录5.1 问题排查速查表现象可能原因排查方法解决方案send() 不报错Consumer 无数据未传 Callback失败被吞加 Callback 打印异常使用异步回调模式Callback 收到 RecordTooLargeException消息超过大小限制检查消息体大小和三个限制参数调大限制或压缩/拆分消息Callback 收到 TimeoutException重试超时检查 delivery.timeout.ms 和网络调大超时或排查 Broker 状态send() 阻塞很久后抛 TimeoutException缓冲区满检查 buffer-available-bytes增加 buffer.memory 或分区数消息发送成功但 offset 为 -1acks0检查 acks 配置改为 acksall消息重复发送重试导致检查是否开启幂等性设置 enable.idempotencetrue消息顺序错乱重试 多 in-flight 请求检查 max.in.flight.requests设为 1 或开启幂等性5.2 三个独家避坑技巧技巧一用拦截器做统一的失败记录。与其在每个 Callback 里写重复的失败处理逻辑不如实现一个ProducerInterceptor在onAcknowledgement()方法里统一处理。这样业务代码只需要关注发送失败处理交给拦截器。拦截器的onAcknowledgement()也是在 Sender 线程执行的同样要注意不要做耗时操作。技巧二本地缓存 定时补偿。对于可靠性要求极高的场景可以在发送前把消息写入本地缓存比如 RocksDB 或本地文件发送成功后删除。如果发送失败或者应用崩溃重启后从本地缓存里恢复未发送的消息重新发送。这就是所谓的“本地消息表”模式能有效解决应用崩溃导致的消息丢失。技巧三压测时故意制造失败。很多人只在正常环境下测试上线后遇到网络抖动就手足无措。我的做法是在测试环境用工具模拟 Broker 宕机、网络延迟、磁盘满等场景观察 Producer 的行为和 Callback 的触发情况。只有亲眼见过失败长什么样才能在真正出问题时快速定位。5.3 一个真实的排查案例之前有个同事找我帮忙说他们的服务发送 Kafka 消息日志里没有任何异常但下游就是收不到数据。我让他把 Producer 的日志级别调到 DEBUG重启后观察日志发现大量这样的记录DEBUG o.a.k.clients.producer.internals.Sender - [Producer clientIdproducer-1] Got error produce response with correlation id 123 on topic-partition test-topic-0, retrying (2 attempts left). Error: NOT_LEADER_FOR_PARTITION原来他们的 Topic 在做分区扩容Leader 发生了切换Producer 一直在重试。但因为delivery.timeout.ms设的是默认的 2 分钟而他们的业务逻辑发送完消息后 1 分钟就去查结果了自然查不到。等 2 分钟后重试成功消息才真正写入但业务方早就认为失败了。这个案例的教训是delivery.timeout.ms要和业务逻辑的超时时间对齐。如果业务方 1 分钟后就认为操作失败那么 Producer 的重试超时就不应该超过 1 分钟否则会出现“业务认为失败但消息最终发送成功”的不一致状态。6. 从源码设计看 Kafka 的取舍哲学把上面这些内容串起来看Kafka Producer 的设计其实体现了一种明确的取舍用异步换吞吐用回调换可靠。它把发送动作和结果确认解耦让主线程可以快速返回继续处理业务这是 Kafka 能支撑高吞吐的根本原因。但代价就是开发者必须主动去处理异步结果否则就会遇到“静默失败”。理解了这一点很多看似奇怪的行为就说得通了。为什么send()不直接抛异常因为异常发生在另一个线程。为什么要有delivery.timeout.ms因为无限重试会阻塞缓冲区。为什么acks0时 offset 是 -1因为根本没有等待响应。我在实际项目中的体会是Kafka 的可靠性不是配置出来的是代码写出来的。再好的参数配置如果代码里不处理 Callback不记录失败日志不做补偿重试该丢的消息还是会丢。反过来即使参数配置不是最优只要代码层面有完善的失败处理机制消息的最终一致性就能得到保障。最后分享一个我常用的检查清单每次写 Producer 代码时过一遍有没有传 CallbackCallback 里有没有判断 exception失败日志有没有记录完整上下文有没有和业务超时对齐 delivery.timeout.ms监控里有没有 record-error-rate 的告警这五个问题都能答“是”基本就不会再遇到“代码不报错但消息发送失败”的情况了。