注:本文依赖于kafka-0.10.0.1-src
我们都知道kafka生产者send一条记录(record)后并没有直接发送到kafka服务端,而是先将它保存到内存(RecordAccumulator)中,用于压缩之后批量发送,这里内存的创建和释放是比较消耗资源的,为了实现内存的高效利用,基本上每个成熟的框架或者工具都有一套内存管理机制,kafka的生产者使用BufferPool来实现内存(Java NIO的ByteBuffer)的复用。
BufferPool是什么呢?如图1:
图1图1包含两个部分,红色和绿色的总和代表BufferPool的总量,用totalMemory表示(由buffer.memory配置);绿色代表可使用的空间,它又包括两个部分:上半部分代表未申请未使用的部分,用availableMemory表示;下半部分代表已经申请但没有使用的部分,用一个ByteBuffer队列(Deque<ByteBuffer>)表示,我们称这个队列为free,队列中的ByteBuffer的大小用poolableSize表示(由batch.size配置)。
下图2总结了从BufferPool中分配固定size大小的内存的步骤:
图2从图2可以看出申请size大小的内存有这么几种结束方式(红色框部分),1、异常结束,比如申请的内存过大超过总量限定2、直接用队列中的ByteBuffer分配内存;3、用avaliableMemory分配内存。
蓝色框内的为大多数的内存分配方式,就是从队列中直接拿想要的ByteBuffer,也是kafka希望的分配方式;黄色的框为分配内存时队列中的内存不符合其分配的条件(队列为空或大小不匹配),从availableMemory中分配;绿色框为当前内存池中内存不足时阻塞等待的情况,具体就是有一个累加器accumulated,如果累加器没有累加到size大小,说明还没有足够的内存释放出来,所以就会阻塞等待内存释放,内存释放之后会唤醒阻塞的线程,将可以分配的内存大小累加到累加器accumulated上,这样直到累加器accumulated大小满足size,就直接分配。这里面还有一个原则就是如果还没给累加器accumulated累加过一次的话,也就是accumulated==0的时候,那么会优先尝试从队列中获取内存(有可能释放的内存释放到队列中)。
释放内存的话就比较简单了,如果释放的大小等于poolableSize的话,就把它放入free队列,否则释放到availableMemory中(availableMemory+=size)。所以只有固定大小的内存块被释放后才会进入池化列表,非常规释放后只会增加可用内存大小。
BufferPool是线程安全的,用一个ReentrantLock来保证,并且用一个Deque<Condition> waiters队列来记录申请不到足够空间而阻塞的线程,此队列中实际记录的是阻塞线程对应的Condition对象。
总结:可以看到BufferPool只针对特定大小(poolableSize)的ByteBuffer进行管理,对于其它大小的并不会缓存进来。因此如果超大消息比较多(大于poolableSize),就不会很好的利用内存池,频繁的申请回收内存效率会降低,并可能带来Full GC或者Out Of Memory Error,这个时候就要调整好batch.size的配置了。
网友评论