runOnce()
序列化key valuekeySerializer.serialize() valueSerializer.serialize
doSend
获取元数据信息waitOnMetadata()
获取准备好发送的分区列表
for循环 按每个节点传输List<ProducerBatch>>sendProduceRequest
sender线程
生产者
kafkaProducer.send()kafkaProducer.doSend()
累加到accumulator
kafkaTemplate.send()
构造clientRequest
broker
强制更新元数据this.metadata.requestUpdate()
for循环找出最小的magic
从所有的batches中获取达到投递时间的batchthis.accumulator.expiredBatches(now)
// 获取该topic的元数据信息1.waitOnMetadata(...)// 序列化key value2. keySerializer.serialize() valueSerializer.serialize// 根据分区路由规则获取到发送到哪个分区3.partition()4. 本条消息的大小(字节数)AbstractRecords.estimateSizeInBytesUpperBound()5.验证 大小是否超过限制ensureValidRecordSize()6. accumulator.append() 缓存消息 将消息累加到对应的批次中7. 若是新的批次6会累加失败，会再一次accumulator.append() 创建一个新的Deque放到batchs里 并将该消息放到最后位置8. 如果该批次空间大小已满或者是一个新的批次唤醒sender线程9.return future
创造发送请求this.accumulator.drain
拦截器
return future
this.accumulator.ready()
对应代码
sendProduceRequests
根据分区规则获取发动到的partition
for循环将records根据magic向下转换
构造requestBuildercallback
如果不是内部请求 !isInternalRequest验证是否能够发送 !canSendRequest
partitioner
更新发送请求相关的metricsupdateProduceRequestMetrics
interceptor
获取version
producer
selector.send(send);
producer.send()
如果存在不知道leader节点的topic
long pollTimeout = sendProducerData(currentTimeMs)
验证该条消息大小
1. inProcess == null 没有在进行中的请求 & 元数据更新时间到了2 链接状态没有ready || 终止时间小于现在时间 3. 该node的kafakaChannel为空或者不是ready4. 该node在InFlightRequest的Deque为空 或者里面的个数大于配置的maxInFlightRequestsPerConnection
将准备发送的批次放到inflightBatch中addToInflightBatches
分区路由
转换后的record放入到一个mapproducerbatch放到一个map
代码发送到RecordAccumulator流程
NetWorkClient.doSend()
序列化器
释放该batch申请的空间从inflightbatch中移除
将上诉两步获取的list合并，for循环failBatch
ProducerBatch
kafka发送流程图
从inflightbatch获取达到投递时间的batchgetExpiredInflightBatches
本地缓存RecordAccumulator
KafkaTemplatedoSend()
移除没有准好的节点
构造了两个map
serializer
