一条消息背后其实只有两个线程
打开一个 KafkaProducer，你会发现它没有线程池，没有 reactor 组，也没有一堆 worker。
它只有两个线程：
- 调用线程。你的业务代码在哪个线程里调
send()，工作的前半段就在哪个线程里完成。 - I/O 线程。名字是
kafka-producer-network-thread | <client.id>，跑的是Sender.run()。
4.3.0 的构造器里，这一行就是全部的证据：
// KafkaProducer.java:467-469
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new Sender.SenderThread(ioThreadName, this.sender, true);
this.ioThread.start();
第三个参数 true 是 daemon 标志。也就是说，如果应用只关了主线程、没有显式 close() 生产者，这个 I/O 线程不会阻止 JVM 退出——它会随着进程一起消失，缓冲区里没发出去的批次也就一起消失了。
NETWORK_THREAD_PREFIX 常量定义在同文件第 245 行，值就是 "kafka-producer-network-thread"。你在 jstack 里看到的名字，前 28 个字符来自这里，竖线后面是你配的 client.id。
数量上的结论很硬：无论生产者被多少个业务线程共用，I/O 线程永远只有一个。所以生产者侧的并行度不来自线程数，而来自「一次请求里塞了多少个分区的批次」。
send() 在源码里只有两行
send(record, callback) 的方法体一共三条语句：
// KafkaProducer.java:958-963
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
// intercept the record, which can be potentially modified; this method does not throw exceptions
ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
return doSend(interceptedRecord, callback);
}
这两行值得单独拎出来说，因为它纠正了一个很常见的印象：拦截器不在 doSend 里，而在 doSend 之前。
后果有两个：
- 拦截器能改的是「即将进入发送流程的记录」，包括 key、value、headers、甚至分区号；它改不了元数据等待、序列化之后的结果。
- 注释写明
onSend不会抛异常。ProducerInterceptors.onSend内部会捕获拦截器抛出的异常并记日志，然后把原记录传下去。所以「我的拦截器挂了，消息发不出去」这种情况不会发生；相反，是「我的拦截器挂了，但它没生效」，问题更隐蔽。
doSend 才是真正的实现。它做六件事，顺序不能乱。
doSend 的六个步骤
第一步，确认生产者还活着：
// KafkaProducer.java:967-970
private void throwIfProducerClosed() {
if (sender == null || !sender.isRunning())
throw new IllegalStateException("Cannot perform operation after producer has been closed");
}
这里的 sender.isRunning() 读的是 Sender 里那个 volatile boolean running。所以「关了没有」这件事在跨线程时是可见的，不需要加锁。
第二步，等元数据：
// KafkaProducer.java:985-996
long nowMs = time.milliseconds();
ClusterAndWaitTime clusterAndWaitTime;
try {
clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);
} catch (KafkaException e) {
if (metadata.isClosed())
throw new KafkaException("Producer closed while send in progress", e);
throw e;
}
nowMs += clusterAndWaitTime.waitedOnMetadataMs;
long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
Cluster cluster = clusterAndWaitTime.cluster;
这一步是调用线程第一次可能被卡住的地方。它要拿到「这个 topic 有几个分区、每个分区的 leader 是谁」。缓存里有就直接返回；没有就得向集群请求，最多等到 max.block.ms。
注意 remainingWaitMs 的算法：元数据等待花掉的时间，会从后面申请缓冲的预算里扣掉。max.block.ms 是整个 send() 的总预算，不是每个阻塞点各给一份。这一点在 7.2 节讲缓冲池时还会用到。
第三步，序列化。key 和 value 分别序列化，ClassCastException 会被包成 SerializationException——因为泛型在运行期擦除，序列化器拿到的类型对不上时，报出来的原始异常是 ClassCastException，直接抛给用户会很难懂。
第四步，定分区：
// KafkaProducer.java:1017
int partition = partition(record, serializedKey, serializedValue, cluster);
如果记录里显式指定了分区，直接用；如果配了自定义 partitioner.class，调它；否则返回 RecordMetadata.UNKNOWN_PARTITION，交给累积器里的内置分区器（粘性分区）去挑。3.3 节讲过粘性分区的行为，这里只强调一点：分区号可能不是在这一行定下来的，真正定下来的位置在 RecordAccumulator.append 内部。所以后面那行 assert appendCallbacks.getPartition() != RecordMetadata.UNKNOWN_PARTITION 才成立。
第五步，算大小并校验：
// KafkaProducer.java:1019-1025
setReadOnly(record.headers());
Header[] headers = record.headers().toArray();
int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(RecordBatch.CURRENT_MAGIC_VALUE,
compression.type(), serializedKey, serializedValue, headers);
ensureValidRecordSize(serializedSize);
long timestamp = record.timestamp() == null ? nowMs : record.timestamp();
ensureValidRecordSize 比的是单条记录估算大小与 max.request.size（默认 1048576 字节）。超了直接抛 RecordTooLargeException。注意它估的是单条，而请求是按批次发的，一批里可以有多条——所以「单条不超」不代表「批次不超」，后者由 broker 的 message.max.bytes 管。
第六步，进累积器并唤醒 I/O 线程：
// KafkaProducer.java:1027-1030
// Append the record to the accumulator. Note, that the actual partition may be
// calculated there and can be accessed via appendCallbacks.topicPartition.
RecordAccumulator.RecordAppendResult result = accumulator.append(record.topic(), partition, timestamp, serializedKey,
serializedValue, headers, appendCallbacks, remainingWaitMs, nowMs, cluster);
// KafkaProducer.java:1041-1045
if (result.batchIsFull || result.newBatchCreated) {
log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), appendCallbacks.getPartition());
this.sender.wakeup();
}
return result.future;
wakeup() 的实现只有一行 this.client.wakeup()（Sender.java:963-965）。它的作用是让 I/O 线程从 Selector.poll() 里立刻返回，不用等超时。这是「应用线程 → I/O 线程」之间唯一的信号机制，没有队列、没有锁、没有条件变量。
异常分两路，这是最该记住的一条
doSend 的 catch 块写得很清楚：
// KafkaProducer.java:1046-1061
// handling exceptions and record the errors;
// for API exceptions return them in the future,
// for other exceptions throw directly
} catch (ApiException e) {
log.debug("Exception occurred during message send:", e);
if (callback != null) {
TopicPartition tp = appendCallbacks.topicPartition();
RecordMetadata nullMetadata = new RecordMetadata(tp, -1, -1, RecordBatch.NO_TIMESTAMP, -1, -1);
callback.onCompletion(nullMetadata, e);
}
this.errors.record();
this.interceptors.onSendError(record, appendCallbacks.topicPartition(), e);
if (transactionManager != null) {
transactionManager.maybeTransitionToErrorState(e);
}
return new FutureFailure(e);
规则是：
| 异常类型 | 例子 | 表现 |
|---|---|---|
ApiException 及其子类 | TimeoutException、RecordTooLargeException、BufferExhaustedException | 不抛，包进返回的 Future |
KafkaException | 生产者关闭、内部状态错误 | 当场抛 |
SerializationException | 序列化器返回 null 或类型不符 | 当场抛（它直接继承 KafkaException，不是 ApiException） |
其它 Exception | 拦截器之外的 NPE 等 | 当场抛 |
TimeoutException 属于 ApiException，所以「元数据等不到」不会在 send() 这一行抛出来，而是让你在 future.get() 时收到。7.2 节会实测到同一条规律用在 BufferExhaustedException 上：阻塞发生在 send()，异常出现在 Future。这两个位置分开，是很多人写 let-it-crash 逻辑时踩空的地方。
回调跑在哪个线程
这是本节最重要的实测。四个业务线程并发 send()，在回调里同时打印「调用线程名」和「回调线程名」：
for (int t = 0; t < 4; t++) {
Thread th = new Thread(() -> {
String caller = Thread.currentThread().getName();
producer.send(new ProducerRecord<>(TOPIC, "k", "payload"),
(md, ex) -> System.out.println("CALLBACK caller=" + caller
+ " callbackThread=" + Thread.currentThread().getName()
+ " partition=" + md.partition() + " offset=" + md.offset()));
}, "app-thread-" + t);
th.start();
}
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：
MAIN calling thread = main
PRODUCER ioThread = kafka-producer-network-thread | b2-thread-model
CALLBACK caller=app-thread-0 callbackThread=kafka-producer-network-thread | b2-thread-model partition=2 offset=1
CALLBACK caller=app-thread-3 callbackThread=kafka-producer-network-thread | b2-thread-model partition=2 offset=2
CALLBACK caller=app-thread-0 callbackThread=kafka-producer-network-thread | b2-thread-model partition=2 offset=3
CALLBACK caller=app-thread-1 callbackThread=kafka-producer-network-thread | b2-thread-model partition=2 offset=4
CALLBACK caller=app-thread-1 callbackThread=kafka-producer-network-thread | b2-thread-model partition=2 offset=5
CALLBACK caller=app-thread-3 callbackThread=kafka-producer-network-thread | b2-thread-model partition=0 offset=0
CALLBACK caller=app-thread-2 callbackThread=kafka-producer-network-thread | b2-thread-model partition=0 offset=1
CALLBACK caller=app-thread-2 callbackThread=kafka-producer-network-thread | b2-thread-model partition=2 offset=6
PROBE futureDoneRightAfterSend=false waitForAckMs=159
THREADS kafka-producer-network-thread | b2-thread-model
四个 app-thread-* 全部把回调交给了同一个 kafka-producer-network-thread。整个进程里与生产者相关的线程只有这一个。
顺带两个数据点：
futureDoneRightAfterSend=false：send()返回时 Future 一定没完成。这是「异步」这个词最实在的证明。从send()到 ack 用了 159 ms（单次运行，随调度波动，不作基准）。- 分区是 2,2,2,2,2,0,0,2 这样成片出现的，不是均匀轮询——粘性分区在起作用，3.3 节讲过。
回调在 I/O 线程上跑，含义很直接：回调里的任何阻塞，都会拖住所有分区的发送。源码的 Javadoc 也写了这句警告，大意是回调应该足够快，否则会延迟其他线程的消息发送；需要阻塞或重计算就用自建 Executor 转一手。
线程安全边界到底在哪
先说一个可能让很多人意外的实测结果：4.3.0 的 KafkaProducer.java 里，synchronized 关键字出现 0 次。
unzip -p ~/.m2/repository/org/apache/kafka/kafka-clients/4.3.0/kafka-clients-4.3.0-sources.jar \
org/apache/kafka/clients/producer/KafkaProducer.java > /tmp/KafkaProducer.java
grep -c "synchronized" /tmp/KafkaProducer.java
本机输出 0。生产者能被多线程共用，靠的不是门面上的一把大锁，而是把共享状态分散到几个各自负责同步的组件里：
| 共享对象 | 同步方式 | 源码位置 |
|---|---|---|
每个分区的 Deque<ProducerBatch> | 以 deque 自身为 monitor | RecordAccumulator.java:314、:340 |
BufferPool | ReentrantLock + Condition | BufferPool.java:107 起 |
Metadata 的快照 | volatile MetadataSnapshot，读不加锁 | Metadata.java:79、:136 |
| 元数据的写路径 | 方法级 synchronized（24 处） | Metadata.java:153 起 |
Sender.running / forceClose | volatile 布尔 | Sender.java:548、:556 |
这个设计的好处是：两个业务线程往不同分区发消息时，它们争的是两把不同的 monitor，不会互相排队；只有当它们恰好命中同一个分区时才串行。I/O 线程读元数据快照也不加锁，所以应用线程的发送不会被元数据刷新卡住。
坏处是「线程安全」有前提，用错了照样出事：
- 不要跨线程复用同一个
ProducerRecord。accumulator.append会把serializedKey、serializedValue、headers数组直接塞进批次，批次异步写出去。你回头改记录里的对象，改的可能是还在缓冲区里的数据。 headers的处置要看清对象。doSend第 1019-1020 行对传进去的那份 headers 调了setReadOnly。但ProducerRecord的构造器会先复制一份（ProducerRecord.java:83：this.headers = new RecordHeaders(headers);），所以被冻结的是内部副本。
本机实测（同上环境）：send() 之后继续往自己持有的 RecordHeaders 里 add，没有抛异常，条数从 1 变 2。也就是说，你手里的对象和生产者手里的对象不是同一个，改了不会影响已入队的记录，也不会被冻结拦住。想把「发出去之后就不能改」当成约束，得靠代码规范，不能靠这个 setReadOnly。
send()之后关闭生产者，再发就是IllegalStateException。本机实测原文：
SEND-AFTER-CLOSE java.lang.IllegalStateException: Cannot perform operation after producer has been closed
close() 为什么不是「加一把锁」
很多资料写「close() 要加锁，防止和 send() 并发」。这个说法在 4.3.0 的源码里对不上——上面数过了，整个类没有 synchronized。
真实的关闭流程是这样的：
// KafkaProducer.java:1398-1417（节选，方法整体为 1390-1443）
boolean invokedFromCallback = Thread.currentThread() == this.ioThread;
if (timeoutMs > 0) {
if (invokedFromCallback) {
log.warn("Overriding close timeout {} ms to 0 ms in order to prevent useless blocking due to self-join. " +
"This means you have incorrectly invoked close with a non-zero timeout from the producer call-back.",
timeoutMs);
} else {
// Try to close gracefully.
final Timer closeTimer = time.timer(timeout);
clientTelemetryReporter.ifPresent(ClientTelemetryReporter::initiateClose);
closeTimer.update();
if (this.sender != null) {
this.sender.initiateClose();
closeTimer.update();
}
if (this.ioThread != null) {
try {
this.ioThread.join(closeTimer.remainingMs());
} catch (InterruptedException t) {
把它拆成状态机看：
sender.initiateClose()：先accumulator.close()，再running = false，再wakeup()。顺序是源码注释里特意强调的——先关累积器，保证 I/O 线程跳出主循环后不会再有新的 append 进来，否则那些批次的回调会永远等不到。ioThread.join(剩余时间)：等 I/O 线程把存量发完。注意 I/O 线程在running=false之后并没有立刻退出，它还有两个收尾循环（Sender.java:258、:267）：一个等「累积器里还有没抽干的批次，或还有在途请求，或有未完成事务」，一个处理事务中止。所以优雅关闭的时间下限由存量决定，不是零。- join 超时，就
forceClose()：置forceClose = true，I/O 线程放弃剩余批次，accumulator.abortIncompleteBatches()把它们的 Future 全部标记失败，然后client.close()。
那为什么不能用一把锁包住 close()？因为回调运行在 I/O 线程上。如果 close() 拿锁等 join，而 I/O 线程此刻正在执行你写的回调、回调里又调了 close()，就成了自己等自己。源码不用锁解决，用线程身份判断解决：Thread.currentThread() == this.ioThread 命中时，把超时改成 0，直接进入强制关闭路径，只做状态切换不做阻塞等待。
flush() 用的是同一个判断（KafkaProducer.java:1216-1219），只不过它的处理更硬——直接抛异常：
// KafkaProducer.java:1216-1220
public void flush() {
if (Thread.currentThread() == this.ioThread) {
log.error("KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.");
throw new KafkaException("KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.");
}
这两条路径本机都跑过，输出如下（同一份程序）：
[kafka-producer-network-thread | b2-thread-model] ERROR org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=b2-thread-model] KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.
FLUSH-IN-CALLBACK org.apache.kafka.common.KafkaException: KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.
[kafka-producer-network-thread | b2-thread-model] WARN org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=b2-thread-model] Overriding close timeout 5000 ms to 0 ms in order to prevent useless blocking due to self-join. This means you have incorrectly invoked close with a non-zero timeout from the producer call-back.
CLOSE-IN-CALLBACK returned, cbThread=kafka-producer-network-thread | b2-thread-model
flush() 立刻失败，close(5s) 被降级成 0 再返回。两个行为都来自同一个判断，也都在日志里留下了原文。
线程归属表
把前面的事实压成一张表，遇到并发问题时按这张表定位：
| 动作 | 在哪个线程 | 会不会阻塞 | 源码依据 |
|---|---|---|---|
send() 的拦截器链 | 调用线程 | 不会抛异常 | KafkaProducer.java:961 |
| 等元数据（首次或过期） | 调用线程 | 最多 max.block.ms | :988 |
| 序列化 | 调用线程 | 取决于你的序列化器 | :997-1017 |
| 分区计算 | 调用线程（无 key 时可能在累积器内） | 否 | :1017、RecordAccumulator.java:301 |
| 申请缓冲 | 调用线程 | 最多剩余 max.block.ms | RecordAccumulator.java:333 |
| 攒批、判满、按节点抽取 | I/O 线程 | 每次 runOnce() 一轮 | Sender.java:310、:380 |
| 发请求、收响应 | I/O 线程 | poll 最长到下次到期时间 | Sender.java:345 |
| 用户回调 | I/O 线程 | 阻塞它会拖慢全部发送 | 实测 |
close() 等待 | 调用线程（回调内则降级） | 到超时为止 | :1390-1443 |
回到 libraryd
libraryd-mq 的 NaiveProducer 一开始是这样用的：一个 Spring 定时任务里 send()，另一个请求线程里也 send()，共用一个静态 KafkaProducer。这个用法本身没问题——生产者是线程安全的，前提是两边不共享 ProducerRecord，也不在回调里做重活。
真出问题的是回调。有人在 onCompletion 里写库、查库、再发一条下游事件。这些操作全部落在 kafka-producer-network-thread 上，一慢就把所有分区的发送拖住，buffer.memory 很快被填满，然后 send() 开始阻塞——应用线程和 I/O 线程在同一个环里互相等。7.2 节会把这条链条上的每个环节用数字补全。
本节只回答一个问题：这条消息在 send() 返回之前，被谁碰过、在哪个线程上碰的。答案是：调用线程碰了拦截器、元数据、序列化、分区和入队；I/O 线程碰了剩下的一切，包括你的回调。
阅读导航：上一节 6.3 KIP-848 新一代消费组协议 ｜ 下一节 7.2 RecordAccumulator 与批次缓存
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。