kafka生产者之内存池设计

一个消息被分区以后，消息就会被放到一个缓存里面，我们看一下里面具体的细节。默认缓存块的大小是 32M，这个缓存块里面有一个重要的数据结构：batches，这个数据结构是 key-value 的结果，key 就是消息主题的分区，value 是一个队列，里面存的是发送到对应分区的批次，Sender 线程就是把这些批次发送到服务端

01 生产者高级设计之自定义数据结构
生产者把批次信息用 batches 这个对象进行存储。
Kafka 这儿采取的方式是自定义了一个数据结构：CopyOnWriteMap。
1、他们存储的信息的是 key-value 的结构，key 是分区，value 是要存到这个分区的对应批次（批次可能有多个，所以用的是队列），故因为是 key-value 的数据结构，所以锁定用 Map 数据结构。
2、这个 Kafka 生产者面临的是一个高并发的场景，大量的消息会涌入这个这个数据结构，所以这个数据结构需要保证线程安全，这样我们就不能使用 HashMap 这样的数据结构了。
3、这个数据结构需要支持的是读多写少的场景。读多是因为每条消息过来都会根据 key 读取 value 的信息，假如有 1000 万条消息，那么就会读取 batches 对象 1000 万次。写少是因为，比如我们生产者发送数据需要往一个主题里面去发送数据，假设这个主题有 50 个分区，那么这个 batches 里面就需要写 50 个 key-value 数据就可以了（大家要搞清楚我们虽然要写 1000 万条数据，但是这 1000 万条是写入 queue 队列的 batch 里的，并不是直接写入 batches,所以就我们刚刚说的这个场景，batches 里只需要最多写 50 条数据就可以了）。
根据第二和第三个场景我们总结出来，Kafka 这儿需要一个能保证线程安全的，支持读多写少的 Map 数据结构。但是 Java 里面并没有提供出来的这样的一个数据，唯一跟这个需求比较接近的是 CopyOnWriteArrayList，但是偏偏它又不是 Map 结构，所以 Kafka 这儿模仿 CopyOnWriteArrayList 设计了 CopyOnWriteMap。采用了读写分离的思想解决了线程安全且支持读多写少等问题。
高效的数据结构保证了生产者的性能。Kafka 生产者往 batches 里插入数据的源码，生产者为了保证插入数据的高性能，采用了多线程，又为了线程安全，使用了分段加锁等多种手段

02 生产者高级设计之内存池设计
刚刚我们看到 batches 里面存储的是批次，批次默认的大小是 16K，整个缓存的大小是 32M，生产者每封装一个批次都需要去申请内存，正常情况下如果一个批次发送出去了以后，那么这 16K 的内存就等着 GC 来回收了。但是如果是这样的话，就可能会频繁的引发 FullGC，故而影响生产者的性能，所以在缓存里面设计了一个内存池（类似于我们平时用的数据库的连接池），一个 16K 的内存用完了以后，把数据清空，放入到内存池里，下个批次用的时候直接从里面获取就可以。这样大大的减少了 GC 的频率，保证了生产者的稳定和高效

以下基于代码解释：

Kafka为了提高吞吐量，Producer采用批量发送数据的模式。Producer可以通过配置设置整个batch缓冲区的大小以及每一个batch的大小：

buffer.memory=         //默认32MB
batch.size=            //默认16KB
long nonPooledAvailableMemory
Deque<ByteBuffer> free           //可用的byteBuffer空间，每一个ByteBuffer大小就是上面配置的batch.size

整个缓冲区可用的空间 = nonPooledAvailableMemory + free batch.size
在最开始整个缓冲区中free是空的，所有的内存空间都在nonPooledAvailableMemory中，每要创建一个batch(batch的大小正好是batch.size)就会从nonPooledAvailableMemory获取空间，用完释放空间时，空间不会回收到nonPooledAvailableMemory中，而是将ByteBuffer放到free中。那么当下一次需要创建batch的时候，如果free中有没有使用的ByteBuffer，就直接从free中获取。
而对于需要创建size大于batch.size的batch时，永远都是直接从nonPooledAvailableMemory获取空间，并且释放时放回nonPooledAvailableMemory中。如果nonPooledAvailableMemory不够时，会从free中释放一些ByteBuffer给nonPooledAvailableMemory。
为什么有一些batch size不是配置的size大小呢？
因为有一些消息体本身很大，大小超过batch.size，这些消息每一条会创建一个ProducerBatch。
Kafka Producer单条消息最大默认是1MB。
对于这些消息，Kafka Producer其实相当于是一条一条发送消息，并且在BufferPool中并没有很好的利用ByteBuffer，所以他们会影响Kafka Producer的吞吐量的。
所以在实际的生产环境中要根据消息的大小调整batch.size的大小
如果nonPooledAvailableMemory + free batch.size的大小也不够创建batch时，程序就会等待别的正在使用的batch释放空间，这个block时间默认是1min。

申请空间

public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {
... ...
try {
    //如何申请的资源正好是BATCH.SIZE的大小，并且free中有没有使用的ByteBuffer，直接使用
    if (size == poolableSize && !this.free.isEmpty())
        return this.free.pollFirst();
    int freeListSize = freeSize() * this.poolableSize;
    //总的可以使用的内存大小 = this.nonPooledAvailableMemory + freeListSize
    if (this.nonPooledAvailableMemory + freeListSize >= size) {
        //nonPooledAvailableMemory内存不够用时，释放掉一些free中的byteBuffer，知道够size用
        freeUp(size);
        this.nonPooledAvailableMemory -= size;
    } else {
        //资源不够用的时候，等待资源
        int accumulated = 0;
        Condition moreMemory = this.lock.newCondition();
        try {
            long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);
            this.waiters.addLast(moreMemory);
            //循环直到内存够用
            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 (this.closed)
                    throw new KafkaException("Producer closed while allocating memory");
                if (waitingTimeElapsed) {
                    //等待超时报错
                    this.metrics.sensor("buffer-exhausted-records").record();
                    throw new BufferExhaustedException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms.");
                }
                remainingTimeToBlockNs -= timeNs;
                //因为上面其他batch释放了资源，所以在此尝试获取资源
                if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {
                    //如果是batch.size，并且释放的资源使得free部位空时，就从free中获取byteBuffer直接使用
                    buffer = this.free.pollFirst();
                    accumulated = size;
                } else {
                    //nonPooledAvailableMemory内存不够用时，释放掉一些free中的byteBuffer给nonPooledAvailableMemory，知道够size用
                    freeUp(size - accumulated);
                    int got = (int) Math.min(size - accumulated, this.nonPooledAvailableMemory);
                    this.nonPooledAvailableMemory -= got;
                    accumulated += got;
                }
            }
            accumulated = 0;
        } finally {
            //没有成功获取到资源，把获取的一部分资源交还给nonPooledAvailableMemory
            this.nonPooledAvailableMemory += accumulated;
            this.waiters.remove(moreMemory);
        }
    }
} finally {
    try {
        //有内存资源剩余的时候，通知的资源等候着
        if (!(this.nonPooledAvailableMemory == 0 && this.free.isEmpty()) && !this.waiters.isEmpty())
        this.waiters.peekFirst().signal();
    } finally {
        lock.unlock();
    }
}
if (buffer == null)
    return safeAllocateByteBuffer(size);
else
    return buffer;
}
private void freeUp(int size) {
    //nonPooledAvailableMemory内存不够用时，释放掉一些free中的byteBuffer，直到够size用
    while (!this.free.isEmpty() && this.nonPooledAvailableMemory < size)
    this.nonPooledAvailableMemory += this.free.pollLast().capacity();
}

释放空间

public void deallocate(ByteBuffer buffer, int size) {
    lock.lock();
    try {
        if (size == this.poolableSize && size == buffer.capacity()) {
            //batch.size大小的资源直接放到free中
            buffer.clear();
        this.free.add(buffer);
    } else {
        //不是batch.size大小的资源放到nonPooledAvailableMemory中
        this.nonPooledAvailableMemory += size;
    }
    Condition moreMem = this.waiters.peekFirst();
    if (moreMem != null)
        moreMem.signal();
    } finally {
        lock.unlock();
    }
}