上次跟大家分享的文章「Kafka Producer 异步发送音讯竟然也会阻塞？」中提到了缓冲池，前面再通过一番浏览源码后，发现了这个缓冲池设计的很棒，被它的设计思维优雅到了，所以忍不住跟大家持续分享一波。
在新版的 Kafka Producer 中，设计了一个音讯缓冲池，在创立 Producer 时会默认创立一个大小为 32M 的缓冲池，也能够通过 buffer.memory 参数指定缓冲池的大小，同时缓冲池被切分成多个内存块，内存块的大小就是咱们创立 Producer 时传的 batch.size 大小，默认大小 16384，而每个 Batch 都会蕴含一个 batch.size 大小的内存块，音讯就是寄存在内存块当中。整个缓冲池的构造如下图所示：
客户端将音讯追加到对应主题分区的某个 Batch 中，如果 Batch 曾经满了，则会新建一个 Batch，同时向缓冲池（RecordAccumulator）申请一块大小为 batch.size 的内存块用于存储音讯。
当 Batch 的音讯被发到了 Broker 后，Kafka Producer 就会移除该 Batch，既然 Batch 持有某个内存块，那必然就会波及到 GC 问题，如下：
以上，频繁的申请内存，用完后就抛弃，必然导致频繁的 GC，造成重大的性能问题。那么，Kafka 是怎么做到防止频繁 GC 的呢？
后面说过了，缓冲池在设计逻辑下面被切分成一个个大小相等的内存块，当音讯发送结束，归还给缓冲池不就能够防止被回收了吗？
缓冲池的内存持有类是 BufferPool，咱们先来看下 BufferPool 都有哪些成员：
public class BufferPool {
// 总的内存大小
private final long totalMemory;
// 每个内存块大小，即 batch.size
private final int poolableSize;
// 申请、偿还内存的办法的同步锁
private final ReentrantLock lock;
// 闲暇的内存块
private final Deque<ByteBuffer> free;
// 须要期待闲暇内存块的事件
private final Deque<Condition> waiters;
/** Total available memory is the sum of nonPooledAvailableMemory and the number of byte buffers in free * poolableSize. */
// 缓冲池还未调配的闲暇内存，新申请的内存块就是从这里获取内存值
private long nonPooledAvailableMemory;
// ...
}从 BufferPool 的成员可看出，缓冲池实际上由一个个 ByteBuffer 组成的，BufferPool 持有这些内存块，并保留在成员 free 中，free 的总大小由 totalMemory 作限度，而 nonPooledAvailableMemory 则示意还剩下缓冲池还剩下多少内存还未被调配。
当 Batch 的音讯发送结束后，就会将它持有的内存块偿还到 free 中，以便前面的 Batch 申请内存块时不再创立新的 ByteBuffer，从 free 中取就能够了，从而防止了内存块被 JVM 回收的问题。
接下来跟大家一起剖析申请内存和偿还内存是如何实现的。
1、申请内存
申请内存的入口：
org.apache.kafka.clients.producer.internals.BufferPool#allocate
1）内存足够的状况
当用户申请申请内存时，如果发现 free 中有闲暇的内存，则间接从中取：
if (size == poolableSize && !this.free.isEmpty()){return this.free.pollFirst();
} 这里的 size 即申请的内存大小，它等于 Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));
即如果你的音讯大小小于 batchSize，则申请的内存大小为 batchSize，那么下面的逻辑就是如果申请的内存大小等于 batchSize 并且 free 不闲暇，则间接从 free 中获取。
咱们无妨想一下，为什么 Kafka 肯定要申请内存大小等于 batchSize，能力从 free 获取闲暇的内存块呢？
后面也说过，缓冲池的内存块大小是固定的，它等于 batchSize，如果申请的内存比 batchSize 还大，阐明一条音讯所须要寄存的内存空间比内存块的内存空间还要大，因而不满足需要，不满组需要怎么办呢？咱们接着往下剖析：
// now check if the request is immediately satisfiable with the
// memory on hand or if we need to block
int freeListSize = freeSize() * this.poolableSize;
if (this.nonPooledAvailableMemory + freeListSize >= size) {
// we have enough unallocated or pooled memory to immediately
// satisfy the request, but need to allocate the buffer
freeUp(size);
this.nonPooledAvailableMemory -= size;
}freeListSize：指的是 free 中曾经调配好并且曾经回收的闲暇内存块总大小；
nonPooledAvailableMemory：缓冲池还未调配的闲暇内存，新申请的内存块就是从这里获取内存值；
this.nonPooledAvailableMemory + freeListSize：即缓冲池中总的闲暇内存空间。
如果缓冲池的内存空间比申请内存大小要大，则调用 freeUp(size); 办法，接着将闲暇的内存大小减去申请的内存大小。
private void freeUp(int size) {while (!this.free.isEmpty() && this.nonPooledAvailableMemory < size)
this.nonPooledAvailableMemory += this.free.pollLast().capacity();
}freeUp 这个办法很乏味，它的思维是这样的：
如果未调配的内存大小比申请的内存还要小，那只能从已调配的内存列表 free 中将内存空间要回来，直到 nonPooledAvailableMemory 比申请内存大为止。
2）内存不足的状况
在我的「Kafka Producer 异步发送音讯竟然也会阻塞？」这篇文章当中也提到了，当缓冲池的内存块用完后，音讯追加调用将会被阻塞，直到有闲暇的内存块。
阻塞期待的逻辑是怎么实现的呢？
// we are out of memory and will have to block
int accumulated = 0;
Condition moreMemory = this.lock.newCondition();
try {long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);
this.waiters.addLast(moreMemory);
// loop over and over until we have a buffer or have reserved
// enough memory to allocate one
while (accumulated < size) {long startWaitNs = time.nanoseconds();
long timeNs;
boolean waitingTimeElapsed;
try {waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
} finally {long endWaitNs = time.nanoseconds();
timeNs = Math.max(0L, endWaitNs - startWaitNs);
recordWaitTime(timeNs);
}
if (waitingTimeElapsed) {throw new TimeoutException("Failed to allocate memory within the configured max blocking time" + maxTimeToBlockMs + "ms.");
}
remainingTimeToBlockNs -= timeNs;
// check if we can satisfy this request from the free list,
// otherwise allocate memory
if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {
// just grab a buffer from the free list
buffer = this.free.pollFirst();
accumulated = size;
} else {
// we'll need to allocate memory, but we may only get
// part of what we need on this iteration
freeUp(size - accumulated);
int got = (int) Math.min(size - accumulated, this.nonPooledAvailableMemory);
this.nonPooledAvailableMemory -= got;
accumulated += got;
}
}以上源码的大抵逻辑：
首先创立一个本次期待 Condition，并且把它增加到类型为 Deque 的 waiters 中（前面在偿还内存中会唤醒），while 循环不断收集闲暇的内存，直到内存比申请内存大时退出，在 while 循环过程中，调用 Condition#await 办法进行阻塞期待，偿还内存时会被唤醒，唤醒后会判断以后申请内存是否大于 batchSize，如果等与 batchSize 则间接将偿还的内存返回即可，如果以后申请的内存大于 大于 batchSize，则须要调用 freeUp 办法从 free 中开释闲暇的内存进去，而后进行累加，直到大于申请的内存为止。
2、偿还内存
申请内存的入口：
org.apache.kafka.clients.producer.internals.BufferPool#deallocate(java.nio.ByteBuffer, int)
public void deallocate(ByteBuffer buffer, int size) {lock.lock();
try {if (size == this.poolableSize && size == buffer.capacity()) {buffer.clear();
this.free.add(buffer);
} else {this.nonPooledAvailableMemory += size;}
Condition moreMem = this.waiters.peekFirst();
if (moreMem != null)
moreMem.signal();} finally {lock.unlock();
}
}偿还内存块的逻辑比较简单：
如果偿还的内存块大小等于 batchSize，则将其清空后增加到缓冲池的 free 中，行将其归还给缓冲池，防止了 JVM GC 回收该内存块。如果不等于呢？间接将内存大小累加到未调配并且闲暇的内存大小值中即可，内存就无需偿还了，期待 JVM GC 回收掉，最初唤醒正在期待闲暇内存的线程。
通过以上的源码剖析之后，给大家指出须要留神的一个问题，如果设置不当，会给 Producer 端带来重大的性能影响：
如果你的音讯大小比 batchSize 还要大，则不会从 free 中循环获取已调配好的内存块，而是从新创立一个新的 ByteBuffer，并且该 ByteBuffer 不会被偿还到缓冲池中（JVM GC 回收），如果此时 nonPooledAvailableMemory 比音讯体还要小，还会将 free 中闲暇的内存块销毁（JVM GC 回收），以便缓冲池中有足够的内存空间提供给用户申请，这些动作都会导致频繁 GC 的问题呈现。
因而，须要依据业务音讯的大小，适当调整 batch.size 的大小，防止频繁 GC。
作者简介
作者张乘辉，善于消息中间件技能，负责公司百万 TPS 级别 Kafka 集群的保护，作者保护的公号「后端进阶」不定期分享 Kafka、RocketMQ 系列不讲概念间接真刀真枪的实战总结以及细节上的源码剖析；同时作者也是阿里开源分布式事务框架 Seata Contributor，因而也会分享对于 Seata 的相干常识；当然公号也会分享 WEB 相干常识比方 Spring 全家桶等。内容不肯定八面玲珑，但肯定让你感触到作者对于技术的谋求是认真的！
公众号：后端进阶
技术博客：https://objcoding.com/
GitHub：https://github.com/objcoding/