〇、前言
在上一篇文章《连Producer端的主线程模块运行原理都不清楚,就敢说自己精通Kafka
》中,我们介绍了Main Thread的工作原理,那么在本篇文章中,我们继续介绍第二部分内容:RecordAccumulator。
在介绍原理之前,大家再重温一下Producer端的整体架构,图示如下所示:
这个图看不懂没有关系,我们会在介绍Producer端原理时一一介绍每个部分的含义及其所复杂的功能。
一、RecordAccumulator
在上文中,我们介绍了主线程(Main Thread
)的执行流程,当我们使用KafkaProducer发送消息的时候,消息会经过拦截器(Interceptor
)、序列化器(Serializer
)和分区器(Partitioner
),最后会暂存到消息收集器(RecordAccumulator
)中,那么,本节就来针对其进行介绍。
RecordAccumulator的主要作用是暂存Main Thread
发送过来的消息,然后Sender Thread
就可以从RecordAccumulator中批量的获取到消息,减少单个消息获取的请求次数,提升性能效率。通过参数buffer.memory
可以设置缓存大小(默认32M
)。
properties.put(
ProducerConfig.BUFFER_MEMORY_CONFIG
, 3210241024);
由于RecordAccumulator的缓存空间有限,如果空间被占满,那么当我们再次调用KafkaProducer的send(...)方法的时候,就会出现阻塞(默认60秒
,可以通过参数max.block.ms
来配置),如果阻塞超时,则会抛出异常。
properties.put(
ProducerConfig.MAX_BLOCK_MS_CONFIG
, 60*1000);
在RecordAccumulator中,我们通过getOrCreateDeque(...)
方法来创建存储消息的数据结构,即:存储ProducerBatch实例对象的双向队列Deque;源码如下所示:
private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches; // 主题分区:双向队列
private Deque<ProducerBatch> getOrCreateDeque(TopicPartition tp) {
Deque<ProducerBatch> d = this.batches.get(tp);
if (d != null)
return d;
d = new ArrayDeque<>(); // 创建双向队列
Deque<ProducerBatch> previous = this.batches.putIfAbsent(tp, d);
if (previous == null)
return d;
else
return previous;
}
其对应关系是通过一个主题分区
对应双向队列Deque<ProducerBatch>
,维护在batches
中的,如下图所示:
这时可能会有同学问?我记得调用KafkaProducer发送消息的时候,我们发送的是ProducerRecord
实例对象,怎么在Deque
双向队列中存储的是ProducerBatch
实例对象,他们两个有啥区别呢?ProducerRecord
是我们使用KafkaProducer发送消息时拼装的单条消息,而ProducerBatch
可以看做是针对一批消息进行的封装,因为会在RecordAccumulator中执行tryAppend
方法将一批消息拼装在一起,可以减少网络请求次数从而提升吞吐量。
Kafka通过ByteBuffer来实现字节形式的网络传输,为了减少频繁创建/释放ByteBuffer所造成的资源消耗,Kafka还提供了缓冲池(BufferPool)来实现ByteBuffer的回收,再其内部维护了Deque<ByteBuffer> free
变量来保存空闲ByteBuffer,还提供了Deque<Condition> waiters
变量来保存阻塞等待中的线程。
如果待分配的size等于缓冲池中ByteBuffer的大小(可由batch.size
参数进行配置,默认为16Kb),则直接从free队列中拿出空余的ByteBuffer供其使用;否则,判断如果缓冲池中空闲ByteBuffer的内存总和加上非缓冲池内存大小是大于待分配size的,则采用非缓冲池
加上缓冲池
混合释放内存的方式进行内存分配。代码如下所示:
关于batch.size
参数,除了可以影响BufferPool中缓存的ByteBuffer是否被立刻复用之外,还与创建ProducerBatch有关。当我们通过KafkaProducer发送一条由ProducerRecord封装的消息,并交由RecordAccumulate处理时,会执行如下步骤:
【1】根据主题分区寻找对应的双向队列Deque<ProducerBatch>,从中获取ProducerBatch;<br> 【2】如果这个ProducerBatch还有剩余空间,则直接写入;如果无法写入,则继续执行如下逻辑;<br> 【3】如果待保存的消息size
小于等于
batch.size,则创建batch.size大小的ProducerBatch,当使用完毕后,交由BufferPool管理复用;<br> 【4】如果待保存的消息size大于
batch.size,那么就创建消息size大小的ProducerBatch,这段内存区域不会被复用。<br>
今天的文章内容就这些了:
写作不易,笔者几个小时甚至数天完成的一篇文章,只愿换来您几秒钟的 点赞 & 分享 。
更多技术干货,欢迎大家关注公众号“爪哇缪斯” ~ \(^o^)/ ~ 「干货分享,每天更新」