Kafka 为了避免 Full GC,竟然还在发送端设计了内存池,自己管理内存,太巧妙了...

2024-09-06 19:18

本文主要是介绍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,竟然还在发送端设计了内存池,自己管理内存,太巧妙了...的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1142884

相关文章

Debezium 与 Apache Kafka 的集成方式步骤详解

《Debezium与ApacheKafka的集成方式步骤详解》本文详细介绍了如何将Debezium与ApacheKafka集成,包括集成概述、步骤、注意事项等,通过KafkaConnect,D... 目录一、集成概述二、集成步骤1. 准备 Kafka 环境2. 配置 Kafka Connect3. 安装 D

golang内存对齐的项目实践

《golang内存对齐的项目实践》本文主要介绍了golang内存对齐的项目实践,内存对齐不仅有助于提高内存访问效率,还确保了与硬件接口的兼容性,是Go语言编程中不可忽视的重要优化手段,下面就来介绍一下... 目录一、结构体中的字段顺序与内存对齐二、内存对齐的原理与规则三、调整结构体字段顺序优化内存对齐四、内

SpringBoot中使用 ThreadLocal 进行多线程上下文管理及注意事项小结

《SpringBoot中使用ThreadLocal进行多线程上下文管理及注意事项小结》本文详细介绍了ThreadLocal的原理、使用场景和示例代码,并在SpringBoot中使用ThreadLo... 目录前言技术积累1.什么是 ThreadLocal2. ThreadLocal 的原理2.1 线程隔离2

MySQL报错sql_mode=only_full_group_by的问题解决

《MySQL报错sql_mode=only_full_group_by的问题解决》本文主要介绍了MySQL报错sql_mode=only_full_group_by的问题解决,文中通过示例代码介绍的非... 目录报错信息DataGrip 报错还原Navicat 报错还原报错原因解决方案查看当前 sql mo

C#多线程编程中导致死锁的常见陷阱和避免方法

《C#多线程编程中导致死锁的常见陷阱和避免方法》在C#多线程编程中,死锁(Deadlock)是一种常见的、令人头疼的错误,死锁通常发生在多个线程试图获取多个资源的锁时,导致相互等待对方释放资源,最终形... 目录引言1. 什么是死锁?死锁的典型条件:2. 导致死锁的常见原因2.1 锁的顺序问题错误示例:不同

Linux内存泄露的原因排查和解决方案(内存管理方法)

《Linux内存泄露的原因排查和解决方案(内存管理方法)》文章主要介绍了运维团队在Linux处理LB服务内存暴涨、内存报警问题的过程,从发现问题、排查原因到制定解决方案,并从中学习了Linux内存管理... 目录一、问题二、排查过程三、解决方案四、内存管理方法1)linux内存寻址2)Linux分页机制3)

Java循环创建对象内存溢出的解决方法

《Java循环创建对象内存溢出的解决方法》在Java中,如果在循环中不当地创建大量对象而不及时释放内存,很容易导致内存溢出(OutOfMemoryError),所以本文给大家介绍了Java循环创建对象... 目录问题1. 解决方案2. 示例代码2.1 原始版本(可能导致内存溢出)2.2 修改后的版本问题在

大数据小内存排序问题如何巧妙解决

《大数据小内存排序问题如何巧妙解决》文章介绍了大数据小内存排序的三种方法:数据库排序、分治法和位图法,数据库排序简单但速度慢,对设备要求高;分治法高效但实现复杂;位图法可读性差,但存储空间受限... 目录三种方法:方法概要数据库排序(http://www.chinasem.cn对数据库设备要求较高)分治法(常

Redis多种内存淘汰策略及配置技巧分享

《Redis多种内存淘汰策略及配置技巧分享》本文介绍了Redis内存满时的淘汰机制,包括内存淘汰机制的概念,Redis提供的8种淘汰策略(如noeviction、volatile-lru等)及其适用场... 目录前言一、什么是 Redis 的内存淘汰机制?二、Redis 内存淘汰策略1. pythonnoe

Java中Springboot集成Kafka实现消息发送和接收功能

《Java中Springboot集成Kafka实现消息发送和接收功能》Kafka是一个高吞吐量的分布式发布-订阅消息系统,主要用于处理大规模数据流,它由生产者、消费者、主题、分区和代理等组件构成,Ka... 目录一、Kafka 简介二、Kafka 功能三、POM依赖四、配置文件五、生产者六、消费者一、Kaf