一、Producer 的网络模型

咱们后面几篇有说 Producer 发送流程的源码剖析，但那个是大的轮廓，波及到发送很多相干的内容，比方：

• 获取 topic 的 metadata 信息
• key 和 value 的序列化
• 获取该 record 要发送到的 partition
• 向 RecordAccmulator 中追加 record 数据
• 唤醒 sender 线程发送 RecordBatch

那这篇老周次要来说下 Producer 的网络模型，这里间接给出 Producer 的网络模型图，如下：

从图中能够看出，KafkaProducer 相当于客户端，与 Sender 调用层交互，Sender 调用 NetworkClient，NetworkClient 调用 Selector，而 Selector 底层封装了 Java NIO 的相干接口。心中有了 Producer 的网络模型大抵轮廓后，咱们接下来就能够来剖析 Producer 的网络模型。

二、Producer 与 Broker 的交互流程

咱们在业务代码通过生产者 producer 调用 send 办法来发送音讯，不难发现都是通过走 Producer 的实现类 KafkaProducer 的 send 办法：

2.1 org.apache.kafka.clients.producer.KafkaProducer#doSend

下面的两个 send 办法最终会走到 doSend 办法里来：

这块的源码老周在前两篇的 Producer 源码解析那一篇剖析了的哈，这里次要说下与 Broker 通信的交互剖析。次要有两点：

• waitOnMetadata()：申请 tp（topic-partition）元数据 metadata 更新，两头会调用 sender.wakeup()。
• ccumulator.append()：将 record 对应的 tp 写入到 deque 中，如果该 tp 对应的 deque batch 是满了或者新建了一个 batch，则会调用 sender.wakeup()。

次要看下 sender.wakeup() 办法，次要作用就是将 Sender 线程从阻塞中唤醒。

2.2 org.apache.kafka.clients.producer.internals.Sender#wakeup

/**
 * Wake up the selector associated with this send thread
 */
public void wakeup() {this.client.wakeup();
}

/**
 * Interrupt the client if it is blocked waiting on I/O.
 */
@Override
public void wakeup() {this.selector.wakeup();
}

/**
 * Interrupt the nioSelector if it is blocked waiting to do I/O.
 */
@Override
public void wakeup() {this.nioSelector.wakeup();
}

不难发现，调用链是：

Sender -> NetworkClient -> Selector（Kafka 封装的）-> Selector（java.nio.channels.Selector Java NIO）

wakeup() 的次要作用就是唤醒阻塞在 select()/select(long) 上的线程，为什么要唤醒？因为注册了新的 channel 或者事件。

再回到 Kafka 这里，KafkaProducer 中 dosend() 办法调用 sender.wakeup() 办法作用就很显著了。作用就是：当有新的 RecordBatch 创立后，旧的 RecordBatch 就能够发送了，如果线程阻塞在 select() 办法中，就将其唤醒，Sender 从新开始运行 run() 办法，在这个办法中，旧的 RecordBatch 将会被选中，进而能够及时将这些申请发送进来。

2.3 org.apache.kafka.clients.producer.internals.Sender#run

跟到 runOnce 办法里去：

持续跟，外围是 Sender 线程每次循环具体执行的中央，即 sendProducerData() 办法：

最初调用 client.poll() 办法，对于 socket 的一些理论的读写操作。

咱们来小结一下 Sender.run() 办法的大抵流程，次要分为以下五步：

• accumulator.ready()：遍历整个 batches（key：TopicPartition，value: Deque<ProducerBatch>>），如果 ProducerBatch 不为空，就将其对应的 leader 选出来，最初会返回一个能够发送 ReadyCheckResult 实例，readyNodes 是次要的成员变量。
  
  
• 如果有 tp 的 leader 是未知的，就强制 metadata 更新。遍历未知 leader 的主题（蕴含 leader 选举未决的主题以及可能曾经过期的主题），再次将主题增加到元数据以确保它被蕴含并申请元数据更新，因为有音讯要发送到该主题，调用 requestUpdate() 办法来更新。
• accumulator.drain()：遍历每个 leader（第一步中选出）节点，获取该节点上所有的 tp，如果该 tp 对应的 ProducerBatch 不在 backoff 期间（没有重试过或者重试了然而距离曾经达到了 retryBackoffMs），并且 ProducerBatch 的大小不大于 maxSize（一个 request 的最大限度默认 1 MB）或 ProducerBatch 的汇合是空的，那么就把这个 ProducerBatch 增加 list 中，最终返回的类型为 Map<Integer, List<ProducerBatch>>，key 为 leader.id，value 为要发送的 ProducerBatch 的列表。
• sendProduceRequests()：发送 Producer 申请，这个办法会调用 NetworkClient.send() 来发送 clientRequest。
• NetworkClient.poll()：对于 socket 的一些理论的读写操作，这个办法会持续调用 Kafka 封装的 Selector.poll()，跟进去底层是调用的 Java NIO 的 Selector.poll()。

2.4 org.apache.kafka.clients.NetworkClient#poll

次要分为以下三步：

• metadataUpdater.maybeUpdate()：判断是否须要更新 metadata 元数据。抉择具备起码未实现申请且至多合乎连贯条件的节点。此办法将更喜爱具备现有连贯的节点，但如果所有现有连贯都在应用，则可能会抉择咱们还没有连贯的节点。如果不存在连贯，则此办法将首选最近连贯尝试起码的节点。这种办法永远不会抉择一个没有现有连贯的节点，并且咱们在从新连贯回退期间断开了连贯，或者抉择了一个被限度的流动连贯。
• selector.poll()：进行 socket 相干的 IO 操作。

• 解决实现后的操作：在一个 select() 过程之后的相干解决

  • handleAbortedSends(responses)：如果因为不受反对的版本异样或断开连接而停止发送，请立刻解决，无需期待 Selector#poll。
  • handleCompletedSends(responses, updatedNow)：解决任何已实现的申请发送。特地是如果预期没有响应，则认为申请已实现。
  • handleCompletedReceives(responses, updatedNow)：解决任何已实现的接管并应用接管到的响应更新响应列表。（MetadataResponse、ApiVersionsResponse 都是在这解决的）
  • handleDisconnections(responses, updatedNow)：解决连贯失败的那些连贯，从新申请 metadata。
  • handleConnections()：解决新建设的那些连贯（还不能发送申请，比方：还未认证）
  • handleInitiateApiVersionRequests(updatedNow)：对那些新建设的连贯，发送 apiVersionRequest（默认状况：第一次建设连贯时，须要向 Broker 发送 ApiVersionRequest 申请）
  • handleTimedOutRequests(responses, updatedNow)：解决 timeout 的连贯，敞开该连贯，并刷新 metadata。

2.5 org.apache.kafka.common.network.Selector#poll

Kafka 中的 Selector 类次要是 Java NIO 相干接口的封装，Socket 相干 IO 操作都是在这个类中实现的。次要操作都是在上面这个办法中调用的：

2.5.1 org.apache.kafka.common.network.Selector#clear

2.5.2 org.apache.kafka.common.network.Selector#select

Selector.select() 办法底层还是调用的 Java NIO 的原生接口，这里的 nioSelector 其实就是 java.nio.channels.Selector 的实例对象，这个办法最坏状况下，会阻塞 ms 的工夫，如果在一次轮询，只有有一个 Channel 的事件就绪，它就会立马返回。

/**
 * Check for data, waiting up to the given timeout.
 *
 * @param timeoutMs Length of time to wait, in milliseconds, which must be non-negative
 * @return The number of keys ready
 */
private int select(long timeoutMs) throws IOException {if (timeoutMs < 0L)
        throw new IllegalArgumentException("timeout should be >= 0");

    if (timeoutMs == 0L)
        return this.nioSelector.selectNow();
    else
        return this.nioSelector.select(timeoutMs);
}

2.5.3 org.apache.kafka.common.network.Selector#pollSelectionKeys

这部分代码是 socket IO 的次要局部，发送 Send 及接管 Receive 都是在这里实现的，在 poll() 办法中，这个办法会调用三次：

• 解决从具备缓冲数据的通道的轮询事件
• 解决曾经就绪的事件，进行相应的 IO 操作。
• 解决新建设的那些连贯，增加缓存及传输层的握手与认证。

2.5.4 org.apache.kafka.common.network.Selector#addToCompletedReceives

/**
 * adds a receive to completed receives
 */
private void addToCompletedReceives(KafkaChannel channel, NetworkReceive networkReceive, long currentTimeMs) {if (hasCompletedReceive(channel))
        throw new IllegalStateException("Attempting to add second completed receive to channel" +                                             channel.id());
        // 增加到 completedReceives 中
    this.completedReceives.put(channel.id(), networkReceive);
    sensors.recordCompletedReceive(channel.id(), networkReceive.size(), currentTimeMs);
}

接管到的所有 Receive 都会被放入到 completedReceives 的汇合中期待后续解决。

三、总结

总结的话我就不复述了，下面的源码流程说的很分明了。最初再来一张 Producer 网络模型的时序图：

心愿对你有所帮忙，咱们下期再见。