〇、前言

在上两篇文章《连Producer端的主线程模块运行原理都不清楚，就敢说自己精通Kafka》和《一文了解Kafka的消息收集器RecordAccumulate》中，我们介绍了Main Thread和RecordAccumulate的工作原理，那么在本篇文章中，我们继续介绍第三部分内容：Sender线程。

在介绍原理之前，大家再重温一下Producer端的整体架构，图示如下所示：

这个图看不懂没有关系，我们会在介绍Producer端原理时一一介绍每个部分的含义及其所复杂的功能。

一、Sender线程

除了我们前面曾经介绍过的Main Thread主线程之外，在KafkaProducer中还启动了一个Sender线程，那么，本节我们就来针对Sender线程进行解析，Send线程启动代码如下所示：

Sender线程负责从RecordAccumulate中获取缓存消息，在获取了以Map<TopicPartition，Deque<ProducerBatch>> 的对应关系存储的消息缓存之后，会通过主题信息和分区信息创建TopicPartition实例对象tp，然后再以此为key，获取ProducerBatch的双向队列，如下所示：

然后，会进一步将映射中key的类型从TopicPartition转换为NodeId，即：Map<NodeId，Deque<ProducerBatch>> 的对应关系。这是由于当Producer端最终发送消息的时候，关注的是向哪个Broker节点发送消息，而并不是关心哪个主题分区，所以此处需要做一个从应用逻辑层面向网络I/O层面的转换。如下所示：

当最后要进行消息发送的时候，还要再次进行封装，封装出用于消息发送的ProduceRequest，此时的对应关系就变成了NodeId和ProduceRequest了，代码如下所示：

请求在从Sender线程发往Kafka之前还会保存到InFlightRequests中，InFlightRequests保存对象的具体形式为 Map<NodeId，Deque<Request>>，它的主要作用是缓存了已经发出去但还没有收到响应的请求。代码如下所示：

如果我们开启幂等配置，则会创建TransactionManager实例对象，如下所示：

当TransactionManager实例对象不为null的时候，Sender现成则会执行下图红框内的代码逻辑，那么在黄框中，我们会看到调用了maybeSendAndPollTransactionalRequest()方法，代码如下所示：

在选择目标节点的时候，如果coordinatorType为空，则会调用client的leastLoadeNode(...)方法，通过该方法可以获得所有Node中负载最小的那一个。那怎么来判断Node节点的负载呢？ 在上面的内容中，我们其实提到了，发出去的消息也会保存到InFlightRequests中，它其实是一个缓存的作用，主要用来缓存已经发出去但是还没有接收到响应确认的消息请求。因此，我们可以通过它来判断那些Node节点未接到响应的请求数量最少，则就是负载最小的。通过这种方式，可以保证负载尽可能的平均，而不会造成某一个节点的重度阻塞从而影响整个消息的发送性能。代码如下所示：

除了上面之外，我们在元数据更新的时候，也是通过leastLoadeNode(...)方法来获得负载最低的节点的，那么，在Kafka中什么是元数据呢？ 当我们发送消息的时候，消息发送到哪个分区，这个分区对应的Broker的地址和端口，已经这个是否配置了Kafka集群，集群中都包含哪些节点等等，都是保存在元数据信息中的。那么，在什么步骤触发了元数据更新呢？ 我们可以把视野转向Sender的runOnce()方法上，在下图红框处，我们调用了client的poll方法，如下是该方法的源码：

此处的client对应的是NetworkClient的实例对象，在该类的poll(...)方法中，执行了更新元数据的逻辑，即下图红框所示：

在maybeUpdate方法中，我们看到了熟悉的一段代码Node node = leastLoadedNode(now); 此处就是获得负载最低节点的地方。那么获得到了这个node之后，就可以调用maybeUpdate(now, node)来尝试更新元数据信息了：

在maybeUpdate(now, node)方法中我们可以看到，更新元数据也是采用发送消息的方式，即：向这个负载最低的Node发送MetadataRequest请求来获取具体的元数据信息。在创建完MetadataRequest之后同样会存入InFlightRequests，之后的步骤就和发送消息时的处理逻辑一样的：