flink内存管理模型(二) ------ 内存分配_flink 1.7.2内存管理

本文主要简单介绍TaskManager的内存管理策略，接上文https://blog.csdn.net/lhy18235303007/article/details/108477130。以下均为笔者个人观点，欢迎大家批评指正。

二、 内存分配

flink在启动一个TM的时候，只会通过两个启动参数限制的JVM的内存使用

1. JVM使用的最大堆内存(-Xms -Xmx)

2. JVM使用的最大堆外内存(-XX:MaxDirectMemorySize)

这两项内存的和即为上述TaskManager使用-tm指定的内存，比如TM的内存为2048M, 计算出来的堆内存(-Xmx)使用为1024M,则堆外内存(-XX:MaxDirectMemorySize)的分配为1024M

那么这个参数(-Xmx)是怎么计算出来的？

2.1 堆内存(-Xmx)计算

在介绍JVM堆内存之前，先简单介绍一下相关配置，以下是笔者从官网找到的相关配置

下面我们来说明如何计算堆内存(设定TM指定的内存为M)

1. 从TM指定的内存中减去保留的内存M1 = M * (1 - containerized.heap-cutoff-ratio), 此时剩余内存为M1

2. 从剩余内存中减去网络内存池中使用的内存M2 = M1 * (1 -taskmanager.network.memory.fraction), 此时剩余内存为M2

3. 如果TM管理的内存为堆外内存(taskmanager.memory.off-heap = true)，则需要从剩余内存M2中减去这部分 M3 = M2 * (1-taskmanager.memory.fraction),否则 M3 = M2

4. M3即为我们得到的堆内存

这样，我们已经得到了启动参数使用的最大堆内存，需要说明的是，在启动TM时，我们设定了-Xms -Xmx两个参数都为此时的最大堆内存。也就是说，我们的JVM至少会占用-Xmx设定的内存使用量。

另一方面，如果在配置文件中使用env.java.opts参数，并不会影响现有的堆内存计算方式，只是将该参数对应的配置项追加到JVM的启动参数中，显然env.java.opts如果设置了-Xms -Xmx参数，会覆盖掉我们之前计算出来的值。

至此，我们得到了JVM的使用的堆内存(-Xmx),接下来，我们将计算网络传输内存池和Flink自己管理的内存分配。

2.2 网络传输内存池的计算

网络内存池和Flink自己管理的内存分配是由当前运行下JVM的最大可用堆内存为基础反推得到的，需要注意的是

前面提到了我们设定了-Xmx参数来设定最大堆内存，但实际计算中并非直接使用的该值而是Runtime中拿到的maxMemory

final long maxMemory = Runtime.getRuntime().maxMemory();

这个内存与前述设定的-Xmx值并不相等，原因在于JVM的内存布局中两个Survivor Space不能同时使用，因此这里的内存是-Xmx指定的内存减去一个Survivor Space占用的内存,例如，我们前述使用-Xmx1024m设置限制了最大堆内存的使用为1024M,假如堆内存三种内存池的分配比例为Java提供的默认值，我们可以计算出这三种内存池使用的内存分别为

因此得到的计算基数为 1024 -34.13 = 989.87。

接下来计算网络内存池的内存数量，需要分两种情况，

当Flink自己管理的内存来自堆外内存时，其计算基数需要在maxMemory的基础上加上Flink自己管理的内存，即使用jvmHeapNoNet作为计算基数

base = maxMemory / (1.0 - taskmanager.memory.fraction);

当Flink自己管理的内存从Heap中划分时，其计算基数为前面计算的maxMemory

base = maxMemory

因此我们可以得到网络内存池使用的内存数量

networkBufBytes = (base / (1.0 - taskmanager.network.memory.fraction) * taskmanager.network.memory.fraction)

网络传输内存池的大小不允许超过堆内存，因此taskmanager.network.memory.fraction这个变量的值必须小于0.5。

2.3 Flink自己管理的内存数量

下面我们来计算Flink自己管理的内存数量

根据前述，Flink自己管理的内存可以根据taskmanager.memory.off-heap分为堆外内存和堆内存

针对堆内内存,其计算基数进行了调整

Runtime r = Runtime.getRuntime();
 
long base = r.maxMemory() - (r.totalMemory() - r.freeMemory())

在计算之前，会显式的调用一次GC.由公式可以看出，计算Flink管理内存时，其基数是当前的可用内存。

此时,我们可以可以得到Flink自己管理的数量

long memorySize = base * taskmanager.memory.fraction

针对堆外内存

如果使用堆外内存，其计算的基数为我们当前之前计算得到的maxMemory,如下：

long directMemorySize = (long) (maxJvmHeapMemory / (1.0 - taskmanager.memory.fraction) * taskmanager.memory.fraction);

还有一个配置项taskmanager.memory.preallocate需要注意,当这个配置项设置为true时,将会立即申请Flink管理的内存数量。

在分配完网络内存池、flink自己管理的内存以及堆内存后，其他的内存作为空闲内存预留在池子里，比如JVM的非堆内存的使用等。

 

这样整个内存就划分完毕。如有不正支持，欢迎批评指正。