Kafka生产者源码解析（二，java面试题汇总及答案

Kafka生产者源码解析（二，java面试题汇总及答案

进入RecordAccumulator类中，可以看到它有很多的属性字段，其中batches这个字段需要引起我们的注意，它是一个以TopicPartition作为key，Deque作为value的ConcurrentMap，TopicPartition存储了topic及partition信息，能够标记消息属于哪个主题和应该发往哪个分区；Deque是一个双端队列，里面存放的是ProducerBatch对象，ProducerBatch用于存储一批将要被发送的消息记录；ProducerBatch通过MemoryRecordsBuilder对象拥有一个DataOutputStream对象的引用，这里就是我们消息存放的最终归宿，根据MemoryRecordsBuilder构造方法的源码可知DataOutputStream里面持有ByteBufferOutputStream，这是一个缓存buffer，所以往DataOutputStream里面写消息数据，就是往缓存里面写消息数

《一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义》

【docs.qq.com/doc/DSmxTbFJ1cmN1R2dB】 完整内容开源分享

据。

 

最后存入RecordAccumulator中的消息将会是这样。

 

二、append方法解析

---

RecordAccumulator的构造方法中通过CopyOnWriteMap初始化了上述谈到的batches对象，同时还初始化了其他的属性内容，这里不再赘述其构造的过程，而是着重分析上一篇中遗留的内容：KafkaProducer是如何通过accumulator.append方法将消息追加到RecordAccumulator消息累加器中的。

public RecordAppendResult append(TopicPartition tp,

long timestamp,

byte[] key,

byte[] value,

Header[] headers,

Callback callback,

long maxTimeToBlock) throws InterruptedException {

//并发数加1，统计正在向RecordAccumulator中追加消息的线程数

appendsInProgress.incrementAndGet();

ByteBuffer buffer = null;

if (headers == null) headers = Record.EMPTY_HEADERS;

try {

//查找TopicPartition对应的Deque，如果没有则创建

Deque dq = getOrCreateDeque(tp);

//追加消息时需要加锁

synchronized (dq) {

if (closed)

throw new KafkaException(“Producer closed while send in progress”);

//尝试往Deque中最后一个ProducerBatch中追加消息记录

RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);

if (appendResult != null)

//消息追加成功返回结果

return appendResult;

}

//来到这一步说明上面消息追加失败

byte maxUsableMagic = apiVersions.maxUsableProduceMagic();

//获取要创建的ProducerBatch的内存大小

int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));

log.trace(“Allocating a new {} byte message buffer for topic {} partition {}”, size, tp.topic(), tp.partition());

//从BufferPool中申请空间用于后面创建新的ProducerBatch

buffer = free.allocate(size, maxTimeToBlock);

//和上面一样，追加消息时需要加锁

synchronized (dq) {

// Need to check if producer is closed again after grabbing the dequeue lock.

if (closed)

throw new KafkaException(“Producer closed while send in progress”);

//在创建新的ProducerBatch之前再次尝试往Deque中最后一个ProducerBatch中追加消息记录，说不定现在成功了呢

RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);

if (appendResult != null) {

//消息追加成功返回结果

return appendResult;

}

//如果消息还是追加失败了。。。

//构造MemoryRecordsBuilder，消息将会存入它拥有的MemoryRecords对象

MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic);

//创建ProducerBatch

ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, time.milliseconds());

//使用batch.tryAppend追加消息

FutureRecordmetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds()));

//将刚创建的ProducerBatch放入Deque双端队列尾部

dq.addLast(batch);

incomplete.add(batch);

//到这里消息已经追加成功，将buffer置空

buffer = null;

//返回结果

return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true);

}

} finally {

if (buffer != null)

//释放之前申请的新空间

free.deallocate(buffer);

//结束，并发数减1

appendsInProgress.decrementAndGet();

}

}

上面的代码已经给出了注释，现将这段代码的流程总结如下：

1. 首先根据TopicPartition参数获取到对应的Deque双端队列，没有则创建。

2. 使用Synchronized关键字对追加消息的 *** 作加锁。

3. 调用tryAppend方法尝试往Deque中最后一个ProducerBatch中追加消息记录，如果成功则返回RecordAppendResult结果，Synchronized解锁。

4. 如果上面追加消息失败，则从BufferPool中申请新的空间用于后面创建新的ProducerBatch。

5. 使用Synchronized关键字对追加消息的 *** 作加锁，然后再次尝试第三步。

6. 到这一步说明上面的第二次尝试仍然没有成功，那么使用第四步申请到的空间创建新的ProducerBatch。

7. 将消息记录追加到新建的ProducerBatch中，然后将新建的ProducerBatch插入到Deque双端队列尾部，并将它放入incomplete集合。

8. 最后Synchronized解锁，到这里消息追加已经成功，返回RecordAppendResult结果，它将作为唤醒Sender线程的条件。

这段代码的核心部分便是batch.tryAppend方法，下面是该方法的部分源码，首先是检查了一下消息存储器的剩余空间是否充足，若不足则直接返回null，后面走申请空间新建ProducerBatch的流程。如果空间剩余充足则MemoryRecordsBuilder会调用append方法进行消息追加。

public FutureRecordmetadata tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long now) {

//检查消息存储器中剩余空间是否充足，若空间不足则直接返回null

if (!recordsBuilder.hasRoomFor(timestamp, key, value, headers)) {

return null;

} else {

//消息写入

Long checksum = this.recordsBuilder.append(timestamp, key, value, headers);

……………………

return future;

}

}

然后像洋葱一样不断剥开append方法的皮，，，，，发现MemoryRecordsBuilder最终会根据KafkaProducer客户端版本的不同去调用下面两个方法之一：appendDefaultRecord和appendLegacyRecord。

private void appendDefaultRecord(long offset, long timestamp, ByteBuffer key, ByteBuffer value,

Header[] headers) throws IOException {

………………

int sizeInBytes = DefaultRecord.writeTo(appendStream, offsetDelta, timestampDelta, key, value, headers);

recordWritten(offset, timestamp, sizeInBytes);

}

private long appendLegacyRecord(long offset, long timestamp, ByteBuffer key, ByteBuffer value) throws IOException {

………………

long crc = LegacyRecord.write(appendStream, magic, timestamp, key, value, CompressionType.NONE, timestampType);

recordWritten(offset, timestamp, size + Records.LOG_OVERHEAD);

return crc;

}