目录

客户端逻辑

客户端由两类线程运行：主线+sender线程
主线程中：由kafkaProducer创建消息，通过拦截器、序列化器、分区器之后缓存到消息累加器中(RecordAccumulator)
sender线程：从RecordAccumulator中获取消息并发送到kafka中

RecordAccumulator用来缓存消息以便sender线程可以批量发送，进而减少网络传输的资源消耗。
RecordAccumulator缓存的大小可以通过buffer.memory配置，默认32MB
如果生产者发送速度超过sender消费速度，那么kafkaProducer的send方法要么阻塞要么抛异常。取决于参数max.block.ms的配置，默认60s

RecordAccumulator内部为每隔分区都维护了一个Deque<ProducerBatch>
主线程从尾部追加，sender从头部读取。

1、当一条消息(ProducerRecord)流入RecordAccumulator时，先去找topic对应的Deque<ProducerBatch>(若无则新建)
2、从末尾获取一个ProducerBatch，判断是否可以将这个ProducerRecord塞进去
3、如果不可以则新建一个ProducerBatch，新建需要注意判断是否会超过batch.size，
如果不超过，那么可以通过BufferPool来管理复用
如果超过，那么就自动创建一个不会用来被复用的内存区域

sender从RecordAccumulator获取缓存的消息之后，会进一步将原本的<topic, Deque<ProducerBatch>>的保存形式转变为：
<Node, List<ProducerBatch>>的形式。
Node指的是broker节点。
对于网络连接来说，不关系具体topic，只关心具体broker阶段。
对于KafkaProducer的应用逻辑而言，只关心向哪个分区进行发送消息。
所以这里需要做一层应用逻辑层到网络I/O层的转换。
<Node，List<ProducerBatch>>的形式之后，Sender 还会进一步封装成<Node，Request>的形式
请求在发往kafka之前还会保存在InFlightRequests中，InFlightRequests保存对象的具体形式为Map<NodeId, Deque<Request>>
主要是缓存已经发出去但还没有收到响应的请求。
max.in.flight.requests.per.connection，默认值为5，即每个连接最多只能缓存5个未响应的请求，超过该数值之后就不能再向这个连接发送更多请求了。

通过比较Deque＜Request＞的size与这个参数的大小来判断对应的Node中是否已经堆积了很多未响应的消息，如果真是如此，那么说明这个 Node 节点负载较大或网络连接有问题，再继续向其发送请求会增大请求超时的可能。

InFlightRequests还可以从所有Node中选择出最小负载的哪一个，这里取决于InFlightRequests中还未确认的请求决定的，未确认的请求越多则认为负载越大。

选择leastLoadedNode发送请求可以使它能够尽快发出，避免因网络拥塞等异常而影响整体的进度。

元数据

元数据指的是kafka集群的元数据，记录了：

  • 集群中有哪些主题
  • 这些主题有哪些分区
  • 每个分区的leader副本分配在哪个节点上，follower副本分配在哪个节点
  • 哪些副本在AR、ISR等集合中
  • 进群有哪些节点
  • 控制器节点是哪个

当没有指定的主题信息，或者超过metadata.max.age.ms 时间没有更新元数据都会引起元数据的更新操作，该更新操作时客户端内部执行的，对外部使用者不可见。

当需要更新元数据时，会先挑选出leastLoadedNode，然后向这个Node发送MetadataRequest请求来获取具体的元数据信息。这个更新操作是由Sender线程发起的，在创建完MetadataRequest之后同样会存入InFlightRequests，之后的步骤就和发送消息时的类似。元数据虽然由Sender线程负责更新，但是主线程也需要读取这些信息，这里的数据同步通过synchronized和final关键字来保障

producer调参

指定分区中必须有多少副本收到这条消息，生产者才认为写入成功。