private final ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;

/**
* Add a record to the accumulator, return the append result
* <p>
* The append result will contain the future metadata, and flag for whether the appended batch is full or a new batch is created
* <p>
*
* @param tp The topic/partition to which this record is being sent
* @param timestamp The timestamp of the record
* @param key The key for the record
* @param value The value for the record
* @param headers the Headers for the record
* @param callback The user-supplied callback to execute when the request is complete
* @param maxTimeToBlock The maximum time in milliseconds to block for buffer memory to be available
*/
public RecordAppendResult append(TopicPartition tp,
                               long timestamp,
                               byte[] key,
                               byte[] value,
                               Header[] headers,
                               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();
  ByteBuffer buffer = null;
  if (headers == null) headers = Record.EMPTY_HEADERS;
  try {
      // check if we have an in-progress batch
      // 其实就是一个putIfAbsent操作的方法，不展开分析
      Deque<ProducerBatch> dq = getOrCreateDeque(tp);
      // batches是线程安全的，但是Deque不是线程安全的
      // 已有在处理中的batch
      synchronized (dq) {
          if (closed)
              throw new IllegalStateException("Cannot send after the producer is closed.");
          RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq);
          if (appendResult != null)
              return appendResult;
      }
      // we don't have an in-progress record batch try to allocate a new batch
      // 创建一个新的ProducerBatch
      byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
      // 分配一个内存
      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());
      // 申请不到内存
      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.");
          // 再次尝试添加，因为分配内存的那段代码并不在synchronized块中
          // 有可能这时候其他线程已经创建好RecordBatch了，finally会把分配好的内存还回去
          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;
          }
          // 创建ProducerBatch
          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是一个Set集合，存放不完整的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);
      }
  } finally {
      // 释放要还的内存
      if (buffer != null)
          free.deallocate(buffer);
      appendsInProgress.decrementAndGet();
  }
}

/**
 *  Try to append to a ProducerBatch.
 *
 *  If it is full, we return null and a new batch is created. We also close the batch for record appends to free up
 *  resources like compression buffers. The batch will be fully closed (ie. the record batch headers will be written
 *  and memory records built) in one of the following cases (whichever comes first): right before send,
 *  if it is expired, or when the producer is closed.
 */
private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, Deque<ProducerBatch> deque) {
    // 获取最新加入的ProducerBatch
    ProducerBatch last = deque.peekLast();
    if (last != null) {
        FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, time.milliseconds());
        if (future == null)
            last.closeForRecordAppends();
        else
            // 记录添加结果类包含future、batch是否已满的标记、是否是新batch创建的标记
            return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false);
    }
    // 如果这个Deque没有ProducerBatch元素，或者已经满了不足以加入本条消息则返回null
    return null;
}