1. 从一条阻塞的产线告警说起：Kafka Sender线程到底在忙什么
凌晨两点被电话叫醒，监控大盘上某个核心业务模块的消息积压量在十分钟内从几百条飙到了几十万条，下游消费端开始出现大面积延迟。登录服务器抓了一份线程栈，发现业务线程几乎全部卡在同一个地方——等待一个叫Sender的线程把数据发出去。这个场景我相信做过消息中间件落地的同学都不陌生。很多人对 Kafka 生产端的理解停留在“调个send()方法就完事了”，但真正决定吞吐、延迟和稳定性的，恰恰是背后那个默默干活的Sender 线程。
这篇内容我想把 Kafka 生产端里 Sender 线程发送数据的完整链路掰开揉碎讲清楚。它适合已经用过 Kafka、但对其内部发送机制一知半解的开发者，也适合正在做性能调优、被积压和超时问题折磨的运维同学。核心要回答的问题很具体：一条消息从调用send()开始，是怎么被攒进缓冲区、怎么被 Sender 线程捞出来、怎么打包成请求、又怎么发到 broker 并处理响应的。搞懂这条链路，你才能明白batch.size、linger.ms、max.in.flight.requests.per.connection这些参数到底在调什么，也才能在出问题时知道该看哪个环节。
我会尽量用“一个合格从业者在真实项目里会怎么理解和排查”的视角来讲，而不是照搬源码注释。涉及源码的地方我会点出关键类和方法名，方便你自己去对照，但重点永远放在“为什么这么设计”和“实际怎么用”上。
2. 消息进入缓冲区之前：RecordAccumulator 的攒批逻辑
2.1 为什么不是来一条发一条
要理解 Sender 线程，得先理解它操作的对象——RecordAccumulator（记录累加器）。这是生产端的内存缓冲区，所有send()发出来的消息先进这里，而不是直接进网络。为什么要有这一层？因为逐条发送的网络开销太大了。每发一条消息都要经历一次 TCP 写入、一次请求头封装、一次 broker 端的请求解析，吞吐量根本上不去。
Kafka 的做法是把发往同一个分区（partition）的消息攒成一个批次（batch），凑够一定大小或者等够一定时间再一起发。这个“攒”的动作就发生在 RecordAccumulator 里。你可以把它想象成快递站的集包区：同一个目的地的包裹先堆在一起，够一车了再发车，而不是来一个包裹就发一辆车。
具体来说，RecordAccumulator 内部维护了一个ConcurrentMap<TopicPartition, Deque<ProducerBatch>>结构。每个 TopicPartition 对应一个双端队列，队列里放的是待发送的批次。当业务线程调用send()时，会先根据消息的目标分区找到对应的队列，然后尝试往队列尾部的那个批次里追加。如果尾部批次已经满了（达到batch.size，默认 16KB），或者压根没有可用批次，就会新建一个 ProducerBatch 追加到队列尾部。
2.2 批次的内存管理：BufferPool 的复用机制
这里有个容易被忽略但极其重要的细节：批次的内存不是每次新建都向 JVM 堆申请。Kafka 自己实现了一个BufferPool，按batch.size为单位预先分配好内存块，批次用完归还，下次复用。为什么要这么干？因为高频地创建和销毁 16KB 的字节数组会给 GC 带来巨大压力，尤其在高吞吐场景下，Young GC 会变得非常频繁。
BufferPool 的核心参数是buffer.memory（默认 32MB），它决定了整个生产端能用的缓冲区总大小。当缓冲区被占满、又没有可复用的内存块时，业务线程的send()就会阻塞，最长阻塞时间由max.block.ms（默认 60 秒）控制。这就是我在开头说的那个场景——业务线程卡住，本质上是缓冲区满了，Sender 线程发得太慢，腾不出内存。
提示：如果你在监控里看到业务线程大量处于
WAITING状态且堆栈指向BufferPool.allocate，基本可以断定是发送速度跟不上生产速度，要么加大buffer.memory，要么排查下游 broker 或网络是不是有瓶颈。
2.3 批次什么时候算“就绪”
一个批次不是攒满才能发。RecordAccumulator 里有个ready()方法，它遍历所有分区的队列，判断哪些批次的头部可以发送。判断条件有两个：一是批次已经满了（batch.size达到），二是等待时间超过了linger.ms。linger.ms默认是 0，意思是只要有批次就立刻发，不额外等待。很多同学为了提升吞吐会把它调到 5 到 100 毫秒，用一点延迟换更大的批次和更高的压缩率。
这里有个经验：linger.ms和batch.size要配合调。如果你只调大linger.ms但batch.size很小，批次很快就满了，等待时间根本用不上；反过来只调大batch.size但linger.ms是 0，批次可能还没攒够就被发走了。我一般建议高吞吐场景下batch.size设到 64KB 到 128KB，linger.ms设 10 到 20 毫秒，实测下来吞吐能提升好几倍，延迟增加也在可接受范围。
3. Sender 线程的主循环：一次完整的发送周期
3.1 run() 方法里的三段式结构
Sender 线程本身是一个Runnable，它的run()方法是一个while死循环，只要生产者没关闭就一直转。这个循环每一轮大致分三步走，我把它叫做“取数据、建请求、发网络”。
第一步是调用accumulator.ready()，拿到当前所有可以发送的批次，得到一个ReadyCheckResult，里面包含就绪的节点集合、是否还有未就绪的批次、以及下次检查的时间。第二步是调用accumulator.drain()，把就绪批次从各个队列里“抽”出来，按目标 broker 节点分组，形成Map<Integer, List<ProducerBatch>>。第三步是把这些批次封装成ProducerRequest，通过NetworkClient发出去。
这个循环的节奏很关键。如果没有任何批次就绪，Sender 不会空转烧 CPU，而是通过NetworkClient.poll()带超时地等待，超时时间由ready()返回的“下次检查时间”决定。这个设计很巧妙——它既保证了有数据时能及时发，又避免了无数据时的忙等。
3.2 drain 的抽取逻辑与“在途批次”概念
drain()这个方法值得单独说。它遍历所有分区的队列，把队首的批次取出来。但注意，取出来不等于从队列里删掉，而是标记为“在途”（in-flight）。为什么要标记？因为批次发出去之后，broker 可能返回成功，也可能返回失败需要重试。如果直接删了，重试就没数据了。
在途批次的管理和max.in.flight.requests.per.connection这个参数直接相关。它限制的是每个 broker 连接上同时未收到响应的请求数，默认是 5。这个值设大了吞吐高，但一旦发生重试，可能出现消息乱序——因为第一批失败了在重试，第二批却已经成功了。如果你对消息顺序有严格要求，要么把它设成 1（吞吐会掉），要么开启幂等性（enable.idempotence=true），让 broker 端帮你保证顺序。
我踩过的一个坑：早期做订单消息时没开幂等，max.in.flight用了默认的 5，结果网络抖动触发重试后，同一订单的“创建”和“取消”两条消息顺序反了，下游处理直接出错。后来开了幂等才彻底解决。所以顺序敏感的场景，这个参数千万别乱调。
3.3 请求的封装：从 ProducerBatch 到 ProduceRequest
批次分组完成后，Sender 会为每个 broker 节点构建一个ProduceRequest。这个请求里包含了多个分区的多个批次，是 Kafka 协议层面的二进制结构。构建过程中会做几件事：计算请求大小、做必要的压缩（如果配置了compression.type）、生成请求头（包含客户端 ID、请求版本等）。
这里有个细节：压缩是在批次级别做的，不是单条消息。所以批次越大，压缩率越高，这也是为什么调大batch.size能省网络带宽。常见的压缩算法有lz4、snappy、zstd、gzip。我的经验是，追求吞吐用lz4，追求压缩率用zstd，gzip太耗 CPU 一般不推荐。compression.type设成producer表示由生产端压缩，broker 原样存储。
请求构建好之后，交给NetworkClient.send()，它会把请求放进对应连接的发送队列，等待网络层真正写出去。注意，send()返回的是 Future，不代表已经发出去了，只代表请求已经入队。
4. 网络层与响应处理：数据真正离开客户端之后
4.1 NetworkClient 的非阻塞 IO 模型
Kafka 客户端用的是Java NIO，不是传统的阻塞 IO。NetworkClient底层维护了一个Selector，管理所有到 broker 的连接。Sender 线程调用poll()时，会做两件事：一是把待发送的请求通过 SocketChannel 写出去，二是检查哪些连接有响应可读，读回来并解析。
这个模型的好处是一个线程就能管理成百上千个连接，不需要为每个 broker 开一个线程。但代价是逻辑复杂，尤其是请求和响应的匹配。Kafka 用请求头里的correlationId来关联请求和响应——发出去的每个请求带一个自增 ID，响应回来时带着同样的 ID，InFlightRequests这个结构就靠它来匹配。
InFlightRequests还负责判断某个连接上是否还有未完成的请求。如果某个 broker 的未完成请求数达到了max.in.flight.requests.per.connection，Sender 就不会再往这个连接发新请求，避免压垮 broker 或造成过多的重试堆积。
4.2 响应回来之后的三种结局
响应回来后，根据 broker 返回的结果，批次会走向三种不同的处理路径。
第一种是成功。broker 返回的响应里会带上每个分区的写入结果，包括偏移量（offset）和时间戳。Sender 会调用completeBatch()，把批次标记为完成，归还内存到 BufferPool，并触发send()时注册的回调（Callback）。业务代码里send(record, callback)的那个 callback 就是在这里被调用的。
第二种是可重试的失败。比如网络超时、leader 切换、NotEnoughReplicas等。这类错误会触发重试，批次被重新放回队列（注意是放回队首，保证顺序），等待下一轮发送。重试次数由retries控制，默认是Integer.MAX_VALUE，配合delivery.timeout.ms（默认 2 分钟）来限制总时长。这里要强调：重试的总时间上限是delivery.timeout.ms，不是retries单独决定的，很多人只调retries却忽略了超时，结果消息在缓冲区里躺太久。
第三种是不可重试的失败。比如消息太大超过max.request.size、序列化错误、认证失败等。这类错误直接标记批次失败，回调里抛出异常，内存归还。
4.3 回调执行的线程归属问题
这里有个特别容易搞混的点：回调是在 Sender 线程里执行的，不是业务线程。也就是说，如果你在 callback 里做了耗时操作（比如写数据库、发 HTTP 请求），会直接拖慢 Sender 线程，进而影响整个生产端的吞吐。我见过有同学在回调里同步写日志到远程服务，结果 Sender 线程被卡住，消息积压暴涨。
正确的做法是回调里只做轻量级的记录或投递到另一个队列，重活交给专门的线程池。或者干脆用send()返回的Future，在业务线程里get()，但这样会阻塞业务线程，也不推荐在高吞吐场景用。
5. 参数调优实战：让 Sender 线程跑得更顺
5.1 吞吐优先型配置
如果你的场景是日志采集、埋点上报这类追求吞吐、对单条延迟不敏感的业务，核心思路是让批次尽量大、发送尽量批量。下面这套配置是我在多个项目里验证过的：
| 参数 | 推荐值 | 作用 |
|---|---|---|
| batch.size | 131072（128KB） | 增大批次，提升压缩率和网络效率 |
| linger.ms | 20 | 给批次留出攒批时间 |
| compression.type | lz4 | 压缩快、CPU 开销低 |
| buffer.memory | 67108864（64MB） | 加大缓冲区，减少阻塞 |
| acks | 1 | 只等 leader 确认，不等副本 |
| max.in.flight.requests.per.connection | 5 | 允许并行请求 |
这套配置下，单分区吞吐能到几 MB/s，多分区叠加后非常可观。注意acks=1会牺牲一点可靠性（leader 挂了可能丢数据），如果业务能接受就大胆用。
5.2 可靠性优先型配置
如果是订单、支付这类不能丢消息的场景，思路就反过来了，要保证不丢、不重、有序：
| 参数 | 推荐值 | 作用 |
|---|---|---|
| acks | all | 等所有 ISR 副本确认 |
| enable.idempotence | true | 开启幂等，保证不重不乱序 |
| retries | Integer.MAX_VALUE | 尽量重试 |
| max.in.flight.requests.per.connection | 5 | 开了幂等后可以放心用 5 |
| delivery.timeout.ms | 120000 | 控制总重试时长 |
开了幂等之后，max.in.flight可以放心设成 5，因为 broker 端会用序列号去重和排序，不会乱序。这是幂等性最大的价值之一。
5.3 一个真实的调优案例
之前有个项目，生产端吞吐一直上不去，监控显示 Sender 线程的poll()大部分时间在等待，但批次又经常没攒满就发了。排查后发现两个问题：一是linger.ms设成了 0，批次一有就发；二是batch.size只有默认的 16KB，稍微大点的消息一条就占满了。把linger.ms调到 10、batch.size调到 64KB 之后，吞吐直接翻了三倍，平均延迟只增加了不到 5 毫秒。
这个案例说明一个道理：Sender 线程的效率不取决于它自己多努力，而取决于上游给它的批次质量。批次太小、太碎，Sender 再快也没用，因为网络请求的数量摆在那里。
6. 排查 Sender 相关问题的完整链路
6.1 从现象到根因的排查顺序
遇到消息发送慢、积压、超时，我一般按这个顺序排查，从外到内逐层缩小范围：
- 看业务线程状态。如果大量卡在
BufferPool.allocate，说明缓冲区满，Sender 发得慢。 - 看 Sender 线程栈。如果卡在
NetworkClient.poll，说明在等网络响应，可能是 broker 慢或网络差。 - 看 broker 端指标。检查请求处理时间、磁盘 IO、ISR 收缩情况。
- 看网络指标。带宽是否打满、是否有丢包重传。
- 看配置。批次大小、linger、压缩、acks 是否合理。
这个顺序的逻辑是：先确认瓶颈在客户端还是服务端，再往下钻。很多同学一上来就调参数，结果方向错了白忙活。
6.2 几个高频问题的对症下药
问题一：消息发送延迟高，但吞吐正常。大概率是linger.ms设太大，或者acks=all等副本确认慢。前者调小 linger，后者检查 ISR 副本的健康状况。
问题二：偶发的TimeoutException。先看delivery.timeout.ms和request.timeout.ms的关系。request.timeout.ms是单次请求的超时，delivery.timeout.ms是包含重试的总超时，后者必须大于前者。如果网络抖动频繁，适当调大这两个值。
问题三：批次频繁重试导致吞吐下降。检查是不是max.in.flight设太大导致 broker 压力大，或者retries太激进。可以适当降低max.in.flight，或者排查 broker 端是否有热点分区。
问题四：内存持续增长不释放。检查是否有批次一直处于在途状态没回来。常见原因是 broker 端处理慢，或者delivery.timeout.ms设得太大，导致失败批次迟迟不释放内存。
6.3 一个容易被忽略的坑：批次大小与消息大小的关系
batch.size是批次的目标大小，但如果单条消息就超过了batch.size，这个批次会直接以单条消息的形式发送，不会等待攒批。更严重的是，如果单条消息超过max.request.size（默认 1MB），会直接抛RecordTooLargeException。我见过有同学把图片 base64 后塞进消息，结果单条几 MB，怎么调都发不出去。
解决办法有两个：一是调大max.request.size和 broker 端的message.max.bytes，二是把大消息拆成多条或者存到对象存储、消息里只放引用。后者是更推荐的做法，Kafka 本身不适合传大消息。
7. 我对 Sender 线程机制的一点个人理解
把 Sender 线程这条链路走通之后，我对 Kafka 生产端的设计有了更深的体会。它本质上是一个生产者-消费者模型：业务线程是生产者，往 RecordAccumulator 这个缓冲区里放数据；Sender 线程是消费者，从缓冲区里取数据发到网络。缓冲区的存在解耦了两者的速度差异，让生产端能应对短时的流量尖峰。
但这个模型有个天然的瓶颈：Sender 是单线程的。无论你有多少个分区、多少个 broker，最终都靠这一个线程去发。所以当分区数特别多、broker 特别多的时候，Sender 线程本身可能成为瓶颈。这也是为什么 Kafka 后来在高版本里对网络层做了不少优化，比如更高效的请求合并、更智能的批次调度。
实际用下来，我的建议是：不要试图把 Sender 线程逼到极限，而是通过合理的批次配置让它“轻松地高效工作”。批次攒得好，Sender 每轮循环发出去的请求少而大，网络利用率高，CPU 也省。反过来，如果批次又小又碎，Sender 疲于奔命，吞吐上不去，延迟还高。
最后分享一个我常用的观察方法：在测试环境把linger.ms从 0 逐步调到 50，同时压测观察吞吐和 P99 延迟的曲线。你会看到吞吐先快速上升然后趋于平缓，而延迟缓慢上升。那个“吞吐拐点”对应的 linger 值，往往就是你这个业务场景的最优解。这个方法比拍脑袋设参数靠谱得多，值得每个做消息中间件的同学试一次。
