import java.nio.ByteBuffer;
public class MemoryRecords extends AbstractRecords {
  public static MemoryRecordsBuilder builder(..){
        // 重載builder 
        return builder(...);
  }
    
  public static MemoryRecordsBuilder builder(
    ByteBuffer buffer,
    // 消息版本
    byte magic,
    // 消息壓縮類型
    CompressionType compressionType,
    // 時(shí)間戳
    TimestampType timestampType,
    // 基本位移
    long baseOffset,
    // 日志追加時(shí)間
    long logAppendTime,
    // 生產(chǎn)者id
    long producerId,
    // 生產(chǎn)者版本
    short producerEpoch,
    // 批次序列號(hào)
    int baseSequence,
    boolean isTransactional,
    // 是否是控制類的批次
    boolean isControlBatch,
    // 分區(qū)leader的版本
    int partitionLeaderEpoch) {
        // 初始化MemoryRecordsBuilder類
        return new MemoryRecordsBuilder(...);
  }
}

public class MemoryRecordsBuilder implements AutoCloseable {
    // 寫操作關(guān)閉的輸出流
    private static final DataOutputStream CLOSED_STREAM = new DataOutputStream(new OutputStream() {
        // 當(dāng)向某個(gè)ByteBuffer關(guān)閉輸出流寫數(shù)據(jù)時(shí)拋異常
        public void write(int b) {
            throw new ...;
        }
    });
    // 日志時(shí)間
    private final TimestampType timestampType;
    // 消息壓縮類型
    private final CompressionType compressionType;
    // kafka對(duì)OutputStream接口的實(shí)現(xiàn)類，對(duì)ByteBuffer實(shí)現(xiàn)了自動(dòng)擴(kuò)容功能
    private final ByteBufferOutputStream bufferStream;
    // 消息的版本
    private final byte magic;
    // ByteBuffer的最初始位置
    private final int initialPosition;
    // 基本位移
    private final long baseOffset;
    // 消息追加的時(shí)間
    private final long logAppendTime;
    // 是否是控制類的批次
    private final boolean isControlBatch;
    // 分區(qū)leader的版本
    private final int partitionLeaderEpoch;
    // 寫入上限
    private final int writeLimit;
    // batch頭大小字節(jié)數(shù)
    private final int batchHeaderSizeInBytes;
    // 評(píng)估壓縮率
    private float estimatedCompressionRatio = 1.0F;
    // 對(duì)bufferStream添加壓縮功能
    private DataOutputStream appendStream;
    // 是否是事務(wù)批次
    private boolean isTransactional;
    // 生產(chǎn)者id
    private long producerId;
    // 生產(chǎn)者版本
    private short producerEpoch;
    // 批次序列號(hào)
    private int baseSequence;
    // 壓縮前要寫入的消息體大小字節(jié)數(shù)
    private int uncompressedRecordsSizeInBytes = 0; 
    // 壓縮前寫入的記錄數(shù)（不包括頭）
    private int numRecords = 0;
    // 實(shí)際壓縮率
    private float actualCompressionRatio = 1;
    // 最大時(shí)間戳
    private long maxTimestamp = RecordBatch.NO_TIMESTAMP;
    // 最大時(shí)間戳偏移量
    private long offsetOfMaxTimestamp = -1;
    // 最后的偏移量
    private Long lastOffset = null;
    // 第一次追加消息的時(shí)間戳
    private Long firstTimestamp = null;
    // 真正保存消息的地方
    private MemoryRecords builtRecords;

public MemoryRecordsBuilder(ByteBuffer buffer,...) {
    // 將MemoryRecordsBuilder關(guān)聯(lián)的ByteBuffer封裝成ByteBufferOutputStream流
    this(new ByteBufferOutputStream(buffer), ...);
}

// 構(gòu)造方法
public MemoryRecordsBuilder(
    ByteBufferOutputStream bufferStream,
    ...
    int writeLimit) {
        ....
        // 初始位置
        this.initialPosition = bufferStream.position();
        // 1. 根據(jù)不同消息版本計(jì)算批次Batch頭的長(zhǎng)度
        this.batchHeaderSizeInBytes = AbstractRecords.recordBatchHeaderSizeInBytes(magic, compressionType);
        // 2. 調(diào)整對(duì)應(yīng)的position
        bufferStream.position(initialPosition + batchHeaderSizeInBytes);
        this.bufferStream = bufferStream;
        // 3. 在bufferStream流外層套一層壓縮流，再套一層DataOutputStream流
        this.appendStream = new DataOutputStream(compressionType.wrapForOutput(this.bufferStream, magic));
    }
}

// 追加新記錄
public Long append(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) {
   return appendWithOffset(nextSequentialOffset(), timestamp, key, value, headers);
}

// 計(jì)算下一個(gè)連續(xù)偏移量
private long nextSequentialOffset() {
  // lastOffset用來記錄當(dāng)前寫入Record的offset，每次當(dāng)有新Record寫入時(shí)，都會(huì)遞增它。
  return lastOffset == null ? baseOffset : lastOffset + 1;
}

// 根據(jù)偏移量追加消息
private Long appendWithOffset(
  long offset,
  boolean isControlRecord, 
  long timestamp, 
  ByteBuffer key,
  ByteBuffer value, 
  Header[] headers) {
    try {
        // 檢查isControl標(biāo)志是否一致
        if (isControlRecord != isControlBatch)
            throw new ...;
        // 保證offset是遞增的
        if (lastOffset != null && offset <= lastOffset)
            throw new ...;
        // 檢查時(shí)間戳      
        if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP)
           throw new ...;
        // 只有V2版本才有header
        if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0)
           throw new ...;
        // 更新firstTimestamp
        if (firstTimestamp == null)
           firstTimestamp = timestamp;
        // V2版本消息寫入
        if (magic > RecordBatch.MAGIC_VALUE_V1)         {
            appendDefaultRecord(offset, timestamp, key, value, headers);
            return null;
        } else {
            //V0、V1 版本消息寫入(此處不進(jìn)行剖析)
            return appendLegacyRecord(offset, timestamp, key, value, magic);
        }
    } catch (IOException e) {
        // 拋異常
    }
}

private void appendDefaultRecord(
  long offset, 
  long timestamp, 
  ByteBuffer key, 
  ByteBuffer value,
  Header[] headers) throws IOException {
    // 1. 檢查appendStream狀態(tài)是否可以寫
    ensureOpenForRecordAppend();
    // 2. 計(jì)算寫入多少偏移量
    int offsetDelta = (int) (offset - baseOffset);
    // 3.計(jì)算本次寫與第一次寫之間時(shí)間差
    long timestampDelta = timestamp - firstTimestamp;
    // 4.使用DefaultRecord.writeTo()方法會(huì)按照V2 版本格式寫入appendStream流中，并返回壓縮前的消息大小
    int sizeInBytes = DefaultRecord.writeTo(appendStream, offsetDelta, timestampDelta, key, value, headers);
    // 5. 消息寫入成功后更新RecordBatch的元信息
    recordWritten(offset, timestamp, sizeInBytes);
}

// 判斷appendStream狀態(tài)是否為CLOSED_STREAM 
private void ensureOpenForRecordAppend() {
    if (appendStream == CLOSED_STREAM)
        throw new ...;
}

// 消息寫入成功后更新RecordBatch的元信息
private void recordWritten(long offset, long timestamp, int size) {
  ....
  // 壓縮前寫入的記錄數(shù) + 1
  numRecords += 1;
  // 壓縮前要寫入的消息體大小字節(jié)數(shù) + size
  uncompressedRecordsSizeInBytes += size;
  // 最后的偏移量 + offset
  lastOffset = offset;
  if (magic > RecordBatch.MAGIC_VALUE_V0 && timestamp > maxTimestamp) {
      // 賦值最大時(shí)間戳
      maxTimestamp = timestamp;
      // 賦值最大時(shí)間戳偏移量
      offsetOfMaxTimestamp = offset;
  }
}

public boolean hasRoomFor(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) {
    // 檢查兩個(gè)狀態(tài)
    // (1)appendStream流狀態(tài)
    // (2)當(dāng)前已經(jīng)寫入的預(yù)估字節(jié)數(shù)是否超過了writeLimit寫入上限
    if (isFull())
        return false;
    // 每個(gè)RecordBatch至少可以寫入一個(gè)Record，此時(shí)如果一個(gè)Record都沒有，則可以繼續(xù)寫入
    if (numRecords == 0)
        return true;
    final int recordSize;
    if (magic < RecordBatch.MAGIC_VALUE_V2) {
        // 預(yù)估V0、V1舊版本的Record大小
        recordSize = Records.LOG_OVERHEAD + LegacyRecord.recordSize(magic, key, value);
    } else {
        // 預(yù)估V2版本寫入的Record大小
        int nextOffsetDelta = lastOffset == null ? 0 : (int) (lastOffset - baseOffset + 1);
        ...
        recordSize = DefaultRecord.sizeInBytes(nextOffsetDelta, timestampDelta, key, value, headers);
    }

    // 已寫入字節(jié)數(shù) + 本次寫入Record的預(yù)估字節(jié)數(shù)不能超過writeLimit寫入上限
    return this.writeLimit >= estimatedBytesWritten() + recordSize;
}

public boolean isFull() {
      return appendStream == CLOSED_STREAM || 
      (this.numRecords > 0 && this.writeLimit <= estimatedBytesWritten());
}

public class ByteBufferOutputStream extends OutputStream {
   // 擴(kuò)容因子1.1倍
   private static final float REALLOCATION_FACTOR = 1.1f;
   // 初始容量
   private final int initialCapacity;
   // 初始位置
   private final int initialPosition;
   // 計(jì)算是否需要擴(kuò)容
   public void ensureRemaining(int remainingBytesRequired) {
     // 當(dāng)寫入字節(jié)數(shù)大于buffer當(dāng)前剩余字節(jié)數(shù)就開啟擴(kuò)容
     if (remainingBytesRequired > buffer.remaining())
     expandBuffer(remainingBytesRequired);
  }
  
  // 擴(kuò)容
  private void expandBuffer(int remainingRequired) {
    // 1. 評(píng)估需要多少空間
    int expandSize = Math.max((int) (buffer.limit() * REALLOCATION_FACTOR), buffer.position() + remainingRequired);
    // 2. 申請(qǐng)新的ByteBuffer
    ByteBuffer temp = ByteBuffer.allocate(expandSize);
    // 3. 獲取寫入上限
    int limit = limit();
    // 4. 寫狀態(tài)轉(zhuǎn)換為讀狀態(tài)
    buffer.flip();
    // 5. 將buffer讀到新申請(qǐng)的temp里
    temp.put(buffer);
    // 6. 修改寫模式的limit上限
    buffer.limit(limit);
    // 7. 更新原來的buffer的position，防止被重復(fù)消費(fèi)
    buffer.position(initialPosition);
    // 8. 將引用指向新申請(qǐng)的ByteBuffer
    buffer = temp;
  }
}

public final class ProducerBatch {
    // 批次最終狀態(tài)
    private enum FinalState { ABORTED, FAILED, SUCCEEDED }
    // 批次創(chuàng)建時(shí)間  
    final long createdMs;
    // 批次對(duì)應(yīng)的主題分區(qū)
    final TopicPartition topicPartition;
    // 請(qǐng)求結(jié)果的future
    final ProduceRequestResult produceFuture;
    // 用來存儲(chǔ)消息的callback和響應(yīng)數(shù)據(jù)
    private final List<Thunk> thunks = new ArrayList<>();
    // 封裝MemoryRecords對(duì)象，用來存儲(chǔ)消息的ByteBuffer
    private final MemoryRecordsBuilder recordsBuilder;
    // batch的失敗重試次數(shù)
    private final AtomicInteger attempts = new AtomicInteger(0);
    // 是否是被分裂的批次
    private final boolean isSplitBatch;
    // ProducerBatch的最終狀態(tài)
    private final AtomicReference<FinalState> finalState = new AtomicReference<>(null);
    // Record個(gè)數(shù)
    int recordCount;
    // 最大Record字節(jié)數(shù)
    int maxRecordSize;
    // 最后一次失敗重試發(fā)送的時(shí)間戳
    private long lastAttemptMs;
    // 最后一次向該P(yáng)roducerBatch追加Record的時(shí)間戳
    private long lastAppendTime;
    // Sender子線程拉取批次的時(shí)間
    private long drainedMs;
    // 是否正在重試過，如果ProducerBatch中的數(shù)據(jù)發(fā)送失敗，則會(huì)重新嘗試發(fā)送
    private boolean retry;
}

// 構(gòu)造函數(shù)
public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long createdMs, boolean isSplitBatch) {
    ...
    // 請(qǐng)求結(jié)果的future
    this.produceFuture = new ProduceRequestResult(topicPartition);
    ...
}

public class ProduceRequestResult {
    // 通過一個(gè)count為1的CountDownLatch對(duì)象間接地實(shí)現(xiàn)了Future的功能。
    private final CountDownLatch latch = new CountDownLatch(1);
    private final TopicPartition topicPartition;
    // 用來記錄broker端關(guān)聯(lián)ProducerBatch中第一條Record分配的offset值
    // 這樣每個(gè)Record的真實(shí)offset就可以根據(jù)自身在ProducerBatch的位置計(jì)算出來了(baseOffset + relativeOffset)
    private volatile Long baseOffset = null;
    
    // 構(gòu)造函數(shù)
    public ProduceRequestResult(TopicPartition topicPartition) {
        this.topicPartition = topicPartition;
    }
    // 當(dāng)?shù)鹊巾憫?yīng)會(huì)會(huì)調(diào)該函數(shù)喚醒阻塞的主線程
    public void done() {
        if (baseOffset == null)
            throw new ...;
        this.latch.countDown();
    }
    // 調(diào)用await()方法的線程會(huì)被掛起，它會(huì)等待直到count值為0才繼續(xù)執(zhí)行
    public void await() throws InterruptedException {
        latch.await();
    }
    // 和await()類似，只不過等待一定的時(shí)間后count值還沒變?yōu)?的話就會(huì)繼續(xù)執(zhí)行
    public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
        return latch.await(timeout, unit);
    }
}

public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long now) {
    // 1.檢查MemoryRecordsBuilder是否還有空間寫入
    if (!recordsBuilder.hasRoomFor(timestamp, key, value, headers)) {
        return null;
    } else {
        // 2.調(diào)用append()方法寫入Record
        Long checksum = this.recordsBuilder.append(timestamp, key, value, headers);
        // 3. 更新最大Record字節(jié)數(shù)
        this.maxRecordSize = Math.max(this.maxRecordSize, AbstractRecords.estimateSizeInBytesUpperBound(magic(),recordsBuilder.compressionType(), key, value, headers));
        ...
        // 4.構(gòu)建FutureRecordMetadata對(duì)象
        FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount,timestamp, checksum,key == null ? -1 : key.length,value == null ? -1 : value.length, Time.SYSTEM);
        // 5. 將Callback和FutureRecordMetadata記錄到thunks集合中
        thunks.add(new Thunk(callback, future));
        // 6. 更新Record記錄數(shù)
        this.recordCount++;
        // 7. 返回FutureRecordMetadata
        return future;
    }
}

public RecordMetadata get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
    ...
    // 依賴ProduceRequestResult的CountDown來實(shí)現(xiàn)阻塞等待
    boolean occurred = this.result.await(timeout, unit);
    ...
    // 調(diào)用value()方法返回RecordMetadata對(duì)象
    return valueOrError();
}
    
RecordMetadata valueOrError() throws ExecutionException {
    ...
    return value();
}

RecordMetadata value() {
    ...
     // 將 partition、baseOffset、relativeOffset、時(shí)間戳（LogAppendTime | CreateTimeStamp）等信息封裝成 RecordMetadata 對(duì)象返回
    return new RecordMetadata(
      result.topicPartition(), 
      ...);
}

private long timestamp() {
    return result.hasLogAppendTime() ? result.logAppendTime() : createTimestamp;
}

public boolean done(long baseOffset, long logAppendTime, RuntimeException exception) {
    // 1.根據(jù)exception決定本次ProducerBatch發(fā)送的最終狀態(tài)
    final FinalState tryFinalState = (exception == null) ? FinalState.SUCCEEDED : FinalState.FAILED;
    ....
    // 2.通過CAS操作更新finalState狀態(tài)，只有第一次更新的時(shí)候，才會(huì)觸發(fā)completeFutureAndFireCallbacks()方法
    if (this.finalState.compareAndSet(null, tryFinalState)) {
        // 3.執(zhí)行回調(diào)
        completeFutureAndFireCallbacks(baseOffset, logAppendTime, exception);
        return true;
    }
    ....
    return false;
}

private void completeFutureAndFireCallbacks(long baseOffset, long logAppendTime, RuntimeException exception) {
  // 1.更新ProduceRequestResult中的相關(guān)字段
  produceFuture.set(baseOffset, logAppendTime, exception);

  // 2.遍歷thunks集合，觸發(fā)每個(gè)Record的Callback回調(diào)
  for (Thunk thunk : thunks) {
      try {
          if (exception == null) {
           // 3.獲取消息元數(shù)據(jù)
           RecordMetadata metadata = thunk.future.value();
           if (thunk.callback != null)
             //4.調(diào)用回調(diào)方法
             thunk.callback.onCompletion(metadata, null);
          } else {
              if (thunk.callback != null)
                  // 4.調(diào)用回調(diào)方法
                  thunk.callback.onCompletion(null, exception);
          }
      } 
      ....
  }
  // 4.調(diào)用底層 CountDownLatch.countDown()方法，阻塞在其上的主線程。
  produceFuture.done();
}

public class BufferPool {
  // 整個(gè)BufferPool總內(nèi)存大小 默認(rèn)32M
  private final long totalMemory;
  // 當(dāng)前BufferPool管理的單個(gè)ByteBuffer大小，16k 
  private final int poolableSize;
  // 因?yàn)橛卸嗑€程并發(fā)分配和回收ByteBuffer，用鎖控制并發(fā)，保證線程安全。
  private final ReentrantLock lock;
  // 對(duì)應(yīng)一個(gè)ArrayDeque<ByteBuffer> 隊(duì)列，其中緩存了固定大小的 ByteBuffer 對(duì)象
  private final Deque<ByteBuffer> free;
  // 此隊(duì)列記錄因申請(qǐng)不到足夠空間而阻塞的線程對(duì)應(yīng)的Condition 對(duì)象
  private final Deque<Condition> waiters;
  // 非池化可用的內(nèi)存即totalMemory減去free列表中的全部ByteBuffer的大小
  private long nonPooledAvailableMemory;
  // 構(gòu)造函數(shù)
  public BufferPool(long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName) {
    ...
    // 總的內(nèi)存
    this.totalMemory = memory;
    // 默認(rèn)的池外內(nèi)存，就是總的內(nèi)存
    this.nonPooledAvailableMemory = memory;
  }
}

// 分配指定空間的緩存，如果緩沖區(qū)中沒有足夠的空閑空間，那么會(huì)阻塞線程，直到超時(shí)或得到足夠空間
public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {
  // 1.判斷申請(qǐng)的內(nèi)存是否大于總內(nèi)存
  if (size > this.totalMemory)
      throw new IllegalArgumentException("Attempt to allocate " + size + " bytes, but there is a hard limit of "+ this.totalMemory + " on memory allocations.");
  // 初始化buffer
  ByteBuffer buffer = null;
  // 2.加鎖，保證線程安全。
  this.lock.lock();
  // 如果當(dāng)前BufferPool處于關(guān)閉狀態(tài)，則直接拋出異常
  if (this.closed) {
      this.lock.unlock();
      throw new KafkaException("Producer closed while allocating memory");
  }
  ....
  try {
      // 3.申請(qǐng)內(nèi)存大小恰好為16k 且free緩存池不為空
      if (size == poolableSize && !this.free.isEmpty())
      // 從free隊(duì)列取出一個(gè)ByteBuffer
      return this.free.pollFirst();
      
      // 對(duì)于申請(qǐng)內(nèi)存大小非16k情況
      // 先計(jì)算free緩存池總空間大小，判斷是否足夠
      int freeListSize = freeSize() * this.poolableSize;
      // 4.當(dāng)前BufferPool能夠釋放出申請(qǐng)內(nèi)存大小的空間
      if (this.nonPooledAvailableMemory + freeListSize >= size) {
          // 5.如果size大于非池化可用內(nèi)存大小，就循環(huán)從free緩存池里釋放出來空閑Bytebuffer補(bǔ)充到nonPooledAvailableMemory中，直到滿足size大小為止。
          freeUp(size);
          // 釋放非池化可用內(nèi)存大小
          this.nonPooledAvailableMemory -= size;
      } else {
          // 如果當(dāng)前BufferPool不夠提供申請(qǐng)內(nèi)存大小，則需要阻塞當(dāng)前線程
          // 累計(jì)已經(jīng)釋放的內(nèi)存
          int accumulated = 0;
          // 創(chuàng)建對(duì)應(yīng)的Condition，阻塞自己等待別的線程釋放內(nèi)存
          Condition moreMemory = this.lock.newCondition();
          try {
              // 計(jì)算當(dāng)前線程最大阻塞時(shí)長(zhǎng)
              long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);
              // 把自己添加到等待隊(duì)列中末尾，保持公平性，先來的先獲取內(nèi)存，防止饑餓
              this.waiters.addLast(moreMemory);
              // 循環(huán)等待直到分配成功或超時(shí)
              while (accumulated < size) {
                  ....
                  try {
                    // 當(dāng)前線程阻塞等待，返回結(jié)果為false則表示阻塞超時(shí)
                   waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
                  } finally {
                      ....
                  }
                  ....   
                  // 申請(qǐng)內(nèi)存大小是16k，且free緩存池有了空閑的ByteBuffer
                  if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {
                    // 從free隊(duì)列取出一個(gè)ByteBuffer
                    buffer = this.free.pollFirst();
                    // 計(jì)算累加器
                    accumulated = size;
                  } else {
                      // 釋放空間給非池化可用內(nèi)存，并繼續(xù)等待空閑空間，如果分配多了只取夠size的空間
                      freeUp(size - accumulated);
                      int got = (int) Math.min(size - accumulated, this.nonPooledAvailableMemory);
                      // 釋放非池化可用內(nèi)存大小
                      this.nonPooledAvailableMemory -= got;
                      // 累計(jì)分配了多少空間
                      accumulated += got;
                  }
              }
              accumulated = 0;
          } finally {
              // 如果循環(huán)有異常，將已釋放的空間歸還給非池化可用內(nèi)存
              this.nonPooledAvailableMemory += accumulated;
              //把自己從等待隊(duì)列中移除并結(jié)束
              this.waiters.remove(moreMemory);
          }
      }
  } finally {
     // 當(dāng)非池化可用內(nèi)存有內(nèi)存或free緩存池有空閑ByteBufer且等待隊(duì)列里有線程正在等待
      try {
          if (!(this.nonPooledAvailableMemory == 0 && this.free.isEmpty()) && !this.waiters.isEmpty())
           // 喚醒隊(duì)列里正在等待的線程
           this.waiters.peekFirst().signal();
      } finally {
          // 解鎖
          lock.unlock();
      }
  }
  // 說明空間足夠，并且有足夠空閑的了?？梢詧?zhí)行真正的分配空間了。
  if (buffer == null)
      // 沒有正好的buffer，從緩沖區(qū)外(JVM Heap)中直接分配內(nèi)存
      return safeAllocateByteBuffer(size);
  else
      // 直接復(fù)用free緩存池的ByteBuffer
      return buffer;
}

private ByteBuffer safeAllocateByteBuffer(int size) {
  boolean error = true;
  try {
      //分配空間
      ByteBuffer buffer = allocateByteBuffer(size);
      error = false;
      //返回buffer
      return buffer;
  } finally {
    if (error) {
        //分配失敗了, 加鎖，操作內(nèi)存pool
        this.lock.lock();
        try {
            //歸還空間給非池化可用內(nèi)存
            this.nonPooledAvailableMemory += size;
            if (!this.waiters.isEmpty())
                //有其他在等待的線程的話，喚醒其他線程
                this.waiters.peekFirst().signal();
        } finally {
            // 加鎖不忘解鎖
            this.lock.unlock();
        }
    }
  }
}

protected ByteBuffer allocateByteBuffer(int size) {
    // 從JVM Heap中分配空間
    return ByteBuffer.allocate(size);
}

// 不斷從free隊(duì)列中釋放空閑的ByteBuffer來補(bǔ)充非池化可用內(nèi)存
private void freeUp(int size) {
    while (!this.free.isEmpty() && this.nonPooledAvailableMemory < size)
        this.nonPooledAvailableMemory += this.free.pollLast().capacity();
}

public void deallocate(ByteBuffer buffer, int size) {
    // 1.加鎖，保證線程安全。
    lock.lock();
    try {
    // 2.如果待釋放的size大小為16k，則直接放入free隊(duì)列中
        if (size == this.poolableSize && size == buffer.capacity()) {
            // 清空buffer
            buffer.clear();
            // 釋放buffer到free隊(duì)列里
            this.free.add(buffer);
        } else {
            //如果非16k，則由JVM GC來回收ByteBuffer并增加非池化可用內(nèi)存
            this.nonPooledAvailableMemory += size;
        }
        // 3.喚醒waiters中的第一個(gè)阻塞線程
        Condition moreMem = this.waiters.peekFirst();
        if (moreMem != null)
            moreMem.signal();
    } finally {
        // 釋放鎖
        lock.unlock();
    }
}

public class CopyOnWriteMap<K, V> implements ConcurrentMap<K, V> {
    // volatile Map
    private volatile Map<K, V> map;
    // 構(gòu)造函數(shù)
    public CopyOnWriteMap() {
        this.map = Collections.emptyMap();
    }

// 獲取集合中隊(duì)列
public V get(Object k) {
    return map.get(k);
}

public synchronized V putIfAbsent(K k, V v) {
    if (!containsKey(k))
        return put(k, v);
    else
        return get(k);
}

// 判斷隊(duì)列是否存在
public boolean containsKey(Object k) {
    return map.containsKey(k);
}

public synchronized V put(K k, V v) {
    Map<K, V> copy = new HashMap<K, V>(this.map);
    V prev = copy.put(k, v);
    this.map = Collections.unmodifiableMap(copy);
    return prev;
}