Kafka源码分析（三）高吞吐核心——RecordAccumul

Kafka为什么会有这么高吞吐？

Kafka的发送逻辑和TCP的像极了，当客户端调用了producer.send(msg)后，Kafka的主线程并不会着急直接调用网络底层将消息发送给Kafka Broker，而是将消息放入一个叫RecordAccumulator的数据结构中。

RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
    serializedValue, headers, interceptCallback, remainingWaitMs);

其实放入RecordAccumulator中只是第一步，接下去真实的发送逻辑甚至不在当前的主线程中，所以发送逻辑整体是以异步调用的方式来组织的。当消息真正被网络层发送并且得到Broker的成功反馈后，是通过Future的形式来通知回调，所以为了不丢失异步链路，在放入RecordAccumulator后，有个RecordAppendResult的返回值。

回过来再看下RecordAccumulator这个数据结构。

如下图所示，RecordAccumulator整体是一个ConcurrentMap<TopicPartition, Deque<ProducerBatch>>混合数据机构，Key就是TopicPartition，Value是一个双向队列Deque，队列的成员是一个个ProducerBatch。

RecordAccumulator

举个栗子，如果是发送TopicPartition(topic1:0)的消息，逻辑可以简述为，首先去找TopicPartition(topic1:0)这个Key所对应的那个Deque队列（如果没有则创建一个），然后从Deque中拿到最后一个ProducerBatch对象，最后将消息放入最后一个ProducerBatch中。

private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers,
                                        Callback callback, Deque<ProducerBatch> deque) {
    ProducerBatch last = deque.peekLast();
    if (last != null) {
        FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, time.milliseconds());
        if (future == null)
            last.closeForRecordAppends();
        else
            return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false);
    }
    return null;
}

可见ProducerBatch也是一个容器型数据结构，从下面的代码可以看出，消息的数据是按顺序放入（MemoryRecordsBuilder recordsBuilder）中，消息的事件回调future是按顺序放入（List<Thunk> thunks）中。

public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long now) {
    if (!recordsBuilder.hasRoomFor(timestamp, key, value, headers)) {
        return null;
    } else {
        Long checksum = this.recordsBuilder.append(timestamp, key, value, headers);
        this.maxRecordSize = Math.max(this.maxRecordSize, AbstractRecords.estimateSizeInBytesUpperBound(magic(),
                recordsBuilder.compressionType(), key, value, headers));
        this.lastAppendTime = now;
        FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount,
                                                                timestamp, checksum,
                                                                key == null ? -1 : key.length,
                                                                value == null ? -1 : value.length);
        // we have to keep every future returned to the users in case the batch needs to be
        // split to several new batches and resent.
        thunks.add(new Thunk(callback, future));
        this.recordCount++;
        return future;
    }
}

至此，放入RecordAccumulator的过程算是讲完了，下一篇聊下从RecordAccumulator拿出来。

在结束这篇前，有几点注意下，Map是Concurrent系的，所以在TopicPartition级别是可以安全并发put、get、remove它的Deque。但是当涉及到的是同一个TopicPartition时，操纵的其实是同一个Deque，而Deque不是一个并发安全的集合，所以在对某一个具体的Deque进行增删改时，需要使用锁。

Deque<ProducerBatch> dq = getOrCreateDeque(tp);

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");

    RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
    if (appendResult != null) {
        // Somebody else found us a batch, return the one we waited for! Hopefully this doesn't happen often...
        return appendResult;
    }

    MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic);
    ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, time.milliseconds());
    FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds()));

    dq.addLast(batch);
    incomplete.add(batch);

    // Don't deallocate this buffer in the finally block as it's being used in the record batch
    buffer = null;

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