Kafka生产者源码解析（二）——RecordAccumulator_kafka recordaccumulator

---

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

最后存入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 b