之前咱们剖析了 Producer 的配置解析、组件剖析、拉取元数据、音讯的初步序列化形式、音讯的路由策略。如下图：
这一节咱们持续剖析发送音讯的内存缓冲器原理—RecordAccumulator.append()。
如何将音讯放入内存缓冲器的？
在 doSend 中的，拉取元数据、音讯的初步序列化形式、音讯的路由策略之后就是 accumulator.append()。
如下代码所示：（去除了多余的日志和异样解决，截取了外围代码）
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
TopicPartition tp = null;
try {
// 拉取元数据、音讯的初步序列化形式、音讯的路由策略
long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);
long remainingWaitMs = Math.max(0, this.maxBlockTimeMs - waitedOnMetadataMs);
byte[] serializedKey = keySerializer.serialize(record.topic(), record.key());
byte[] serializedValue = valueSerializer.serialize(record.topic(), record.value());
int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
ensureValidRecordSize(serializedSize);
tp = new TopicPartition(record.topic(), partition);
long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();
Callback interceptCallback = this.interceptors == null ?
callback : new InterceptorCallback<>(callback, this.interceptors, tp);
// 将路由后果、初步序列化的音讯放入到音讯内存缓冲器中
RecordAccumulator.RecordAppendResult result =
accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);
if (result.batchIsFull || result.newBatchCreated) {this.sender.wakeup();
}
return result.future;
} catch (Exception e) {throw e;}
// 省略其余各种异样捕捉
}accumulator.append() 它次要是将路由后果、初步序列化的音讯放入到音讯内存缓冲器中。
剖析如何将音讯放入内存缓冲器之前，须要回顾下它外部的根本构造。之前组件剖析的时候，咱们初步剖析过 RecordAccumulator 的大体构造，如下图：
1）设置了一些参数 batchSize、totalSize、retryBackoffMs、lingerMs、compression 等
2）初始化了一些数据结构，比方 batches 是一个 new CopyOnWriteMap<>()
3）初始化了 BufferPool 和 IncompleteRecordBatches
回顾了 RecordAccumulator 这个组件之后，咱们就来看看到底如何将音讯放入内存缓冲器的数据结构中的。
public RecordAppendResult append(TopicPartition tp,
long timestamp,
byte[] key,
byte[] value,
Callback callback,
long maxTimeToBlock) throws InterruptedException {
// We keep track of the number of appending thread to make sure we do not miss batches in
// abortIncompleteBatches().
appendsInProgress.incrementAndGet();
try {
// check if we have an in-progress batch
Deque<RecordBatch> dq = getOrCreateDeque(tp);
synchronized (dq) {if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
RecordAppendResult appendResult = tryAppend(timestamp, key, value, callback, dq);
if (appendResult != null)
return appendResult;
}
// we don't have an in-progress record batch try to allocate a new batch
int size = Math.max(this.batchSize, Records.LOG_OVERHEAD + Record.recordSize(key, value));
log.trace("Allocating a new {} byte message buffer for topic {} partition {}", size, tp.topic(), tp.partition());
ByteBuffer buffer = free.allocate(size, maxTimeToBlock);
synchronized (dq) {
// Need to check if producer is closed again after grabbing the dequeue lock.
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
RecordAppendResult appendResult = tryAppend(timestamp, key, value, callback, dq);
if (appendResult != null) {
// Somebody else found us a batch, return the one we waited for! Hopefully this doesn't happen often...
free.deallocate(buffer);
return appendResult;
}
MemoryRecords records = MemoryRecords.emptyRecords(buffer, compression, this.batchSize);
RecordBatch batch = new RecordBatch(tp, records, time.milliseconds());
FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, callback, time.milliseconds()));
dq.addLast(batch);
incomplete.add(batch);
return new RecordAppendResult(future, dq.size() > 1 || batch.records.isFull(), true);
}
} finally {appendsInProgress.decrementAndGet();
}
}整个办法的脉络，看着逻辑比拟多，波及了很多数据结构，咱们一步一步来剖析下。第一次看的话，大体你能够梳理如下脉络：
1）getOrCreateDeque 这个办法应该是才创立一个双端队列，队列放的每一个元素不是单条音讯 Record，而是音讯的汇合 RecordBatch。
2）free.allocate 应该是在分配内存缓冲器中的内存
3）tryAppend 应该是将音讯放入内存中
创立寄存音讯汇合的队列
在将音讯放入内存缓冲器之前，首先通过 getOrCreateDeque 创立的是一个寄存音讯汇合的队列。代码如下：
private final ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches;
public RecordAccumulator(int batchSize,
long totalSize,
CompressionType compression,
long lingerMs,
long retryBackoffMs,
Metrics metrics,
Time time) {
// 省略...
this.batches = new CopyOnWriteMap<>();
// 省略...
}
/**
* Get the deque for the given topic-partition, creating it if necessary.
*/
private Deque<RecordBatch> getOrCreateDeque(TopicPartition tp) {Deque<RecordBatch> d = this.batches.get(tp);
if (d != null)
return d;
d = new ArrayDeque<>();
Deque<RecordBatch> previous = this.batches.putIfAbsent(tp, d);
if (previous == null)
return d;
else
return previous;
}这个创立的内存构造能够看到，是 一个变量 batches，它是一个 CopyOnWriteMap。这个数据结构之前咱们组件图初步剖析过。再联合这段代码，不难理解它的脉络：
这个 map 次要依据 Topic 分区信息作为 key，value 是一个队列外围数据结构是 RecordBatch，因为是第一次给某个 topic 分区发送的音讯，value 为空，须要初始化队列，否则阐明已经给这个 topic 的分区发送给数据，value 非空，间接返回之前的队列。
因为咱们这里是第一次向 test-topic 发送音讯，所以能够失去下图的数据结构：
之后执行了一段加锁逻辑，之前提到，tryAppend 应该是将音讯放入内存中。然而因为队列是刚创立的，deque.peekLast(); 必定是空，所以这段加锁的代码不会执行。
synchronized (dq) {if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
RecordAppendResult appendResult = tryAppend(timestamp, key, value, callback, dq);
if (appendResult != null)
return appendResult;
}
private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, Deque<RecordBatch> deque){RecordBatch last = deque.peekLast();
if (last != null) {FutureRecordMetadata future = last.tryAppend(timestamp, key, value, callback, time.milliseconds());
if (future == null)
last.records.close();
else
return new RecordAppendResult(future, deque.size() > 1 || last.records.isFull(), false);
}
return null;
}然而到这里你会发现代码一个显著的特点，应用了 synchronized 加锁和线程平安的内存构造 CopyOnWriteMap，这些都是显著线程平安的管制。
为什么呢？因为同一个 Producer 能够应用多线程进行发送音讯，必然要思考线程平安的很多货色。
为什么选用 CopyOnWriteMap, 而不必 ConcurrentHashMap 呢？你能够思考下。（这里给个提醒，JDK 成长记提到过，CopyOnWriteMap 它的底层是写时复制，适宜读多写少的场景）
synchronized 加锁代码块应用了，分段加锁，并没有暴力的在办法上加 synchronized。这也是一个应用亮点。
写在结尾的话
到这里，你会发现在中间件会大量的见到并发包下的组件的应用，工作中你用到可能都是百里挑一，这些组件的应用是咱们钻研中间件源码值得学习的一点。
你肯定要多思考为什么，不要停留在是什么，怎么用上，这个思维须要刻意训练，心愿你能够缓缓养成。
好了，明天的内容就到这里，之前有同学反馈，每一节的只是太过于干了，实实在在的干货！看起来有时候会比拟吃力，所以之后的章节尽量会防止上万字的大章节，会管制在 6000 字左右。
另外，除了成长记外，我偶然也会分享我本人的故事和行业中遇见的事件，心愿大家从我的经验中能够有另一番成长和播种，比方我是如何学习和晋升技术的？我是如何画图的？我如何做技术分享的等等。
本文由博客一文多发平台 OpenWrite 公布！