本来以为生产者流程分析写到第4篇就差不多了结果真把代码一层层扒下去才发现从send()到消息真正落盘中间值得展开讲的东西还多得很。这次是“Kafka从入门到上天”系列第十四篇也是生产者流程分析的第5篇咱们把最后几块硬骨头啃完消息确认与重试机制、幂等与事务、元数据更新和缓冲区的联动以及生产环境里怎么盯指标、怎么排问题。往前翻第1到第4篇的时候我们已经把KafkaProducer启动、send/doSend、拦截器、序列化器、分区器、RecordAccumulator构造批次、Sender线程拉取这些环节一路捋了一遍。读到这里你应该已经清楚一条消息从业务代码出发先被包装成ProducerRecord经过各种处理器进入内存缓冲区再由后台Sender线程批量发到对应Broker的完整经过。不过很多人在自己写生产者的时候仍然会遇到超时、乱序、重复、性能上不去这些问题原因恰恰不是“发送”这一步本身而是发送背后的确认、重试、幂等、事务、元数据这些机制没有组合好。所以第5篇的核心就是把这些容易被忽视但直接影响结果的细节全部补上。建议阅读前先准备好Kafka 3.x版本的源码或者一份比较完整的producer配置文档边读边对照会更容易理解。1. 生产者流程收口前面几篇讲了什么这次补什么在动手调参之前先把整条链路在脑子里重新过一遍。1.1 一条消息从send()到“落盘”到底经过了哪些关卡一段非常简化的生产者内部流程是这样的KafkaProducer.send() - 拦截器链interceptor - 序列化器key/value - 分区器partitioner - RecordAccumulator 缓冲区 - 构建 ProducerBatch - 等待 batch 满 / linger.ms 超时 - 后台 Sender 线程 - 按节点聚合可发送的批次 - 检查/更新元数据如果分区 leader 未知 - NetworkClient 建立连接、发送请求 - 收到响应处理结果或触发重试 - 回调Callback通知业务方这条流程里前4篇我已经重点讲了三块一是send/doSend的执行骨架包括ProducerRecord怎么被组装、Future怎么返回二是拦截器、序列化器、分区器这些“加工环节”包括分区器怎么按照key散列、粘性分区是怎么让批次更紧凑的三是RecordAccumulator里内存池、ProducerBatch、BufferPool的细节包括为什么缓冲区能抗住高峰流量、batch.size和linger.ms是怎么影响批次大小的。第5篇接着往里走主要补四块内容acks与重试参数如何协同、幂等和事务机制如何工作、元数据更新和缓冲区阻塞如何拖慢或者保护整个发送链路、生产环境下通过哪些指标判断生产者的健康度以及最常见的异常该怎么排。1.2 生产者流程设计的三个关键思想在进入细节前值得先提三个贯穿始终的设计思想后面所有参数分析基本都围绕它们展开。第一是异步化。KafkaProducer.send()本身不会直接发网络请求而是把消息写进内存缓冲区由后台Sender线程批量发送。异步带来的好处是吞吐高不会每条消息都等一个RTT坏处是消息如果一直留在缓冲区里业务方感知不到它到底发送成功没有一旦长时间积压就会出现“send()很快但是消息全卡在队列里”的假象。所以生产上判断变慢第一步要看缓冲区积压而不是看send()耗时。第二是批次化。单条消息发网络请求成本非常高把多条消息攒成一个ProducerBatch再去发送能大幅度减少请求次数。batch.size、linger.ms、压缩算法等参数本质上都是在控制“什么时候把批次凑大点、凑满点”。第三是失败可重试。网络抖动、leader切换、Broker返回限流这些都不能简单当作“发送失败”直接丢弃。所以生产者设计了可重试错误和不可重试错误的区分配合retries、request.timeout.ms、delivery.timeout.ms这些参数来决定某条消息到底能在失败后挣扎多久。很多人没理解透的是重试不只是个“次数”问题还和乱序、重复、延迟强相关。只要把这三个思想拿住后面参数怎么配、指标怎么看不至于乱套。2. 消息确认acks、retries、max.in.flight三兄弟2.1 acks三档的语义和生产场景选择acks这个参数是生产者在投递消息时对Broker提出的确认要求。它有三个取值。acks取值控制端语义数据丢失风险延迟典型场景0发出去就算成功不等Broker任何确认高leader宕机等都会丢最低日志、监控指标等可容忍少量丢失的数据1等分区leader写入本地日志后返回确认中leader写完后崩溃且副本未同步时可能丢中大部分对吞吐有要求、接受极端情况下丢一条的数据all或-1等所有ISR内副本都写入成功后才返回确认低前提是min.insync.replicas配置合理最高订单、支付、交易流水等关键数据生产上我见过很多人把重要业务消息也配acks1理由是“反正我们有副本ISR里面有3个副本不可能丢”。但实际上acks1只等leader写入如果leader写入成功后马上宕机ISR里的follower不一定已经拉完这条消息副本切换后消息就可能丢了。所以关键业务要么acksall要么做好接受丢失的准备中间地带最危险。acksall也不是万能。它保证的是副本都写入了但如果min.insync.replicas配置成1其实和acks1差不多因为ISR里只有leader自己等所有ISR也就是等leader。所以对重要数据我通常建议同时配置acksall min.insync.replicas2min.insync.replicas是Broker端的参数意思是最少要有多少个ISR副本完成写入才能返回成功。两个参数配合才能把“确认强度”真正提上来。这里要注意如果可用副本数低于最小值写入会直接报NotEnoughReplicasException处理不当会引发大面积写入失败所以配置前要想清楚集群容错到底要什么。2.2 重试机制哪些错误会重试哪些直接送走Kafka把发送错误分成两类可重试retriable和不可重试non-retriable。可重试错误典型的有NETWORK_EXCEPTION网络瞬断、连接未建立等NOT_LEADER_FOR_PARTITION分区leader正在切换LEADER_NOT_AVAILABLEleader还没选出来UNKNOWN_TOPIC_OR_PARTITIONtopic刚刚创建元数据还没同步REQUEST_TIMED_OUTBroker处理超时这些错误的特点是过一小段时间后大概率能恢复重试一次也许就成功了。不可重试错误典型的有RECORD_TOO_LARGE消息体积超过Broker的message.max.bytes或topic的max.message.bytes限制INVALID_REQUIRED_ACKSacks配置不合法UNSUPPORTED_VERSION客户端和服务端协议版本不兼容INVALID_CONFIG配置非法这类错误无论如何重试都救不回来生产者会直接结束掉这条消息并触发异常回调不需要浪费重试名额。源码里判断可重试的方法是看异常是否实现了RetriableException接口排查时看日志里的异常类型也能区分。配置层面和重试相关的主要有三个参数retries重试次数上限默认值是Integer.MAX_VALUE实际被delivery.timeout.ms限制。3.0以后我建议显式配置一个合理值。retry.backoff.ms两次重试之间的间隔默认100ms是为了避免雪崩。request.timeout.ms单次请求超时时间默认30s指从发送请求到收到响应或连接失败的总体超时。真正限制消息生命周期的是另一个参数delivery.timeout.ms默认120s。它规定了消息从交给send()开始到最后一次尝试结束最多能撑多久。这三者配合的近似关系是delivery.timeout.ms request.timeout.ms retries * (request.timeout.ms retry.backoff.ms)实际计算没这么严格这是一个经验判断方式。如果request.timeout.ms是30sretries是5retry.backoff.ms是100ms那么理论最差情况下消息可以挣扎150s以上。如果delivery.timeout.ms设置成120s实际重试次数就会被delivery.timeout强制截断。所以不要只看retries超时问题排查一定要检查delivery.timeout.ms。实践上我的习惯是把delivery.timeout.ms当成“业务能等多长时间”的输入retries、request.timeout.ms、retry.backoff.ms决定重试节奏。比如一个高实时性场景消息超过10s没发出去就不等了我会设delivery.timeout.ms10000retries3request.timeout.ms2000retry.backoff.ms100。这样单次请求2s超时3次重试最多10s左右和期望接近。2.3 max.in.flight.requests.per.connection与消息乱序max.in.flight.requests.per.connection这个参数默认值是5含义是每个Broker连接上最多可以同时有多少个未确认的请求在途。in-flight请求多意味着不再需要等上一个请求返回才能发下一个网络利用率更高吞吐更好。但它和重试组合会引入一个经典问题乱序。设想一个场景没有开启幂等max.in.flight5retries0。生产者向同一分区先发了批次A后发了批次B。A因为网络抖动失败被放入重试B成功了。过了一会儿A重试也成功但此时B已经先被Broker写入。消费者先看到了B然后才看到A顺序就乱了。解决乱序有两个办法第一把max.in.flight.requests.per.connection设为1强制一个连接同时只能有一个在途请求自然保证顺序第二开启幂等也就是enable.idempotencetrue这样即使乱序到达Broker也会根据SequenceNumber拒绝旧的、断档的批次。毫无疑问推荐第二种因为第一种虽然保序但牺牲了吞吐而且3.0之后幂等已经默认开启。Kafka 3.0开始enable.idempotence默认是true并且对相关参数做了约束一次必须为acksall、retries0、max.in.flight5。也就是说新版生产者默认情况下就处于一种“不能被重试打乱数据”的保护状态。理解了这套约束再看源码里doSend中对这几个参数的校验就明白为什么异常消息里会明确告诉用户“must be set to ...”。3. 幂等和事务生产者流程里的“精确一次”利器重试保证了不丢消息但“重试”这两个字天生就带着重复风险网络超时后实际写入成功客户端又重试Broker收到两次一样的消息。这个时候就需要幂等机制。3.1 幂等生产者的工作原理enable.idempotencetrue时每个生产者进程在初始化时会向Broker申请一个PIDProducerId同时给发往每个分区的消息维护一个从0开始单调递增的SequenceNumber。这个消息的PID、分区号、SequenceNumber会一起放进请求Broker端会为每个PID分区校验序列号是否连续如果收到的序列号是上一条已写入消息的序列号1正常写入如果收到的序列号小于等于已写入的最大序列号说明是重复消息直接丢弃如果收到的序列号大于已写入的最大序列号1说明中间有消息丢了或者乱序到达返回异常并触发重试或者报错。这样即使客户端因为网络原因把同一个批次发了两次Broker也能识别出重复并丢弃第二次。生产者内部当某个批次发送成功后序列号会推进发送失败时会阻塞后续批次的重试不会让序号产生断档。不过要清楚幂等的边界只保证单个分区内的严格一次语义不保证跨分区原子性。只保证生产者进程存活期间的去重如果进程重启PID会变无法识别旧进程的重复消息。只保证生产者发送侧的重复应用逻辑内生成重复数据它管不了。所以幂等是事务的基础但它并不是“精确一次”的银弹。这也是为什么很多场景还需要事务机制。3.2 事务生产者的完整流程Kafka的事务解决的是“一批消息要么全部成功、要么全部失败”的问题并且可以跨多个分区、多个topic。生产者开启事务后会引入一个组件叫Transaction Coordinator事务协调者它专门负责记录事务状态并推进事务的提交或中止。开启事务需要配置transactional.idtxn-001 enable.idempotencetruetransactional.id由应用自己指定要求全局唯一。它和PID的作用不同PID会随进程重启变化而transactional.id是给事务协调者识别“同一个生产者逻辑实例”用的。协调者通过transactional.id维持事务的EPOCH纪元如果两个生产者进程使用了同一个transactional.id后者会把前者的epoch提升并中止旧实例尚在进行的旧事务避免两个实例同时写同一事务导致冲突。事务生产者的代码流程通常是这样producer.initTransactions(); try { producer.beginTransaction(); producer.send(record1); producer.send(record2); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); }落到内部流程上可以拆成几个步骤initTransactions()生产者会向事务协调者注册transactional.id获取PID并且协调者会标记该transactional.id对应的epoch为活动状态。如果之前有未完成事务这里会尝试中止掉让新实例能接管。beginTransaction()本地状态进入事务中此时后续send的消息都会被标记为“属于当前事务”。send()和普通发送一样走RecordAccumulator和Sender但是消息头里带了事务标记并且生产者会记录当前事务涉及了哪些分区。commitTransaction()生产者先向事务协调者提交“准备提交”PrepareCommit协调者会写入事务标记Transaction Marker然后通知参与的所有分区完成事务提交。只有所有涉及分区都提交完成commitTransaction才会返回。abortTransaction()如果业务出错生产者会通知协调者中止事务涉及的分区会标记事务为中止状态消费者在Read Committed模式下就不会读到这些消息。对于消费者如果想只消费已提交事务的消息需要设置isolation.levelread_committed如果保持默认的read_uncommitted消费者依然能看到事务中已经写入但最终被中止的消息适合对写入即时延迟更敏感、可以容忍脏读的场景。3.3 使用事务后的注意事项第一事务有额外开销。事务开启会引入transaction coordinator的多次交互涉及分区也要写入事务标记性能上有损耗。建议只对需要跨分区原子性或精确一次语义的业务开启事务不要把它当成默认配置。第二注意transaction.timeout.ms。事务超时默认60s如果事务执行时间事务内发送提交的整个过程超过这个值协调者会主动中止事务。业务代码里如果有耗时的外部RPC调用夹在beginTransaction和commitTransaction之间很容易踩坑。第三进程崩溃后的事务恢复。如果生产者事务执行中进程死了未提交事务会一直挂在那里直到事务超时被协调者强制中止。新的生产者使用相同transactional.id初始化时协调者会主动把旧事务中止掉所以不是死锁。但是恢复时间可能受事务超时时间影响等太久的话需要手动调整或者用运维命令查看事务状态。第四事务状态是记录在Kafka内部topic__transaction_state上的该topic的副本因子、清理策略也需要规划好否则事务量大了会成瓶颈。这一点在构建集群时就要考虑到别等线上出问题才想起来。4. 看不见的关键链路元数据更新与内存缓冲区生产者SDK看着不复杂但真正影响线上表现的两个“隐形环节”一个是元数据更新另一个是缓冲区与背压。很多人调了半天参数没效果就是没意识到问题出在这两个地方。4.1 Metadata生产者怎么知道消息该发到哪个Broker生产者发起请求前必须先知道目标topic有多少个分区、每个分区leader在哪个Broker上。这些信息存在客户端的metadata缓存里。缓存不是一开始就全量拉取的而是按需更新。当KafkaProducer创建时客户端会根据bootstrap.servers拿到连接信息此时metadata缓存是空的。第一次调用send()时如果分区信息未知send线程会触发一次metadata更新等待更新完成后才能把批次放入对应的分区批次中。这也是为什么测试环境只有一个Broker但第一次发送有时会感到短暂的阻塞。元数据的更新机制大概是这样的触发更新分区leader需要同步时、上层代码访问未知topic、缓存过期metadata.max.age.ms默认5分钟都会触发更新更新过程客户端向任意一个负载较轻的Broker发送MetadataRequest拉取目标topic的所有分区信息和对应的leader更新失败如果连不上Broker或超时客户端会看是否存在已有可用元数据没有的话这次send会一直阻塞直到max.block.ms默认60s超时抛TimeoutException。实际排查中最常见的“突然大量TimeoutException”就是元数据拉不到导致的。比如你新建了一个topic生产者在代码里写死topic名但第一次send时topic的leader还没就绪就会触发这类问题。解决办法也很简单建topic后做一次预热或者把topic创建提前到服务发布之前。这里补充一个细节topic.metadata.refresh.interval.ms这个参数在某些版本里也会影响元数据刷新频率如果发现生产者长时间拿不到新分区信息检查它是否被配得过大。4.2 RecordAccumulator背压、批次与吞吐的平衡术RecordAccumulator是整个生产者内存缓冲的核心。它的总内存大小由buffer.memory控制默认32MB。所有线程send()的消息都会放进同一个缓冲区Sender线程不断从中取走可发送的批次。当缓冲区满了新的send()调用会阻塞在accumulator上等待有空间或者等待超时。与之配合的max.block.ms默认60s如果60s还没有空间会抛TimeoutException并顺带抛出“Buffer pool limit was exceeded”之类的报错。这三个值——buffer.memory、max.block.ms、delivery.timeout.ms——组成了完整的背压机制。消费者或者下游处理慢或者Broker写入跟不上时缓冲区就会积压。如果积压到满生产者会开始阻塞业务线程避免无限制堆积内存如果一直阻塞超过max.block.ms就会直接失败避免业务假死。理解了这个调参就有方向了业务高峰期吞吐很大可以调大buffer.memory减少阻塞概率但要保证JVM内存能承受。常见做法是把buffer.memory调到64MB或128MB。如果业务可以接受短暂等待max.block.ms不要设得太小避免流量毛刺一来就报超时。batch.size和linger.ms是用来控制批次大小的。batch.size默认16KB如果单条消息就有10KB那一个batch只能放一两条发送效率很低需要调大batch.size如果单条消息只有几百字节那么默认16KB其实很合适。linger.ms默认0表示只要Sender有线程去取批次就立即发送不额外等待。如果把linger.ms调成5~10ms消息可以在缓冲区里多攒一会批次更大吞吐更高代价是延迟多了几毫秒。简单算一下假设单条消息1KB目标分区消息速率5万条/秒batch.size16KB每条消息1KB。理想情况下一个批次大约16条消息要达到5万条/秒每秒需要约3100个批次。如果linger.ms0不管batch有没有满Sender线程都会尽量快的把“当前已形成的批次”发出去。如果网络延迟是1ms单连接吞吐大约1000批次/秒那确实只能发1.6万条/秒左右。这时调大linger.ms5让批次在缓冲区多等5ms一个批次能攒到80条左右每秒需要发625个批次网络就能跟上。这个推演非常粗糙但能解释为什么有些场景掉吞吐就是因为批次太小、请求次数太多。5. 监控指标与调优别等线上出问题才想起来5.1 生产者端必须盯的几个指标Kafka生产者自带非常完善的metrics通过JMX或者Micrometer可以很方便接入监控。生产上我最常盯这几个指标名前缀producer-metrics含义什么情况需要关注record-queue-time-avg / record-queue-time-max消息在RecordAccumulator中等待的时间持续增长说明发送速度跟不上生产速度record-send-total累计发送消息总数用于核对业务量request-latency-avg请求从发出到收到响应的平均延迟偏高说明网络/Broker处理慢bytes-out-rate每秒发出字节数观察吞吐曲线buffer-available-bytes缓冲区可用字节接近0说明buffer_memory不足batch-size-avg平均批次大小远小于batch.size说明linger.ms太小或批次拆分严重connection-count连接数异常增长说明节点发现有问题producer-node-metrics 下的 request-latency-avg按Broker节点维度的延迟定位是不是单个节点慢我在实际项目中一般会配一个“生产者核心指标”看板重点看两条record-queue-time-avg和request-latency-avg。前者反映了应用侧到内存缓冲区的健康度后者反映了网络侧和Broker侧的健康度。如果queue-time暴涨问题在应用生产速度太快或缓冲区太小如果request-latency暴涨问题在Broker处理能力、网络、拉取leader的节点。5.2 一次“从每秒几千条到十万条”的调优复盘分享一个我调过的典型场景不涉及具体业务。当时某个数据同步应用将业务库变更事件按用户ID分区分批写入Kafka。最初配置基本是默认值加上了failOnInvalidLargeMessage之类的防御参数但实际吞吐只有几千条/秒而下游任务已经堆积到上百万条。排查过程是这样走的先看producer-metrics发现record-queue-time-avg涨到几百毫秒说明消息进缓冲区的速度远大于Sender发送速度缓冲区在持续积压。再看request-latency-avg只有2ms说明网络和Broker不是瓶颈问题可能在批次构造和发送效率。打开日志看到message.max.bytes相关异常说明单条消息虽然不大但某些字段偶尔膨胀到接近上限。通过调整序列化逻辑拆掉了无用的大字段单条消息从6KB降到1.5KB。调整linger.ms从0到5batch.size从16KB调到32KB单批次发送量从10多条涨到200多条请求数明显降低。由于消息按用户ID散列部分热门用户导致分区不均但这轮只做吞吐验证暂时没改分区逻辑。之后把压缩算法从none换成lz4bytes-out-rate平稳吞吐到了十万条/秒级别。这里想说的一点是不要一上来就堆并发。很多团队遇到吞吐不行第一反应是加线程。但KafkaProducer是线程安全的多线程调用send()可以但真正负责网络发送的是后台Sender线程瓶颈往往在批次大小和网络往返次数上。加线程未必有正面帮助反而会提高线程切换开销和buffer_memory竞争。先看指标定位瓶颈再动参数是最省事的路径。5.3 一份比较稳的生产者配置样例下面是一份我常用的生产者配置可以作为模板实际使用时根据消息体大小、延迟要求、数据重要性微调bootstrap.serversbroker1:9092,broker2:9092,broker3:9092 key.serializerorg.apache.kafka.common.serialization.StringSerializer value.serializerorg.apache.kafka.common.serialization.StringSerializer # 语义与重试 acksall enable.idempotencetrue retries5 retry.backoff.ms100 request.timeout.ms5000 delivery.timeout.ms30000 # 批量与缓冲 batch.size32768 linger.ms5 buffer.memory67108864 max.block.ms10000 # 吞吐与顺序 max.in.flight.requests.per.connection5 compression.typelz4 # 监控 client.idorder-service-producer这套配置的语义是数据不能丢、默认开启幂等、重试最多5次、整体发送生命周期最多30秒单批次目标32KB为吞吐牺牲一点点延迟缓冲区64MB可以抗住一定程度的流量毛刺。如果你的业务对延迟很敏感可以把linger.ms降为0batch.size降到16KB但吞吐会相应下降如果对吞吐很敏感可以把linger.ms调到10以上同时把compression.type换成zstd压缩率更高CPU消耗也更高。这里再补一个容易被忽略的点不要在生产者端用“每发送一条就flush一次”的写法。某些客户端实现里flush会强制让Sender把所有批次立即发送并等待这在低QPS下没什么问题但在高吞吐场景下会严重拖慢速度而且会让批处理形同虚设。如果确实需要确认发完一批用send回调统计或者批量确认。6. 常见问题与排查技巧实录生产环境里Kafka生产者的问题翻来覆去其实就那么几类。我把高频问题整理成一个速查表顺便补上排查思路能帮你少走不少弯路。6.1 写不进消息TimeoutException排查清单出现TimeoutException的方式有很多最常见的是这两种调用send()直接抛TimeoutException同时报Expiring 1 record(s) for xxx due to xxx ms has passed batch expiry deadline。这表示delivery.timeout.ms到期消息最终没发出去。调用future.get()时抛TimeoutException表示业务侧设置了future等待超时但消息可能还在缓冲区或传输中。排查步骤看这次异常是可重试错误还是不可重试错误。日志里如果出现LeaderNotAvailableException、NetworkException大概率是集群节点或网络抖动先确认Broker健康。看record-queue-time-avg如果偏高说明缓冲区积压严重。查看buffer.memory分配、下游消费速度和Broker处理能力。看是否有大量批次在等ack时超时。request.timeout.ms设置太长会让delivery.timeout.ms内能重试的次数变少设置太短又容易因为Broker短暂繁忙而误判超时。建议结合P99请求延迟去设置。检查metadata相关异常。如果topic不存在或分区leader长时间未就绪发送会持续阻塞直到max.block.ms超时。可以先在Kafka工具里查一下topic的ISR情况。不要忘记检查client.id对应的JMX指标。如果error-rate突然飙升辅助判断是哪个环节出错。6.2 消息乱序、重复到底是怎么出现的先说乱序。最容易出现乱序的配置组合是没有开启幂等同时设置了retries0和max.in.flight1。正如前面讲的多个批次同时in-flight后发的先成功、先发的后重试成功顺序就乱了。还有一类乱序是代码层面的业务线程先调用send(msg1)紧接着又调用send(msg2)但msg1因为delivery.timeout超时被丢弃msg2成功了这在单分区上也会表现为“收到msg2但永远没有msg1”。这种不算严格意义的乱序但消费端看到的效果一样需要靠delivery.timeout和业务补偿去处理。然后是重复。Kafka默认语义是at-least-once也就是不丢消息但可能重复。只要发生过网络重试就有重复风险。幂等机制可以消除“同一个PID下因重试导致的重复”但它消除不了“发送成功但是回调里误判失败导致业务重新生成了一条新消息”这种重复。所以做精确一次消费往往要结合消费者的幂等逻辑比如用业务唯一键去重和事务机制。排查重复问题时建议给消息增加业务唯一ID在消费端做去重不要指望Kafka本身帮你做到完全去重。6.3 事务挂起与恢复的实战处理用事务生产者的业务最容易遇到两种情况。第一种是事务执行时间超时。业务在beginTransaction()之后调用了外部接口结果外部接口耗时很长导致整个事务超过transaction.timeout.ms协调者强制中止事务应用侧再调用commitTransaction()就会报InvalidTxnStateException。解决思路有两个调大transaction.timeout.ms或者把耗时逻辑移到事务发起前/提交后。第二种是生产者进程崩溃导致事务悬挂。如果应用在commitTransaction()之前被杀掉这个事务就会一直挂着直到协调者检测到超时并中止。这个等待期对下游消费者很不友好如果下游在等这批数据延迟就会被拉长。恢复手段是让新的生产者使用同一个transactional.id启动初始化时协调者会主动清理旧事务。如果环境里多个实例同时抢同一个transactional.id调度器要保证同一时刻只有一个实例否则会出现事务反复被抢占中止的问题。处理这类问题我建议先看事务coordinator所在节点的日志确认有哪些活跃事务再决定是等待超时恢复还是手动中止。有些Kafka版本提供了事务运维命令可以通过命令行查看和操作事务具体命令以你使用的版本为准。6.4 一个定位“发送变慢”的典型路径最后分享一个通用定位流程可以套用在绝大多数“发送慢”的问题上打开producer-metrics先看record-queue-time-avg。异步场景下send()耗时无意义队列时间才是真实的积压信号。再看request-latency-avg和bytes-out-rate。延迟高、吞吐低先怀疑Broker压力和网络延迟正常、吞吐低先怀疑批次太小、linger.ms太短。看batch-size-avg。如果平均批次大小远小于batch.size说明消息根本攒不满要么linger.ms太小要么分区数太多导致消息被平均分散到了太多batch里。对照broker日志和JMX确认是不是有慢节点。可以通过producer-node-metrics下按节点维度看request-latency哪个节点延迟明显偏高就排查哪个节点。最后才调参。先调linger.ms和batch.size再调buffer.memory最后才考虑改客户端并发或改分区策略。改一次参数盯一段时间指标确认瓶颈到底移没移走。我自己在多个项目里都是这么排查的基本都能在半小时内把问题收敛到“网络/Broker/批次/缓冲区”这四类原因之一。写到这里Kafka生产者流程分析这个系列总算能画上一个阶段性的句号了。从最开始的send/doSend到最后的元数据、缓冲区、重试、幂等、事务、监控其实生产者设计的核心逻辑并不复杂异步攒批、网络发送、失败重试、有序去重再加上一层监控。真正复杂的是业务场景里对延迟、吞吐、语义的要求千差万别需要把这些参数组合成一套最适合自己的配置。我个人经验里最重要的一条调参之前先让监控指标说话别凭空猜测重试和数据丢失之间的平衡要靠delivery.timeout守住幂等和事务能解决重复和跨分区原子性但也要为性能买单。先把这几个边界想清楚再去看参数基本就不会翻车了。