Kafka由浅入深6 Sender线程执行源码

一、KafkaProducer消息流程图

1.1 KafkaProducer 消息架构图

1.2 KafkaProducer 消息架构分为两部分

第一部分是KafkaProducer主线程

主要逻辑提供消息拦截器、序列化器、和分区器的默认实现和对外自定义扩展功能，已经将消息追加并缓存到累加器RecordAccumulator中，为Sender线程提供批量数据队列和发送分区的准备工作。

备注：

1、对于主线程的原理分析，可以看之前的文章

2、详细解说RecordAccumulator，可以看这边文章

第二部分是Sender线程

1.3 Sender线程执行流程图

二、Sender线程源码分析

Sender 是一个 Kafka 生产者消息执行的网络线程，无限循环运行在后台将 ProduceRequests 发送到 Kafka 集群中。

2.1 Sender线程的定义和创建

2.1.1 Sender线程定义

Sender继承了Runnable接口，而Sender线程是在KafkaProducer 的构造器中进行实例化和线程启动。当KafkaProducer被创建时，也会在构造器中实例化以 “kafka-producer-network-thread |[clientId]”命名的线程，而这个线程作为守护线程，伴随着KafkaProducer整个生命周期。

从下面的源码可以看出，Sender线程的run()方法中，核心发送入口方法为runOnce()。每一次执行runOnce()方法，Sender 将从RecordAccumulator记录累加器中获取1~max.request.size 个 ProducerBatch的消息数据，并最终将 ProduceRequests 发送到对应活动的 Kafka broker服务器中。

1.  
   public class Sender implements Runnable {
2.  
   /**
3.  
   * The main run loop for the sender thread
4.  
   */
5.  
   @Override
6.  
   public void run() {
7.  
   log.debug("Starting Kafka producer I/O thread.");
8.  
    
9.  
   // main loop, runs until close is called
10.  
    while (running) {
11.  
    try {
12.  
    runOnce();
13.  
    } catch (Exception e) {
14.  
    log.error("Uncaught error in kafka producer I/O thread: ", e);
15.  
    }
16.  
    }
17.  
     
18.  
    log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");
19.  
     
20.  
    ....省略....
21.  
    }
22.  
    }

2.1.2 KafkaProducer中实例化Sender线程

1.  
   public class KafkaProducer<K, V> implements Producer<K, V> {
2.  
   ...省略
3.  
    
4.  
   // KafkaProducer构造器中，对Sender线程进行创建和线程启动
5.  
   KafkaProducer(ProducerConfig config,
6.  
   Serializer<K> keySerializer,
7.  
   Serializer<V> valueSerializer,
8.  
   ProducerMetadata metadata,
9.  
   KafkaClient kafkaClient,
10.  
    ProducerInterceptors<K, V> interceptors,
11.  
    Time time) {
12.  
    ...省略
13.  
    // Sender线程的创建和启动
14.  
    this.sender = newSender(logContext, kafkaClient, this.metadata);
15.  
    String ioThreadName = NETWORK_THREAD_PREFIX " | " clientId;
16.  
    this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
17.  
    // Sender线程启动
18.  
    this.ioThread.start();
19.  
    ...省略
20.  
    }
21.  
     
22.  
    // Sender线程的创建
23.  
    Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
24.  
    // 最大InFlightConnection请求链接数
25.  
    int maxInflightRequests = producerConfig.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
26.  
    // 请求超时时间
27.  
    int requestTimeoutMs = producerConfig.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG);
28.  
    // 创建ChannelBuilder,提供服务Channel
29.  
    ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(producerConfig, time, logContext);
30.  
    ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
31.  
    // kafka数据传感器
32.  
    Sensor throttleTimeSensor = Sender.throttleTimeSensor(metricsRegistry.senderMetrics);
33.  
    // 创建KafkaClient客户端，默认为NetworkClient
34.  
    KafkaClient client = kafkaClient != null ? kafkaClient : new NetworkClient(
35.  
    new Selector(producerConfig.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
36.  
    this.metrics, time, "producer", channelBuilder, logContext),
37.  
    metadata,
38.  
    clientId,
39.  
    maxInflightRequests,
40.  
    producerConfig.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),
41.  
    producerConfig.getLong(ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG),
42.  
    producerConfig.getInt(ProducerConfig.SEND_BUFFER_CONFIG),
43.  
    producerConfig.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),
44.  
    requestTimeoutMs,
45.  
    producerConfig.getLong(ProducerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG),
46.  
    producerConfig.getLong(ProducerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG),
47.  
    time,
48.  
    true,
49.  
    apiVersions,
50.  
    throttleTimeSensor,
51.  
    logContext);
52.  
     
53.  
    short acks = Short.parseShort(producerConfig.getString(ProducerConfig.ACKS_CONFIG));
54.  
    // 创建Sender线程
55.  
    return new Sender(logContext,
56.  
    client,
57.  
    metadata,
58.  
    this.accumulator,
59.  
    maxInflightRequests == 1,
60.  
    producerConfig.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
61.  
    acks,
62.  
    producerConfig.getInt(ProducerConfig.RETRIES_CONFIG),
63.  
    metricsRegistry.senderMetrics,
64.  
    time,
65.  
    requestTimeoutMs,
66.  
    producerConfig.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
67.  
    this.transactionManager,
68.  
    apiVersions);
69.  
    }
70.  
     
71.  
    }

2.1.3 Sender实例化参数说明

参数和实现功能说明

InFlightRequest: 飞行途中的请求，Sender线程发送到Kafka之前会保存数据到InFlightRequest中，主要作用是缓存已经发送但未接收到响应的请求，保存的数据模型为Map<NodeId,Deque<Request>>。

KafkaProducer 在创建Sender时会设置如下属性：

1、max.in.flight.requests.per.connection(maxInflightRequests)： 设置每一个InFlightRequest链接最多缓存多少个未收到响应的请求，默认值为5。如果缓存的数据到达了最大值，则消息发送处于阻塞状态。如果需要保证消息顺序发送，可以将配置设置为1，但会影响到消息发送的执行效率，降低消息发送的吞吐量

2、request.timeout.ms(requestTimeoutMs):

KafkaProducer 等待请求响应的最长时间，默认值为 30000 ms。 请求超时之后可以选择进行重试。 注意这个参数需要比 broker 端参数 replica.lag.time.max.ms 的值要大，这样可以减少因客户端重试而引起的消息重复的概率。

3、max.request.size

这个参数用来限制生产者客户端能够发送的消息的最大值，默认值为 1048576 B，即 1 MB。 一般情况下，这个默认值就可以满足大多数的应用场景了。 不建议盲目地增大这个参数的配置值，尤其是在对 Kafka 整体脉络没有足够把控的时候。 因为这个参数还涉及一些其他参数的联动，比如 broker 端的 message.max.bytes 参数，如果配置错误可能会引起一些不必要的一场。 比如讲 broker 端端 message.max.bytes 参数配置为 10， 而 max.request.size 参数配置为 20，那么当我们发送一条消息大小为 15 的消息时，生产者客户端就会报出异常： org.apache.kafka.common.errors.RecordTooLargeException: The reqeust included a message larger than the max message size the server will accept.

4、retries 和 retry.backoff.ms retries 参数用来配置生产者重试的次数，默认值为 0，即发生异常的时候不进行任何的重试动作。 消息在从生产者发出道成功写入服务器之前可能发生一些临时性的异常，比如网络抖动、Leader 副本的选举等，这种异常往往是可以自行恢复的，生产者可以通过配置 retries 大于 0 的值，以此通过内部重试来恢复而不是一味的将异常抛给生产者的应用程序。 如果重试达到设定的次数，那么生产者就会放弃重试并返回异常。 不过并不是所有的异常都是可以通过重试来解决的，比如消息太大，超过 max.request.size 参数配置的值时，这种方式就不行了。 重试还和另一个参数 retry.backoff.ms 有关，这个参数的默认值为 100，它用来设定两次重试之间的时间间隔，避免无效的频繁重试。 在配置 retries 和 retry.backoff.ms 之前，最好先估算一下可能的异常恢复时间，这样可以设定总的重试时间大于这个异常恢复时间，以此来避免生产者过早地放弃重试。 Kafka 可以保证同一个分区中的消息时有序的。 如果生产者按照一定的顺序发送消息，那么这些消息也会顺序的写入分区，进而消费者也可以按照同样的顺序消费它们。

5、acks ： 消息应答机制，默认值即为 1

ack =1 —— 生产者发送消息后，只要分区的leader副本写入消息成功，则生产者客户端就会收到服务器的成功响应，视为消息发送成功。

ack 为1 的时候，保证客消息发送的吞吐量，但是在消息可靠性上不能完全保证。会出现如下的情况：

        a、如果消息无法写入leader副本，则生产者会收到错误响应，无法写入的情况通常情况为：leader副本崩溃、leader副本的重新选举期间。所以为了避免消息发送丢失的情况，生产者可以进行消息发送的失败重试。

        b、在极端情况下可能会出现，消息写入成功写入leader副本，并且生产者客户端收到了服务器的成功响应。而此时leader副本崩溃，但是其他follower副本未及时更新到leader副本的数据，在新选举出的leader副本中则服务查询到这一条数据的记录，也就出现了消息丢失的情况。

acks = 0—— 生成者发送消息后，不用等待任何服务端的成功与否的响应。如果消息写入到Kafak服务端出现异常，从而可能导致消息丢失的情况。设置为0 可以达到消息发送的最大吞吐量，但是也可能出现消息丢失的情况。

acks = -1 或 acks = all ——生产者发送后，需要等待 leader副本将消息同步到所有follower副本，则返回消息成功，从而保证消息的可靠性。

2.2 Sender的线程的唤起执行wakeup

首先，我们需要知道Sender线程的执行是在哪里阻塞的。Kafka客户端的底层网络通信是通过NIO实现，Kafka 的Selecotor类中，我们可以看到java.nio.channels.Selector nioSelector，对Kafak服务器的通信核心是通过nioSelector 来实现。

当调用nioSelector.select()方法时，调用者线程会进入阻塞状态，直到有就绪的Channel才会返回。而java.nio.channels.Selector 也提供了从阻塞状态唤起的方式，则是可以通过其他的线程调用nioSelector.wakeup进行阻塞线程的唤醒，则select()方法也会立即返回。

1.  
   public class Selector implements Selectable, AutoCloseable {
2.  
    
3.  
   private final java.nio.channels.Selector nioSelector;
4.  
    
5.  
   /**
6.  
   * Interrupt the nioSelector if it is blocked waiting to do I/O.
7.  
   */
8.  
   @Override
9.  
   public void wakeup() {
10.  
    this.nioSelector.wakeup();
11.  
    }
12.  
     
13.  
    }

wake()被调用的地方：

KafkaProducer 的doSend(),initTransactions(), sendOffsetsToTransaction(), commitTransaction(), abortTransaction(), waitOnMetadata()和 flush()方法

Sender类的 initiateClose()方法

三、Sender线程核心流程源码分析

3.1 Sender线程run()方法逻辑

Sender线程的状态通过两个变量控制，分别是running和forceClose

running: Sender线程是否正在运行中

forceClose: 是否强制关闭正在发送和待发送的消息，

强制关闭状态，忽略未发送和正在发送中的消息

分三种情况

1、状态一 running状态:

sender线程处于running状态，则一直循环执行runOnce()

2、状态二 非强制关闭状态：

     a、停止接受新的请求，如果事务管理器、累加器或等待确认的过程中可能仍有请求，请等到这些完成。

     b、如果任何提交或中止未通过事务管理器的队列，则中止事务

3、强制关闭状态

     a、将所有未完成的事务请求和batchs置为失败，并唤醒正在等待的线程
     b、我们需要使所有不完整的事务请求和批处理失败，并唤醒等待未来的线程。

 

Sender.run()源码分析

1.  
   // sender线程运行状态
2.  
   private volatile boolean running;
3.  
    
4.  
   // 强制关闭状态，忽略未发送和正在发送中的消息
5.  
   private volatile boolean forceClose;
6.  
    
7.  
   @Override
8.  
   public void run() {
9.  
   log.debug("Starting Kafka producer I/O thread.");
10.  
     
11.  
    // 状态一 running状态: sender线程处于running状态，则一直循环执行runOnce()
12.  
    while (running) {
13.  
    try {
14.  
    runOnce();
15.  
    } catch (Exception e) {
16.  
    log.error("Uncaught error in kafka producer I/O thread: ", e);
17.  
    }
18.  
    }
19.  
     
20.  
    log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");
21.  
     
22.  
    // 状态二 非强制关闭状态
23.  
    //停止接受新的请求，如果事务管理器、累加器或等待确认的过程中可能仍有请求，请等到这些完成。
24.  
    while (!forceClose && ((this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0) || hasPendingTransactionalRequests())) {
25.  
    try {
26.  
    runOnce();
27.  
    } catch (Exception e) {
28.  
    log.error("Uncaught error in kafka producer I/O thread: ", e);
29.  
    }
30.  
    }
31.  
     
32.  
     
33.  
    //如果任何提交或中止未通过事务管理器的队列，则中止事务
34.  
    while (!forceClose && transactionManager != null && transactionManager.hasOngoingTransaction()) {
35.  
    if (!transactionManager.isCompleting()) {
36.  
    log.info("Aborting incomplete transaction due to shutdown");
37.  
    transactionManager.beginAbort();
38.  
    }
39.  
    try {
40.  
    runOnce();
41.  
    } catch (Exception e) {
42.  
    log.error("Uncaught error in kafka producer I/O thread: ", e);
43.  
    }
44.  
    }
45.  
     
46.  
    // 状态三 强制关闭状态
47.  
    if (forceClose) {
48.  
    // 将所有未完成的事务请求和batchs置为失败，并唤醒正在等待的线程
49.  
    //我们需要使所有不完整的事务请求和批处理失败，并唤醒等待未来的线程。
50.  
    if (transactionManager != null) {
51.  
    log.debug("Aborting incomplete transactional requests due to forced shutdown");
52.  
    transactionManager.close();
53.  
    }
54.  
    log.debug("Aborting incomplete batches due to forced shutdown");
55.  
    this.accumulator.abortIncompleteBatches();
56.  
    }
57.  
    try {
58.  
    // KafkaClient Kafka客户端关闭
59.  
    this.client.close();
60.  
    } catch (Exception e) {
61.  
    log.error("Failed to close network client", e);
62.  
    }
63.  
     
64.  
    log.debug("Shutdown of Kafka producer I/O thread has completed.");
65.  
    }

3.2 Sender线程runOnce()方法逻辑

从run()方法的逻辑可以看出，runOnce()方法是消息发送的入口

a、当事务存在的时候,进行事务处理逻辑，详细的事务处理，将在后续技术文章中进行讲解

b、 调用sendProducerData()方法，则是Kafka消息发送的核心处理逻辑

1.  
   void runOnce() {
2.  
    
3.  
   if (transactionManager != null) {
4.  
   try {
5.  
   // 事务的发送处理逻辑
6.  
   //尝试解析未解析的序列。如果所有正在执行的请求都已完成，并且某些分区仍无法解析，则如果可能，请中断epoch，或者转换为致命错误
7.  
   transactionManager.maybeResolveSequences();
8.  
    
9.  
   // 如果事务管理器处于失败状态，则不要继续发送
10.  
    if (transactionManager.hasFatalError()) {
11.  
    RuntimeException lastError = transactionManager.lastError();
12.  
    if (lastError != null)
13.  
    maybeAbortBatches(lastError);
14.  
    client.poll(retryBackoffMs, time.milliseconds());
15.  
    return;
16.  
    }
17.  
     
18.  
    //检查我们是否需要新的producerId。如果是这样，我们将在下面发送InitProducerId请求
19.  
    transactionManager.bumpIdempotentEpochAndResetIdIfNeeded();
20.  
    //如果发送或轮询事务请求，或者FindCoordinator请求已排队，则返回true
21.  
    if (maybeSendAndPollTransactionalRequest()) {
22.  
    return;
23.  
    }
24.  
    } catch (AuthenticationException e) {
25.  
    // This is already logged as error, but propagated here to perform any clean ups.
26.  
    log.trace("Authentication exception while processing transactional request", e);
27.  
    transactionManager.authenticationFailed(e);
28.  
    }
29.  
    }
30.  
     
31.  
    long currentTimeMs = time.milliseconds();
32.  
     
33.  
    // 发送ProducerData核心逻辑
34.  
    long pollTimeout = sendProducerData(currentTimeMs);
35.  
    client.poll(pollTimeout, currentTimeMs);
36.  
    }

3.3 Sender线程sendProducerData()的核心逻辑梳理

虚线框内的逻辑，正式sendProducerData()的实现逻辑

1.  
   private long sendProducerData(long now) {
2.  
   // 获取当前运行并且未阻塞的Kafka集群元数据信息
3.  
   Cluster cluster = metadata.fetch();
4.  
   // get the list of partitions with data ready to send
5.  
   // 获取准备发送数据的分区（node节点）集合
6.  
   RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
7.  
    
8.  
   // if there are any partitions whose leaders are not known yet, force metadata update
9.  
   // 如果存在分区的leader未知，则强制更新元数据
10.  
    if (!result.unknownLeaderTopics.isEmpty()) {
11.  
    //未知leader的主题集包含leader选举待定的topic以及可能已过期的topic。
12.  
    // 再次将topic添加到ProducerMetadata元数据中，以确保包含该topic并请求元数据更新，因为这些消息要需要发送到对应主题。
13.  
    for (String topic : result.unknownLeaderTopics)
14.  
    this.metadata.add(topic, now);
15.  
     
16.  
    log.debug("Requesting metadata update due to unknown leader topics from the batched records: {}",
17.  
    result.unknownLeaderTopics);
18.  
    this.metadata.requestUpdate();
19.  
    }
20.  
     
21.  
    // 移除还没有准备好的node节点
22.  
    Iterator<Node> iter = result.readyNodes.iterator();
23.  
    long notReadyTimeout = Long.MAX_VALUE;
24.  
    while (iter.hasNext()) {
25.  
    Node node = iter.next();
26.  
    if (!this.client.ready(node, now)) {
27.  
    //仅更新延迟统计的readyTimeMs，以便每次批处理就绪时都向前移动（那么readyTimeM和drainTimeMs之间的差异将表示数据等待节点的时间）。
28.  
    this.accumulator.updateNodeLatencyStats(node.id(), now, false);
29.  
    iter.remove();
30.  
    notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
31.  
    } else {
32.  
     
33.  
    //更新readyTimeMs和drainTimeMs，这将“重置”节点延迟。
34.  
    this.accumulator.updateNodeLatencyStats(node.id(), now, true);
35.  
    }
36.  
    }
37.  
     
38.  
     
39.  
    // 创建produce请求，加入到Map<TopicPartition, List<ProducerBatch>> inFlightBatches中
40.  
    Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);
41.  
    addToInflightBatches(batches);
42.  
     
43.  
    if (guaranteeMessageOrder) {
44.  
    // Mute all the partitions drained
45.  
    // 将所有排水分区静音
46.  
    for (List<ProducerBatch> batchList : batches.values()) {
47.  
    for (ProducerBatch batch : batchList)
48.  
    this.accumulator.mutePartition(batch.topicPartition);
49.  
    }
50.  
    }
51.  
     
52.  
    // 重置消息累加器中下一批次的过期时间
53.  
    accumulator.resetNextBatchExpiryTime();
54.  
    // 获取已达到超时待发送的in-flight batches 数据
55.  
    List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
56.  
    // 获取累积器中放置时间过长且需要过期的批次列表。
57.  
    List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);
58.  
    expiredBatches.addAll(expiredInflightBatches);
59.  
     
60.  
     
61.  
    //如果之前已将过期批次发送给代理，请重置生产者id。同时更新过期批次的指标。请参阅@TransactionState.resetIdempotentProducerId的文档，以了解为什么需要在此处重置生产者id。
62.  
    if (!expiredBatches.isEmpty())
63.  
    log.trace("Expired {} batches in accumulator", expiredBatches.size());
64.  
    for (ProducerBatch expiredBatch : expiredBatches) {
65.  
    String errorMessage = "Expiring " expiredBatch.recordCount " record(s) for " expiredBatch.topicPartition
66.  
    ":" (now - expiredBatch.createdMs) " ms has passed since batch creation";
67.  
    failBatch(expiredBatch, new TimeoutException(errorMessage), false);
68.  
    if (transactionManager != null && expiredBatch.inRetry()) {
69.  
    // This ensures that no new batches are drained until the current in flight batches are fully resolved.
70.  
    //这可确保在当前in-flight batches 完全解决之前，不会排出新批次。
71.  
    transactionManager.markSequenceUnresolved(expiredBatch);
72.  
    }
73.  
    }
74.  
    sensors.updateProduceRequestMetrics(batches);
75.  
     
76.  
     
77.  
    //如果我们有任何节点准备发送 具有可发送数据，则轮询0超时，这样可以立即循环并尝试发送更多数据。
78.  
    // 否则，超时将是下一批到期时间与检查数据可用性的延迟时间之间的较小值。
79.  
    // 请注意，由于延迟、后退等原因，节点可能有尚未发送的数据。这特别不包括具有尚未准备好发送的可发送数据的节点，因为它们会导致繁忙的循环。
80.  
    long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
81.  
    pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now);
82.  
    pollTimeout = Math.max(pollTimeout, 0);
83.  
    if (!result.readyNodes.isEmpty()) {
84.  
    log.trace("Nodes with data ready to send: {}", result.readyNodes);
85.  
     
86.  
    //如果某些分区已经准备好发送，则选择时间将为0；
87.  
    // 否则，如果某个分区已经积累了一些数据，但还没有准备好，则选择时间将是现在与其延迟到期时间之间的时间差；
88.  
    // 否则，选择时间将是现在与元数据到期时间之间的时间差；
89.  
     
90.  
    pollTimeout = 0;
91.  
    }
92.  
    // 消息请求发送
93.  
    sendProduceRequests(batches, now);
94.  
    return pollTimeout;
95.  
    }

3.3.1 获取当前运行并且未阻塞的Kafka集群元数据信息

Cluster cluster = metadata.fetch();

3.3.2 从RecordAccumulator 消息累加器中，获取到准备发送数据的分区集合

RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

3.3.3 如果准备发送的分区集合中，有分区的leader出现未知情况。

未知leader的主题集：包含leader选举待定的topic以及可能已过期的topic。 再次将topic添加到ProducerMetadata元数据中，以确保包含该topic并请求元数据更新，因为这些消息要需要发送到对应主题。

通过调用ProducerMetadata 元数据的requestUpdate()方法打上needFullUpdate标识，标识Kafka集群需要进行强制更新。

1.  
   // 如果存在分区的leader未知，则强制更新元数据
2.  
   if (!result.unknownLeaderTopics.isEmpty()) {
3.  
    
4.  
   // 再次将topic添加到ProducerMetadata元数据中，以确保包含该topic并请求元数据更新，因为这些消息要需要发送到对应主题。
5.  
   for (String topic : result.unknownLeaderTopics)
6.  
   this.metadata.add(topic, now);
7.  
    
8.  
   log.debug("Requesting metadata update due to unknown leader topics from the batched records: {}",
9.  
   result.unknownLeaderTopics);
10.  
     
11.  
    this.metadata.requestUpdate();
12.  
    }

3.3.4 移除未准备好的node节点

• 检查当前 KafkaProducer 是否与目标 Node 建立了网络连接，如果没有建立，则尝试初始化网络连接，如果初始化失败，则直接返回 false，表示此时不适合向该 Node 发送请求。
• 其次就是检查当前已发送但未响应的请求是否已经达到上限，要是有很多这种请求存在，可能是 broker 宕机了或是 broker 处理能力不足，此时也不适合继续发送请求。
• 除了进行网络方面的检查之外，还会检查 kafka 元数据是否需要更新，如果需要更新的话，也不能发送请求。毕竟使用过期的或是错误的元数据来发送数据，请求也不会发送成功

不适合发送请求的 Node 节点会从 readyNodes 集合中删除。

1.  
   // RecordAccumulator.ReadyCheckResult 获取到所有待发送的节点
2.  
   Iterator<Node> iter = result.readyNodes.iterator();
3.  
   long notReadyTimeout = Long.MAX_VALUE;
4.  
   while (iter.hasNext()) {
5.  
   Node node = iter.next();
6.  
   // 通过KafkaClient进行检查，node节点是否已经准备好
7.  
   if (!this.client.ready(node, now)) {
8.  
    
9.  
   //仅更新延迟统计的readyTimeMs，以便每次批处理就绪时都向前移动（那么readyTimeM和drainTimeMs之间的差异将表示数据等待节点的时间）。
10.  
    this.accumulator.updateNodeLatencyStats(node.id(), now, false);
11.  
    iter.remove();
12.  
    notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));
13.  
    } else {
14.  
    // Update both readyTimeMs and drainTimeMs, this would "reset" the node
15.  
    // latency.
16.  
    //更新readyTimeMs和drainTimeMs，这将“重置”节点延迟。
17.  
    this.accumulator.updateNodeLatencyStats(node.id(), now, true);
18.  
    }
19.  
    }

3.3.5 创建produce请求，将ProducerBatch加入到InFlightBatches中

a、调用记录累加器RecordAccumlator.drain(),获取当前可以集群下的所有数据，并将它们整理成一个批次列表

b、将消息将入到Map<TopicPartition, List<ProducerBatch>> inFlightBatches 的队列中，记录正在发送的这一批数据

c、如果需要进行顺序消息发送，则将通过数据将ProducerBatch的主题信息记录到RecordAccumlator 的Set<TopicPartition> muted 中

1.  
    
2.  
   //获当前可以集群下的所有数据，并将它们整理成一个批次列表
3.  
   Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);
4.  
   // 加入到InFlightBatch队列中
5.  
   addToInflightBatches(batches);
6.  
    
7.  
   if (guaranteeMessageOrder) {
8.  
   // 保证消息顺序发送
9.  
   // 将所有批次的主题加入到Set<TopicPartition> muted中
10.  
    for (List<ProducerBatch> batchList : batches.values()) {
11.  
    for (ProducerBatch batch : batchList)
12.  
    this.accumulator.mutePartition(batch.topicPartition);
13.  
    }
14.  
    }

3.3.6 获取已经超时待发送的消息数据

a、重置消息累加器中下一批次的过期时间

b、获取已达到超时待发送的in-flight batches 批次数据

c、获取累积器中放置时间过长且需要过期的批次数据。

d、将超时待发送的批次数据进行汇总

1.  
   // 重置消息累加器中下一批次的过期时间
2.  
   accumulator.resetNextBatchExpiryTime();
3.  
   // 获取已达到超时待发送的in-flight batches 数据
4.  
   List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
5.  
   // 获取累积器中放置时间过长且需要过期的批次列表。
6.  
   List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);
7.  
   expiredBatches.addAll(expiredInflightBatches);

3.3.7 处理已超时的消息批次，通知该批消息发送失败

1.  
   //如果之前已将过期批次发送给代理，请重置生产者id。同时更新过期批次的指标。请参阅@TransactionState.resetIdempotentProducerId的文档，以了解为什么需要在此处重置生产者id。
2.  
   if (!expiredBatches.isEmpty())
3.  
   log.trace("Expired {} batches in accumulator", expiredBatches.size());
4.  
   for (ProducerBatch expiredBatch : expiredBatches) {
5.  
   String errorMessage = "Expiring " expiredBatch.recordCount " record(s) for " expiredBatch.topicPartition
6.  
   ":" (now - expiredBatch.createdMs) " ms has passed since batch creation";
7.  
   failBatch(expiredBatch, new TimeoutException(errorMessage), false);
8.  
   if (transactionManager != null && expiredBatch.inRetry()) {
9.  
   // This ensures that no new batches are drained until the current in flight batches are fully resolved.
10.  
    //这可确保在当前in-flight batches 完全解决之前，不会排出新批次。
11.  
    transactionManager.markSequenceUnresolved(expiredBatch);
12.  
    }
13.  
    }
14.  
    // 收集统计指标
15.  
    sensors.updateProduceRequestMetrics(batches);

3.3.8 构建请求，发送数据

1.  
   //如果我们有任何节点准备发送 具有可发送数据，则轮询0超时，这样可以立即循环并尝试发送更多数据。
2.  
   // 否则，超时将是下一批到期时间与检查数据可用性的延迟时间之间的较小值。
3.  
   // 请注意，由于延迟、后退等原因，节点可能有尚未发送的数据。这特别不包括具有尚未准备好发送的可发送数据的节点，因为它们会导致繁忙的循环。
4.  
   // 设置下一次的发送延时
5.  
   long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
6.  
   pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now);
7.  
   pollTimeout = Math.max(pollTimeout, 0);
8.  
   if (!result.readyNodes.isEmpty()) {
9.  
   log.trace("Nodes with data ready to send: {}", result.readyNodes);
10.  
     
11.  
    //如果某些分区已经准备好发送，则选择时间将为0；
12.  
    // 否则，如果某个分区已经积累了一些数据，但还没有准备好，则选择时间将是现在与其延迟到期时间之间的时间差；
13.  
    // 否则，选择时间将是现在与元数据到期时间之间的时间差；
14.  
     
15.  
    pollTimeout = 0;
16.  
    }
17.  
     
18.  
    sendProduceRequests(batches, now);

3.4 Sender线程sendProduceRequest()的核心逻辑梳理

1.  
   /**
2.  
   * 从给定的记录批次创建生产请求
3.  
   */
4.  
   private void sendProduceRequest(long now, int destination, short acks, int timeout, List<ProducerBatch> batches) {
5.  
   if (batches.isEmpty())
6.  
   return;
7.  
    
8.  
   final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());
9.  
    
10.  
    // 找到创建记录集时使用的最小魔法版本
11.  
    byte minUsedMagic = apiVersions.maxUsableProduceMagic();
12.  
    for (ProducerBatch batch : batches) {
13.  
    if (batch.magic() < minUsedMagic)
14.  
    minUsedMagic = batch.magic();
15.  
    }
16.  
    // ProduceRequestData
17.  
    ProduceRequestData.TopicProduceDataCollection tpd = new ProduceRequestData.TopicProduceDataCollection();
18.  
     
19.  
    for (ProducerBatch batch : batches) {
20.  
    TopicPartition tp = batch.topicPartition;
21.  
    MemoryRecords records = batch.records();
22.  
     
23.  
    //如有必要，向下转换到使用的最小魔法的消息格式。
24.  
    // 通常情况下生产者开始构建批次以及我们发送请求的时间选择了基于过时元数据的消息格式。
25.  
    // 在最坏的情况下，如果选择使用新的消息格式，但发现代理不支持它，所系需要在客户端发送之前进行消息格式的向下转换。
26.  
    // 这旨在处理围绕集群升级的边缘情况，其中代理可能并非所有消息都支持相同的消息格式版本。
27.  
    // 例如，如果分区从代理迁移它支持新的魔法版本，但不支持，就需要转换。
28.  
    if (!records.hasMatchingMagic(minUsedMagic))
29.  
    records = batch.records().downConvert(minUsedMagic, 0, time).records();
30.  
    ProduceRequestData.TopicProduceData tpData = tpd.find(tp.topic());
31.  
    if (tpData == null) {
32.  
    tpData = new ProduceRequestData.TopicProduceData().setName(tp.topic());
33.  
    tpd.add(tpData);
34.  
    }
35.  
    tpData.partitionData().add(new ProduceRequestData.PartitionProduceData()
36.  
    .setIndex(tp.partition())
37.  
    .setRecords(records));
38.  
    recordsByPartition.put(tp, batch);
39.  
    }
40.  
     
41.  
    String transactionalId = null;
42.  
    if (transactionManager != null && transactionManager.isTransactional()) {
43.  
    transactionalId = transactionManager.transactionalId();
44.  
    }
45.  
     
46.  
    ProduceRequest.Builder requestBuilder = ProduceRequest.forMagic(minUsedMagic,
47.  
    new ProduceRequestData()
48.  
    .setAcks(acks)
49.  
    .setTimeoutMs(timeout)
50.  
    .setTransactionalId(transactionalId)
51.  
    .setTopicData(tpd));
52.  
    RequestCompletionHandler callback = response -> handleProduceResponse(response, recordsByPartition, time.milliseconds());
53.  
     
54.  
    // 创建clientRequest请求
55.  
    String nodeId = Integer.toString(destination);
56.  
    ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0,
57.  
    requestTimeoutMs, callback);
58.  
    // 通过KafkaClient的实现类NetworkClient 调用发送
59.  
    client.send(clientRequest, now);
60.  
    log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
61.  
    }

说明：源码分析基于Kafak 3.3版本 

这篇好文章是转载于：学新通技术网