大家好，我是 华仔, 又跟大家见面了。
原文完整版在星球里面，如果感兴趣可以扫文末二维码加入。
上篇主要带大家深度剖析了「号称承载 Kafka 客户端消息快递仓库 RecordAccmulator 的架构设计」，消息被暂存到累加器中，今天主要聊聊「发送网络 I/O 的 Sender 线程的架构设计」，深度剖析下消息是如何被发送出去的。
这篇文章干货很多，希望你可以耐心读完。
通过「场景驱动」的方式，来看看消息是如何从客户端发送出去的。
在上篇中，我们知道了消息被暂存到 Deque<ProducerBatch> 的 batches 中，等「批次已满」或者「有新批次被创建」后，唤醒 Sender 子线程将消息批量发送给 Kafka Broker 端。
接下来我们就来看看，「Sender 线程的架构实现以及发送处理流程」，为了方便大家理解，所有的源码只保留骨干。
在《图解Kafka生产者初始化核心流程》这篇中我们知道 KafkaProducer 会启动一个后台守护进程，其线程名称：kafka-producer-network-thread + "|" + clientId。
在 KafkaProducer.java 类有常量定义：NETWORK_THREAD_PREFIX，并启动 守护线程 KafkaThread 即 ioThread，如果不主动关闭 Sender 线程会一直执行下去。
github 源码地址如下：
https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/RequestCompletionHandler.java
https://github.com/apache/kafka/blob/2.7/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
复制
public class KafkaProducer<K, V> implements Producer<K, V> {public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread";// visible for testing@SuppressWarnings("unchecked")KafkaProducer(Map<String, Object> configs,Serializer<K> keySerializer, Serializer<V> valueSerializer,ProducerMetadata metadata, KafkaClient kafkaClient,ProducerInterceptors<K, V> interceptors,Time time) {try {
...this.sender = newSender(logContext, kafkaClient, this.metadata);String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;this.ioThread = new KafkaThread(ioThreadName, this.sender, true);this.ioThread.start();
...log.debug("Kafka producer started");
} catch (Throwable t) {
...
}
}
}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
从上面得出 Sender 类是一个线程类， 我们来看看 Sender 线程的重要字段和方法，并讲解其是如何发送消息和处理消息响应的。
复制
/**
* The background thread that handles the sending of produce requests to the Kafka cluster. This thread makes metadata
* requests to renew its view of the cluster and then sends produce requests to the appropriate nodes.
*/public class Sender implements Runnable {/* the state of each nodes connection */private final KafkaClient client; // 为 Sender 线程提供管理网络连接进行网络读写/* the record accumulator that batches records */private final RecordAccumulator accumulator; // 消息仓库累加器/* the metadata for the client */private final ProducerMetadata metadata; // 生产者元数据/* the flag indicating whether the producer should guarantee the message order on the broker or not. */private final boolean guaranteeMessageOrder; // 是否保证消息在 broker 端的顺序性 /* the maximum request size to attempt to send to the server */private final int maxRequestSize; //发送消息最大字节数。 /* the number of acknowledgements to request from the server */private final short acks; // 生产者的消息发送确认机制/* the number of times to retry a failed request before giving up */private final int retries; // 发送失败后的重试次数，默认为0次/* true while the sender thread is still running */private volatile boolean running; // Sender 线程是否还在运行中 /* true when the caller wants to ignore all unsent/inflight messages and force close. */private volatile boolean forceClose; // 是否强制关闭，此时会忽略正在发送中的消息。/* the max time to wait for the server to respond to the request*/private final int requestTimeoutMs; // 等待服务端响应的最大时间,默认30s /* The max time to wait before retrying a request which has failed */private final long retryBackoffMs; // 失败重试退避时间 /* current request API versions supported by the known brokers */private final ApiVersions apiVersions; // 所有 node 支持的 api 版本/* all the state related to transactions, in particular the producer id, producer epoch, and sequence numbers */private final TransactionManager transactionManager; // 事务管理，这里忽略 后续会有专门一篇讲解事务相关的// A per-partition queue of batches ordered by creation time for tracking the in-flight batchesprivate final Map<TopicPartition, List<ProducerBatch>> inFlightBatches; // 正在执行发送相关的消息批次集合， key为分区，value是 list<ProducerBatch> 。
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
从该类属性字段来看比较多，这里说几个关键字段：
client：KafkaClient 类型，是一个接口类，Sender 线程主要用它来实现真正的网络I/O，即 NetworkClient。该字段主要为 Sender 线程提供了网络连接管理和网络读写操作能力。
accumulator：RecordAccumulator类型，上一篇的内容 图解 Kafka 源码之快递仓库 RecordAccumulator 架构设计，Sender 线程用它获取待发送的 node 节点及批次消息等能力。
metadata：ProducerMetadata类型，生产者元数据。因为发送消息时要知道分区 Leader 在哪些节点，以及节点的地址、主题分区的情况等。
guaranteeMessageOrder：是否保证消息在 broker 端的顺序性，参数：max.in.flight.requests.per.connection。
maxRequestSize：单个请求发送消息最大字节数，默认为1M，它限制了生产者在单个请求发送的记录数，以避免发送大量请求。
acks：生产者的消息发送确认机制。有3个可选值：0，1，-1/all。
retries：生产者发送失败后的重试次数。默认是0次。
running：Sender线程是否还在运行中。
forceClose：是否强制关闭，此时会忽略正在发送中的消息。
requestTimeoutMs：生产者发送请求后等待服务端响应的最大时间。如果超时了且配置了重试次数，会再次发送请求，待重试次数用完后在这个时间范围内返回响应则认为请求最终失败，默认 30 秒。
retryBackoffMs：生产者在发送请求失败后可能会重新发送失败的请求，其目的就是防止重发过快造成服务端压力过大。默认100 ms。
apiVersions：ApicVersions类对象，保存了每个node所支持的api版本。
inFlightBatches：正在执行发送相关的消息批次集合， key为分区，value是 list<|ProducerBatch>。
Sender 类的方法也不少，这里针对关键方法逐一讲解下。
Sender 线程实现了 Runnable 接口，会不断的调用 runOnce()，这是一个典型的循环事件机制。
复制
/**
* The main run loop for the sender thread
*/@Overridepublic void run() {log.debug("Starting Kafka producer I/O thread.");// 这里使用 volatile boolean 类型的变量 running，判断 Sender 线程是否在运行状态中。// 1. 如果 Sender 线程在运行状态即 running=true，则一直循环调用 runOnce() 方法。while (running) {try {// 将缓冲区的消息发送到 broker。runOnce();
} catch (Exception e) {log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");// 2. 如果（没有强制关闭 && ((消息累加器中还有剩余消息待发送 || 还有等待未响应的消息 ) || 还有事务请求未完成))，则继续发送剩下的消息。while (!forceClose && ((this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0) || hasPendingTransactionalRequests())) {try {// 继续执行将剩余的消息发送完毕runOnce();
} catch (Exception e) {log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}// 3. 对进行中的事务进行中断，则继续发送剩下的消息。while (!forceClose && transactionManager != null && transactionManager.hasOngoingTransaction()) {if (!transactionManager.isCompleting()) {log.info("Aborting incomplete transaction due to shutdown");transactionManager.beginAbort();
}try {runOnce();
} catch (Exception e) {log.error("Uncaught error in kafka producer I/O thread: ", e);
}
}// 4. 如果强制关闭，则关闭事务管理器、终止消息的追加并清空未完成的批次if (forceClose) {if (transactionManager != null) {log.debug("Aborting incomplete transactional requests due to forced shutdown");// 关闭事务管理器transactionManager.close();
}log.debug("Aborting incomplete batches due to forced shutdown");// 终止消息的追加并清空未完成的批次this.accumulator.abortIncompleteBatches();
}// 5. 关闭 NetworkClienttry {this.client.close();
} catch (Exception e) {log.error("Failed to close network client", e);
}log.debug("Shutdown of Kafka producer I/O thread has completed.");
}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
当 Sender 线程启动后会直接运行 run() 方法，该方法在 4种情况下会一直循环调用去发送消息到 Broker。
我们来看看这个执行业务处理的方法，关于事务的部分后续专门文章讲解。
复制
/**
* Run a single iteration of sending
*/void runOnce() {if (transactionManager != null) {
... //事务处理方法 后续文章专门讲解 }// 1. 获取当前时间的时间戳。long currentTimeMs = time.milliseconds();// 2. 调用 sendProducerData 发送消息,但并非真正的发送，而是把消息缓存在 把消息缓存在 KafkaChannel 的 Send 字段里。下一篇会讲解 NetworkClient。long pollTimeout = sendProducerData(currentTimeMs);// 3. 读取消息实现真正的网络发送client.poll(pollTimeout, currentTimeMs);
}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
该方法比较简单，主要做了3件事情：
获取当前时间的时间戳。
调用 sendProducerData 发送消息,但并非真正的发送，而是把消息缓存在 NetworkClient 的 Send 字段里。下一篇会讲解 NetworkClient。
读取 NetworkClient 的 send 字段消息实现真正的网络发送。
该方法主要是获取已经准备好的节点上的批次数据并过滤过期的批次集合，最后暂存消息。
复制
private long sendProducerData(long now) {// 1. 获取元数据Cluster cluster = metadata.fetch();// get the list of partitions with data ready to send// 2. 获取已经准备好的节点以及找不到 Leader 分区对应的节点的主题RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);// if there are any partitions whose leaders are not known yet, force metadata update// 3. 如果主题 Leader 分区对应的节点不存在，则强制更新元数据if (!result.unknownLeaderTopics.isEmpty()) {// 添加 topic 到没有拉取到元数据的 topic 集合中，并标识需要更新元数据for (String topic : result.unknownLeaderTopics)this.metadata.add(topic, now);
...// 针对这个 topic 集合进行元数据更新this.metadata.requestUpdate();
}// 4. 循环 readyNodes 并检查客户端与要发送节点的网络是否已经建立好了Iterator<Node> iter = result.readyNodes.iterator();long notReadyTimeout = Long.MAX_VALUE;while (iter.hasNext()) {Node node = iter.next();// 检查客户端与要发送节点的网络是否已经建立好了if (!this.client.ready(node, now)) {// 移除对应节点iter.remove();notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
}
}// create produce requests// 5. 获取上面返回的已经准备好的节点上要发送的 ProducerBatch 集合Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);// 6. 将从消息累加器中读取的数据集，放入正在执行发送相关的消息批次集合中addToInflightBatches(batches);// 7. 要保证消息的顺序性，即参数 max.in.flight.requests.per.cnotallow=1if (guaranteeMessageOrder) {// Mute all the partitions drainedfor (List<ProducerBatch> batchList : batches.values()) {for (ProducerBatch batch : batchList)// 对 tp 进行 mute，保证只有一个 batch 正在发送this.accumulator.mutePartition(batch.topicPartition);
}
}// 重置下一批次的过期时间accumulator.resetNextBatchExpiryTime();// 8. 从正在执行发送数据集合 inflightBatches 中获取过期集合List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);// 9. 从 batches 中获取过期集合List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);// 10. 从 inflightBatches 与 batches 中查找已过期的消息批次(ProducerBatch)，判断批次是否过 期是根据系统当前时间与 ProducerBatch 创建时间之差是否超过120s，过期时间可以通过参数 delivery.timeout.ms 设置。expiredBatches.addAll(expiredInflightBatches);// 如果过期批次不为空 则输出对应日志if (!expiredBatches.isEmpty())log.trace("Expired {} batches in accumulator", expiredBatches.size());// 11. 处理已过期的消息批次，通知该批消息发送失败并返回给客户端for (ProducerBatch expiredBatch : expiredBatches) {// 处理当前过期ProducerBatch的回调结果 ProduceRequestResult,并且设置超时异常 new TimeoutException(errorMessage)String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition+ ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";// 通知该批消息发送失败并返回给客户端failBatch(expiredBatch, -1, NO_TIMESTAMP, new TimeoutException(errorMessage), false);// ... 事务管理器的处理忽略}// 收集统计指标，后续会专门对 Kafka 的 Metrics 进行分析sensors.updateProduceRequestMetrics(batches);// 设置下一次的发送延时long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now);pollTimeout = Math.max(pollTimeout, 0);if (!result.readyNodes.isEmpty()) {log.trace("Nodes with data ready to send: {}", result.readyNodes);pollTimeout = 0;
}// 12. 发送消息暂存到 NetworkClient send 字段里。sendProduceRequests(batches, now);return pollTimeout;
}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
该方法主要做了12件事情，逐一说明下：
首先获取元数据，这里主要是根据元数据的更新机制来保证数据的准确性。
获取已经准备好的节点。accumulator#reay() 方法会通过发送记录对应的节点和元数据进行比较，返回结果中包括两个重要的集合:「准备好发送的节点集合 readyNodes」、「找不到 Leader 分区对应节点的主题 unKnownLeaderTopic」。
如果主题 Leader 分区对应的节点不存在，则强制更新元数据。
循环 readyNodes 并检查客户端与要发送节点的网络是否已经建立好了。在 NetworkClient 中维护了客户端与所有节点的连接，这样就可以通过连接的状态判断是否连接正常。
获取上面返回的已经准备好的节点上要发送的 ProducerBatch 集合。accumulator#drain() 方法就是将 「TopicPartition」-> 「ProducerBatch 集合」的映射关系转换成 「Node 节点」->「ProducerBatch 集合」的映射关系，如下图所示，这样的话按照节点方式只需要2次就完成，大大减少网络的开销。
将从消息累加器中读取的数据集，放入正在执行发送相关的消息批次集合中。
要保证消息的顺序性，即参数 max.in.flight.requests.per.cnotallow=1，会添加到 muted 集合，保证只有一个 batch 在发送。
从正在执行发送数据集合 inflightBatches 中获取过期集合。
从 accumulator 累加器的 batches 中获取过期集合。
从 inflightBatches 与 batches 中查找已过期的消息批次(ProducerBatch)，判断批次是否过期是根据系统当前时间与 ProducerBatch 创建时间之差是否超过120s，过期时间可以通过参数 delivery.timeout.ms 设置。
处理已过期的消息批次，通知该批消息发送失败并返回给客户端。
发送消息暂存到 NetworkClient send 字段里。
从上面源码可以看出，SendProducerData 方法中调用到了 Sender 线程类中多个方法，这里就不一一讲解了。
通过前两部分的源码解读和剖析，我们可以得出 Sender 线程的处理流程可以分为两大部分：「发送请求」、「接收响应结果」。
从 runOnce 方法可以得出发送请求也分两步：「消息预发送」、「真正的网络发送」。
复制
void runOnce() {
// 1. 把消息缓存在 KafkaChannel 的 Send 字段里。
long pollTimeout = sendProducerData(currentTimeMs);
// 2. 读取消息实现真正的网络发送
client.poll(pollTimeout, currentTimeMs);
}
1.
2.
3.
4.
5.
6.
等 Sender 线程收到 Broker 端的响应结果后，会根据响应结果分情况进行处理。
原文完整版在星球里面，如果感兴趣可以扫文末二维码加入。
这里，我们一起来总结一下这篇文章的重点。
1、开篇总述消息被暂存到 Deque<ProducerBatch> 的 batches 中，等「批次已满」或者「有新批次被创建」后，唤醒 Sender 子线程将消息批量发送给 Kafka Broker 端，从而引出「Sender 线程」。
2、带你深度剖析了「Sender 线程」 的发送消息以及响应处理的相关方法。
3、最后带你串联了整个消息发送的流程，让你有个更好的整体认知。