学习Kafka生产者的缓冲池设计_kafka缓存池

大家一定都了解Java的线程池，线程池有什么好处呢？如果没有线程池，我们每次创建线程都要新建一个线程，这样对CPU的消耗比较大。那么利用线程池我们可以对已经创建好的线程复用，线程就不用频繁创建和销毁了。

同样，我们的内存池也是这个原理，producerBatch需要空间存储消息的时候，就去缓存池申请一块内存，而不用频繁地创建和销毁内存，也就避免了频繁地GC。

BufferPool简介

下面的结构图简单说明了BufferPool的组成结构和处理缓存的流程：

整个BufferPool的大小默认为32M，内部内存区域分为两块：固定大小内存块集合free、非池化缓存nonPooledAvailableMemory。固定大小内存块默认大小为16k。当ProducerBatch向BufferPool申请一个大小为size的内存块时，BufferPool会根据size的大小判断由哪个内存区域分配内存块。同时，free和nonPooledAvailableMemory这两块区域的内存可以交换。

接下来，我们通过代码来学习Kafka底层提供的高效的内存池设计。

类BufferPool

重要字段如下：

public class BufferPool {
   

    static final String WAIT_TIME_SENSOR_NAME = "bufferpool-wait-time";

    private final long totalMemory;//默认32M
    private final int poolableSize;//池化大小16k
    private final ReentrantLock lock;//分配和回收时用的锁。
    private final Deque<ByteBuffer> free;//池化的内存
    private final Deque<Condition> waiters;//阻塞线程对应的Condition集合
    private long nonPooledAvailableMemory;//非池化可使用的内存
}
1
2
3
4
5
6
7
8
9
10
11
12

• totalMemory：整个BufferPool内存大小，默认是32M。
• poolableSize：池化缓存区一块内存块的大小，默认是16k。
• lock：类型是ReentrantLock。因为会有多线程并发和回收ByteBuffer，所以使用锁控制并发，保证了线程的安全。
• free：类型是Deque。缓存了指定大小的ByteBuffer对象。
• waiters：类型是Deque队列。因为会有申请不到足够内存的线程，线程为了等待其他线程释放内存而阻塞等待，对应的Condition对象会进入该队列。
• nonPooledAvailableMemory：非池化可使用的内存。

接下来，我再来介绍下重要的方法。

allocate()方法是向BufferPool申请ByteBuffer。

public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {
   
    //1.验证申请的内存是否大于总内存
    if (size > this.totalMemory)
        throw new IllegalArgumentException("Attempt to allocate " + size
                                           + " bytes, but there is a hard limit of "
                                           + this.totalMemory
                                           + " on memory allocations.");

    ByteBuffer buffer = null;
    //2.加锁，保证线程安全。
    this.lock.lock();

    if (this.closed) 
1
2
3
4
5
6
7
8
9
10
11
12
13