本文主要是介绍Kafka 为了避免 Full GC,竟然还在发送端设计了内存池,自己管理内存,太巧妙了...,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
一、开篇引出一个 Full Gc 的问题
在上一篇文章中,我们讲到了 Kafka 发送消息的八个流程,并且着重讲了 Kafka 封装了一个内存结构,把每个分区的消息封装成批次,缓存到内存里。
如下图所示:
上图中,整体是一个 Map 结构,Map 的 key 是分区,Map 的值是一个队列;队列里有一个个的小批次,里面是很多消息。
这样好处就是可以一次性的把消息发送出去,不至于来一条发送一条,浪费网络资源。
但由此也带来了问题,生产者端消息这么多,一个批次发送完了就不管了去等待 JVM 的垃圾回收的时候,很有可能会触发 full gc。
一次 full gc,整个 Producer 端的所有线程就都停了,所有消息都无法发送了,由此带来的损耗也是不可小觑。
这个严重的问题,当然 Kafka 的开发者也考虑到了这一点,所以作者设计了一个内存池,用来反复利用被发送出去 RecordBatch,以减少 full gc。
二、什么是内存池
可以类比连接池,连接池缓存了很多 jdbc 连接,避免不必要的创建连接的开销;内存池也一样,可以对 RecordBatch 做到反复利用。
那我们看看 Kafka 内存池是怎么设计的:
Kafka 内存设计有两部分,下面的绿色的是可用的内存(未分配的内存,初始的时候是 32M),上面红色的是已经被分配了的内存,每个小 batch 是 16K,然后这一个个的 batch 就可以被反复利用,不需要每次都申请内存。
两部分加起来是 32M。
这个 32M 的配置在 ProducerConfig 这个类里面:
三、申请内存的过程
(发送消息的流程在上一篇文章讲过了,可以回去复习下)
我们从发送消息的大流程的第七步开始看(当前位置:KafkaProducer):
进入到 RecordAccumulator 类里,当发现还没有队列的时候,创建了一个队列,然后去申请内存(当前类位置:RecordAccumulator):
本次我们主要看的就是这个 allocate 方法。点到 allocate 里面,到了 BufferPool 类,BufferPool 是对内存池的封装。然后来一行行看这个申请内存的方法。
(1)如果申请的内存大小超过了整个缓存池的大小,则抛错出来
(2)对整个方法加锁:
this.lock.lock();
(3)如果申请的大小是每个 recordBatch 的大小(16K),并且已分配内存不为空,则直接取出来一个返回。
if (size == poolableSize && !this.free.isEmpty())return this.free.pollFirst();
(4)如果要申请的内存大小不是 16K 或者已分配内存没有了的情况。
如果整个内存池大小比要申请的内存大小大 (this.availableMemory + freeListSize >= size),则直接从可用内存(即上图绿色的区域)申请一块内存。
并且可用内存要去掉申请的那一块内存。
int freeListSize = this.free.size() * this.poolableSize;
if (this.availableMemory + freeListSize >= size) {// we have enough unallocated or pooled memory to immediately// satisfy the requestfreeUp(size);this.availableMemory -= size;lock.unlock();return ByteBuffer.allocate(size);
}
(5)下面是 else 分支,表示申请的内存大小不是 16 K,或者已分配内存区域没有,并且所有的内存加起来都不够了。
首先创建一个 Condition。Condition 就是用来替代传统的 Object 的 wait() 和 notify() 方法来实现线程间的协作。Condition 必须在 lock 和 unlock 代码块中间才可使用。
Condition moreMemory = this.lock.newCondition();
将 Condition 加入到 waiters 里面。为什么会有多个 Condition 呢?因为这里可能很多个线程都在使用生产者发送消息,可能很多个线程都没有足够的内存分配了,都在等待。
this.waiters.addLast(moreMemory);
然后线程开始睡眠,等待释放资源(唤醒条件有两个,一个是睡眠时间到了,一个是有其他线程释放了内存,被唤醒了):
(7)如果等了指定时间(默认配置是 60s - 获取元数据的时间),还没被唤醒,则直接抛一个缓存超时的异常出去
if (waitingTimeElapsed) {this.waiters.remove(moreMemory);throw new TimeoutException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms.");
}
(8)如果有其他线程释放内存,被唤醒了,从 waiters 列表里面移除自己,然后去看看有没有内存可以用。
这里仍然有两个分支,一个是首先看已分配内存里面有没有内存(16K),如果有的话,直接拿一个 batch 出来
if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {// just grab a buffer from the free listbuffer = this.free.pollFirst();accumulated = size;
}
另一个分支是,如果要申请的不是 16K,或者已分配内存空间不是空的
// 从已分配内存取一个出来放到可用内存区域
freeUp(size - accumulated);
// 申请一块,有可能只能申请到2K
int got = (int) Math.min(size - accumulated, this.availableMemory);
// 做扣减
this.availableMemory -= got;
accumulated += got;
有可能这里只能申请到一部分内存,比如3K,5K,没有达到想申请的那个数量,则会继续走 while 循环。
(9)最后发现内存有富余,则唤醒其他线程
if (this.availableMemory > 0 || !this.free.isEmpty()) {if (!this.waiters.isEmpty())this.waiters.peekFirst().signal();
}
四、释放内存的过程
释放内存的过程很简单了,如果释放的是一个批次的大小(16K),则直接加到已分配内存里面
如果没有,则把内存放到可用内存里面,这部分内存等待虚拟机垃圾回收。
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.availableMemory += size;}Condition moreMem = this.waiters.peekFirst();if (moreMem != null)moreMem.signal();} finally {lock.unlock();}
}
这里可能会有一个疑问:
为什么释放了一个批次大小(16K)内存的时候,才放到已分配内存里面。我想释放个 1M 的内存,为什么不能往已分配内存里面呢?
假设我们往已分配内存里释放了个 1M 的批次到内存里。
然后发送消息其实是有条件的,要么是许多消息把批次撑满了发送出去,要么是一个批次累积消息的时间到了,就会立马发出去。
如果是一个 1M 的内存批次,才攒了几条消息,一个批次才用了 几十K,时间到了,就把这个 1M 的内存批次发送出去了。
那么可想而知,内存的使用率是会非常低的。
所以这里控制了,已分配内存必须是 16K 的,每个批次的大小必须一致,这样才能充分利用内存空间。
五、总结
本文我们讨论了 Kafka 生产者端设计了一个内存池的结构,反复利用每一个批次,减少 Java 虚拟机的内存回收。
本文中,还涉及到了一个高并发锁的代码,比如 可重入锁 ReentrantLock,Condition,如果有不明白的地方,可以把这部分复习一下,再看这段代码就很容易明白了。
这篇关于Kafka 为了避免 Full GC,竟然还在发送端设计了内存池,自己管理内存,太巧妙了...的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!