在Kafka原始碼分析－序列2中，我們提到了整個Producer client的架構圖，如下所示：

其它幾個元件我們在前面都講過了，今天講述最後一個元件RecordAccumulator.

Batch傳送

在以前的kafka client中，每條訊息稱為 “Message”，而在Java版client中，稱之為”Record”，同時又因為有批量傳送累積功能，所以稱之為RecordAccumulator.

RecordAccumulator最大的一個特性就是batch訊息，扔到佇列中的多個訊息，可能組成一個RecordBatch，然後由Sender一次性發送出去。

每個TopicPartition一個佇列

下面是RecordAccumulator的內部結構，可以看到，每個TopicPartition對應一個訊息佇列，只有同一個TopicPartition的訊息，才可能被batch。

public final class RecordAccumulator {
    private final ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches;

   ...
}

batch的策略

那什麼時候，訊息會被batch，什麼時候不會呢？下面從KafkaProducer的send方法看起：

//KafkaProducer
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        try
 
 {
            // first make sure the metadata for the topic is available
            long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);

            ...

            RecordAccumulator.RecordAppendResult result = accumulator.append(tp, serializedKey, serializedValue, callback, remainingWaitMs);   //核心函式：把訊息放入佇列

            if
 
 (result.batchIsFull || result.newBatchCreated) {
                log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
                this.sender.wakeup();
            }
            return result.future;

從上面程式碼可以看到，batch邏輯，都在accumulator.append函式裡面：

    public RecordAppendResult append(TopicPartition tp, byte[] key, byte[] value, Callback callback, long maxTimeToBlock) throws InterruptedException {
        appendsInProgress.incrementAndGet();
        try {
            if (closed)
                throw new IllegalStateException("Cannot send after the producer is closed.");
            Deque<RecordBatch> dq = dequeFor(tp);  //找到該topicPartiton對應的訊息佇列
            synchronized (dq) {
                RecordBatch last = dq.peekLast(); //拿出佇列的最後1個元素
                if (last != null) {  
                    FutureRecordMetadata future = last.tryAppend(key, value, callback, time.milliseconds()); //最後一個元素, 即RecordBatch不為空，把該Record加入該RecordBatch
                    if (future != null)
                        return new RecordAppendResult(future, dq.size() > 1 || last.records.isFull(), false);
                }
            }

            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.");
                RecordBatch last = dq.peekLast();
                if (last != null) {
                    FutureRecordMetadata future = last.tryAppend(key, value, callback, time.milliseconds());
                    if (future != null) {
                        // Somebody else found us a batch, return the one we waited for! Hopefully this doesn't happen often...
                        free.deallocate(buffer);
                        return new RecordAppendResult(future, dq.size() > 1 || last.records.isFull(), false);
                    }
                }

                //佇列裡面沒有RecordBatch，建一個新的，然後把Record放進去
                MemoryRecords records = MemoryRecords.emptyRecords(buffer, compression, this.batchSize);
                RecordBatch batch = new RecordBatch(tp, records, time.milliseconds());
                FutureRecordMetadata future = Utils.notNull(batch.tryAppend(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();
        }
    }

    private Deque<RecordBatch> dequeFor(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;
    }

從上面程式碼我們可以看出Batch的策略：
1。如果是同步傳送，每次去佇列取，RecordBatch都會為空。這個時候，訊息就不會batch，一個Record形成一個RecordBatch

2。Producer 入隊速率 < Sender出隊速率 && lingerMs = 0 ，訊息也不會被batch

3。Producer 入隊速率 > Sender出對速率， 訊息會被batch

4。lingerMs > 0，這個時候Sender會等待，直到lingerMs > 0 或者 佇列滿了，或者超過了一個RecordBatch的最大值，就會發送。這個邏輯在RecordAccumulator的ready函式裡面。

    public ReadyCheckResult ready(Cluster cluster, long nowMs) {
        Set<Node> readyNodes = new HashSet<Node>();
        long nextReadyCheckDelayMs = Long.MAX_VALUE;
        boolean unknownLeadersExist = false;

        boolean exhausted = this.free.queued() > 0;
        for (Map.Entry<TopicPartition, Deque<RecordBatch>> entry : this.batches.entrySet()) {
            TopicPartition part = entry.getKey();
            Deque<RecordBatch> deque = entry.getValue();

            Node leader = cluster.leaderFor(part);
            if (leader == null) {
                unknownLeadersExist = true;
            } else if (!readyNodes.contains(leader)) {
                synchronized (deque) {
                    RecordBatch batch = deque.peekFirst();
                    if (batch != null) {
                        boolean backingOff = batch.attempts > 0 && batch.lastAttemptMs + retryBackoffMs > nowMs;
                        long waitedTimeMs = nowMs - batch.lastAttemptMs;
                        long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
                        long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
                        boolean full = deque.size() > 1 || batch.records.isFull();
                        boolean expired = waitedTimeMs >= timeToWaitMs;
                        boolean sendable = full || expired || exhausted || closed || flushInProgress();  //關鍵的一句話
                        if (sendable && !backingOff) {
                            readyNodes.add(leader);
                        } else {

                            nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
                        }
                    }
                }
            }
        }

        return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeadersExist);
    }

為什麼是Deque？

在上面我們看到，訊息佇列用的是一個“雙端佇列“，而不是普通的佇列。
一端生產，一端消費，用一個普通的佇列不就可以嗎，為什麼要“雙端“呢？

這其實是為了處理“傳送失敗，重試“的問題：當訊息傳送失敗，要重發的時候，需要把訊息優先放入佇列頭部重新發送，這就需要用到雙端佇列，在頭部，而不是尾部加入。

當然，即使如此，該訊息發出去的順序，還是和Producer放進去的順序不一致了。