《Java 消息中间件·Kafka 内核与客户端》11.1 网络线程与请求队列

从一条 Produce 请求说起

生产者把一批消息发到 broker，网络包到达网卡之后，Kafka 并没有「一个连接一个线程」地去处理。它把网络读写和请求处理拆成两组线程，中间隔着一个有界队列。理解这个结构，才能回答三个日常问题：

• num.network.threads 和 num.io.threads 到底谁在干什么？
• 请求排队排在哪里，看哪个指标？
• 为什么同一条连接上的请求不能并行处理？

这一节用一个两节点 KRaft 集群把这些结构拆开。先给结论：网络线程只做非阻塞的收发与协议解析，真正的请求处理在另一组线程里；两者之间是一个容量为 queued.max.requests 的 ArrayBlockingQueue；一条连接在请求入队后立刻被 mute，直到响应发回才 unmute。这三点决定了 broker 的过载行为。

实验环境与一个容易忽略的版本细节

本节的集群是两节点 KRaft，两个节点都承担 broker 与 controller：

docker network create mqc-net
docker run -d --name mq-c1 --network mqc-net -p 19221:19092 \
  -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS='PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:19092' \
  -e KAFKA_ADVERTISED_LISTENERS='PLAINTEXT://mq-c1:9092,EXTERNAL://127.0.0.1:19221' \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP='CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT' \
  -e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS='1@mq-c1:9093,2@mq-c2:9093' \
  -e KAFKA_HEAP_OPTS='-Xmx384m -Xms384m' apache/kafka:4.3.0

mq-c2 用同样的参数，只改 KAFKA_NODE_ID、容器名和对外端口。容器内执行 CLI 一律用 127.0.0.1:9092（PLAINTEXT 内部 listener），宿主机客户端才用映射端口。

这里先记一个实测细节：镜像 apache/kafka:4.3.0 内部运行的是 OpenJDK 21.0.11，不是我们写客户端用的 Temurin 25：

$ docker exec mq-c1 /opt/java/openjdk/bin/java --list-modules | grep management
java.management@21.0.11
java.management.rmi@21.0.11

所以本节所有 JMX 与线程数据都来自这个 JRE，而不是宿主机的 25。客户端的版本断言仍以 AppInfoParser.getVersion() 为准，两者不要混。

四道关：acceptor、processor、队列、handler

Kafka 的服务端网络层可以画成一条直线：

客户端连接
   │
   ▼
acceptor（每个 listener 一个）
   │  只做 accept，把新连接分给某个 processor
   ▼
processor（每个 listener 有 num.network.threads 个）
   │  非阻塞读写、解析请求头与请求体
   ▼
RequestChannel 的阻塞队列（容量 queued.max.requests）
   │
   ▼
KafkaRequestHandler（num.io.threads 个）
   │  真正调用 API 层：写日志、读日志、处理元数据
   ▼
processor 把响应写回连接

关键的分工是：processor 永远不做会阻塞的磁盘操作。它只负责把完整的请求拼出来、丢进队列，然后继续 poll 其他连接。磁盘写、等待副本、读日志都在 handler 线程里做。这个设计让少量网络线程可以服务成千上万条连接。

源码里能看到这个边界。RequestChannel 用一个有界数组队列保存待处理请求：

// kafka/network/RequestChannel.scala（Kafka 4.3.0）
private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
metricsGroup.newGauge(RequestQueueSizeMetric, () => requestQueue.size)

def sendRequest(request: RequestChannel.Request): Unit = {
  requestQueue.put(request)
}

put 而不是 offer 很关键：队列满时 processor 会被阻塞，而不是丢请求或报错。这个细节下一节展开。

实测：一个 broker 到底有多少线程

先看 broker 启动时打印的生效配置。这些值不是查文档抄的，是从容器内的 server.log 里读出来的：

$ docker exec mq-c1 grep -nE 'num.io.threads|num.network.threads|queued.max.requests' \
    /opt/kafka/logs/server.log | head -3
377:	num.io.threads = 8
378:	num.network.threads = 3
397:	queued.max.requests = 500

三个 listener（PLAINTEXT、EXTERNAL、CONTROLLER）各有一套 acceptor 和 processor。用 JMX 的 DiagnosticCommand.threadPrint 把线程名完整打出来后，可以精确数出来：

16 个 handler 是最容易看错的地方。线程名只有 request-handler-0 到 request-handler-7，但每个名字出现两次。源码解释了原因：handler 池是按节点角色创建的，nodeName 取 broker 或 controller，两个池都用同一套命名规则：

// kafka/server/KafkaRequestHandler.scala（Kafka 4.3.0）
private val perPoolIdleMeterName = if (nodeName == "broker") {
  "BrokerRequestHandlerAvgIdlePercent"
} else if (nodeName == "controller") {
  "ControllerRequestHandlerAvgIdlePercent"
} else {
  throw new IllegalArgumentException("Invalid node name:" + nodeName)
}
...
KafkaThread.daemon("data-plane-kafka-request-handler-" + id, runnables(id)).start()

合并模式（process.roles=broker,controller）下，broker 侧和 controller 侧各有一个 8 线程的池，所以 num.io.threads=8 实际产生了 16 个 handler 线程。JMX 里也能看到两个独立的池指标：

kafka.server:type=KafkaRequestHandlerPool,name=BrokerRequestHandlerAvgIdlePercent
kafka.server:type=KafkaRequestHandlerPool,name=ControllerRequestHandlerAvgIdlePercent
kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent

最后一个是两个池的聚合值。看监控时如果只看到聚合指标，会把 controller 侧的空闲也算进去，判断 broker 侧是否繁忙时要看 BrokerRequestHandlerAvgIdlePercent。

把三族线程相加：3 + 9 + 16 = 28 个 SocketServer 线程。JVM 的 Java 线程总数是 104（压测采样时为 105，随连接与调度小幅波动），操作系统层面的线程数是 115：

$ docker exec mq-c1 grep '^Threads:' /proc/1/status
Threads:	115
$ # JMX java.lang:type=Threading 的 ThreadCount
104

两个数字不相等是正常的：/proc/<pid>/status 统计的是 OS 线程，包含 GC、JIT 编译器等 VM 内部线程；JMX 的 ThreadCount 只统计 Java 线程。做容量评估时要说清楚用的是哪个口径，否则「线程数暴涨」的结论可能只是 JIT 线程在活动。

mute/unmute：一条连接同一时刻只有一个请求在处理

请求入队之后，processor 做的下一件事不是继续读这条连接，而是把它 mute 掉：

// kafka/network/SocketServer.scala，Processor.processCompletedReceives
requestChannel.sendRequest(req)
selector.mute(connectionId)
handleChannelMuteEvent(connectionId, ChannelMuteEvent.REQUEST_RECEIVED)

响应发完之后才尝试恢复：

// Processor.processCompletedSends
handleChannelMuteEvent(send.destinationId, ChannelMuteEvent.RESPONSE_SENT)
tryUnmuteChannel(send.destinationId)

KafkaChannel 用一个显式的状态机记录这件事，状态不止「静音/不静音」两种：

// org/apache/kafka/common/network/KafkaChannel.java（Kafka 4.3.0）
public enum ChannelMuteState {
    NOT_MUTED,
    MUTED,
    MUTED_AND_RESPONSE_PENDING,
    MUTED_AND_THROTTLED,
    MUTED_AND_THROTTLED_AND_RESPONSE_PENDING
}

读请求时进入 MUTED_AND_RESPONSE_PENDING，表示「请求已经被 API 层拿走了，但响应还没发出去」。如果这时又触发了配额限流，会再叠加成 MUTED_AND_THROTTLED_AND_RESPONSE_PENDING。只有响应真正发回、且限流窗口结束，才回到 NOT_MUTED。

为什么要这么设计？至少三个原因：

1. 响应顺序。同一条 TCP 连接上的响应必须按请求顺序返回。如果两个请求并行处理，先完成的那个可能先写回去，客户端就会把响应配错请求。
2. 内存上界。一个连接如果允许无限个 in-flight 请求，慢请求会堆积大量待回响应，broker 的内存不可控。mute 把每条连接的在途请求压到 1 个。
3. 背压传递。处理不过来时，正确的做法是让上游慢下来，而不是继续收包再丢弃。mute 天然做到了这一点。

注意这跟客户端的 max.in.flight.requests.per.connection 不矛盾。生产者可以在一条连接上流水线式地发出多个请求，但服务端读到一个完整请求后就 mute 了，后面的请求留在 socket 缓冲区里，等响应发完才继续读。所以客户端并发度高，不代表服务端会并行处理同一条连接上的请求。

请求队列：深度看哪里，排队时间看哪里

队列本身只有一个，属于该 SocketServer 的数据平面，所有 processor 共享。它的大小就是 queued.max.requests，默认 500。可观测的有两个指标：

本机实测（Temurin 25.0.5+8 客户端 / Kafka 4.3.0 KRaft 两节点 / 2026-10-29）：先建 6 分区、RF=2 的 topic，用 kafka-producer-perf-test.sh 单生产者打 30 万条 1 KB 消息，acks=all、linger.ms=5、batch.size=16384。期间每秒采一次 RequestQueueSize 与 NetworkProcessorAvgIdlePercent，16 次采样全部是：

queue=0 netIdle=0.9992~1.0000 threads=105

单个生产者跑到约 22,114 条/秒（21.6 MB/s）。再并排起 4 个生产者，聚合约 39,000 条/秒（约 38 MB/s），14 次采样里队列深度只有一次是 1，其余为 0；同时采到的空闲率是：

t=1 queue=1 netIdle=1.0000 handlerIdle=0.9964 threads=105
t=2 queue=0 netIdle=0.9992 handlerIdle=0.9967 threads=105
...
（14 次采样中 netIdle 0.9989~1.0000，handlerIdle 0.9946~0.9967）

结论要说得保守：在这台 4 vCPU 的 colima 虚拟机上，我们没能把 500 深的请求队列压满。 网络线程空闲率 99.9% 左右，handler 空闲率 99.5% 左右，说明瓶颈不在请求处理层，而在副本同步、磁盘与虚拟机调度。队列深度采样为 0，不代表从来没有排过队——同一轮压测结束后立刻读请求维度的排队时间，更可靠：

kafka.network:type=RequestMetrics,name=RequestQueueTimeMs,request=Produce
Min=0.0 Max=29.0 Mean=0.885 50th=0.0 95th=4.0 99th=10.0 99.9th=14.97

均值 0.885 ms、99 线 10 ms、最大 29 ms。也就是说请求确实在队列里短暂停留过，只是停留时间太短，1 秒一次的瞬时采样抓不到。「队列深度为 0」和「没有排队」是两件事，这一点在写告警规则时经常被混淆。

这里也要说清楚没有测到什么：我们无法从外部稳定观测到队列被填满后的行为，因为要压满它需要远高于本机能力的吞吐。下面关于「队列满会阻塞 processor」的结论来自源码，不是实测。

队列满的时候发生什么

sendRequest 用的是 ArrayBlockingQueue.put。队列满时，调用它的 processor 线程会阻塞，直到 handler 取走一个请求。后果不是「返回错误」，而是这条 processor 负责的所有连接一起停摆：

• 它不再 poll selector，所以分配给它的连接都读不到新数据；
• 这些连接上的 Fetch、Produce、心跳全部延迟；
• 客户端的请求超时开始累积，重试进一步放大负载。

这就是为什么 queued.max.requests 不是一个可以随便调大的参数。调大只是把内存里的等待变长，并不会提高处理能力；真正的解法是找到 handler 为什么慢：磁盘、副本等待、还是锁竞争。

还有一条路径也会 mute 连接：配额限流。客户端超过 producer_byte_rate / consumer_byte_rate 之后，broker 会把响应延迟到配额窗口结束，状态进入 MUTED_AND_THROTTLED。它和队列满的表现都是「客户端变慢」，但原因完全不同：一个是被动过载，一个是主动限速。

调参的判断顺序

拿到一个「Kafka 变慢」的现场，建议按下面的顺序看，而不是先改线程数：

1. 看 BrokerRequestHandlerAvgIdlePercent。长期低于 0.3，说明 handler 真的忙，再考虑加 num.io.threads；如果接近 1，加线程没有意义。
2. 看 NetworkProcessorAvgIdlePercent。它统计的是 processor 的空闲比例。连接数特别多、单条消息特别大时，网络线程可能先成为瓶颈。
3. 看 RequestQueueTimeMs 的 99 线。它比队列深度更能说明排队是否影响尾延迟。
4. 最后才看 queued.max.requests。默认 500 对多数集群够用；把它调到几千只会让故障时的内存更高。

关于线程数的两个易错点：

• num.network.threads 是每个 listener 的数量，不是全 broker 的总数。3 个 listener × 3 = 9。每增加一个 listener（比如加 SASL_SSL 端口），网络线程就多一组。
• num.io.threads 在合并模式下会创建两个池（broker 与 controller），所以 8 会变成 16 个 handler 线程。纯 broker 角色只有一个池。

本机这组配置下，网络线程与 handler 都处于 99% 以上的空闲，说明 3/8 的默认值对这个小负载绰绰有余。把它当成「默认值太小」的反例更合适：先证明瓶颈在这里，再动参数。

和客户端参数的连接点

服务端的 mute 机制让「一条连接同时只处理一个请求」，但客户端仍然可以通过多连接提高并发。生产者对每个 broker 维护一条连接，max.in.flight.requests.per.connection 控制这条连接上允许多少个未确认请求。服务端会按顺序一个个处理，所以：

• 提高 in-flight 能填满网络管道，但不会让服务端并行处理同一条连接；
• 真正的并行度来自分区数、broker 数和多连接；
• 开启幂等后，max.in.flight 仍可大于 1，broker 按 producer id 与分区序号去重，但重试时的顺序由客户端保证。

理解了这一层，再看 3.x 时代「把 num.network.threads 调到 8」的经验贴，就知道它解决的是连接数或大消息导致的网络线程瓶颈，不是通用加速开关。

本节判据

读完这一节，你应该能回答：

1. num.network.threads 和 num.io.threads 的分界线在哪里？ 答：前者只做非阻塞收发与协议解析，后者才碰磁盘和副本；中间隔着 queued.max.requests 深的有界队列。
2. 为什么同一条连接不能并行处理两个请求？ 答：processor 在请求入队后立刻 mute 连接，响应发完才 unmute；这保证响应顺序、限制在途内存、并把过载背压回客户端。
3. 队列深度为 0 是否意味着没有排队？ 答：不是。瞬时采样会错过尖峰，要看 RequestQueueTimeMs 的尾部分位。

下一节把视角移到副本：follower 是怎么把数据拉过去的，acks=1 与 acks=all 在同一负载下差多少，以及 ISR 收缩之后吞吐为什么会「变快」——那不是优化，是少等了一个人。

阅读导航：上一节 10.3 磁盘与文件描述符规划 ｜ 下一节 11.2 副本同步的吞吐与延迟