flink内存管理（二）：MemorySegment的设计与实现:（1）架构、（2）管理堆内/外内存、（3）写入/读取内存、（4）垃圾清理器

文章目录

一. MemorySegment架构概览
二. MemorySegment详解
- - 1.基于MemorySegment管理堆内存
  - 2.基于MemorySegment管理堆外内存
  - 3.基于Unsafe管理MemorySegment
  - 4.写入和读取内存数据
  - 5.创建MemoryCleaner垃圾清理器

在flink内存管理（一）中我们已经知道：在Flink中会将对象序列化成二进制格式数据，然后写入预先分配的内存块，而这个内存块就是MemorySegment。
MemorySegments作为Flink内存管理的最小内存分配单元，能够申请堆内存和堆外内存空间，并对上层提供丰富且高效的内存数据读写方法。

一. MemorySegment架构概览

在flink1.16.1中MemorySegment已作为单独的一个类用于处理：堆内内存、堆外直接内存或堆外不安全内存。

单独的原因：关于效率的注意事项：为了获得最佳效率，MemorySegment不会通过继承来分离不同内存类型的实现，以避免在调用抽象方法时寻找具体实现的开销。

在这里插入图片描述

MemorySegment架构描述

为了尽可能避免直接实例化MemorySegment对象，Flink通过MemorySegmentFactory工厂类创建了MemorySegment。这是因为使用工厂模式控制类的创建，能够帮助JIT执行虚化（de-virtualized）和内联（inlined）的性能优化。
DataOutputView接口扩展了java.io.DataOutput接口，提供了对一个或多个MemorySegment执行写入操作的视图。使用DataOutputView提供的方法可以灵活高效地将数据按顺序写入连续的MemorySegment内存块中。
DataInputView接口扩展了java.io.DataInput接口，提供了对一个或多个MemorySegment执行读取操作的视图。使用DataInputView提供的方法可以灵活高效地按顺序读取MemorySegment中的内存数据。
MemoryManager主要用于管理排序、哈希和缓存等操作对应的内存空间，且这些操作主要集中在离线计算场景中。
NetworkBufferPool通过MemorySegmentFactory申请用于存储NetworkBuffer的MemorySegment内存空间。

JIT编译

JIT（即时编译）是一种在程序运行时将代码编译成机器码的编译技术。JIT编译的优势在于它可以根据程序的实际运行情况进行优化，使得在特定平台上达到更好的性能。

MemorySegment是系统最底层的内存管理单元，可想而知，MemorySegment在整个系统中的使用频率是非常高的。在JIT编译过程中，最好的处理方式就是明确需要调用的方法，早期MemorySegment因为是独立的Final类，JIT编译时要调用的方法都是确定的。

之前的版本将HybridMemorySegment和HeapMemorySegment两个子类加载到JVM，此时JIT编译器只有在真正执行方法的时候才会确认是哪一个子类的方法，这样就无法提前判断使用哪一个实现的虚方法，也就无法直接调用，就会影响JVM的性能。

二. MemorySegment详解

1.基于MemorySegment管理堆内存

我们已经知道，MemorySegment只能通过MemorySegmentFactory创建，并且在MemorySegmentFactory中直接提供了基于堆内存创建MemorySegment的方法。

如下代码所示，在MemorySegmentFactory.wrap()方法中可以直接将byte[] buffer数组封装成MemorySegment，其中byte[]数组中的内存空间实际上就是从堆内存中申请的。

/**
*创建一个以给定堆内存区域为新的memory segments。
此方法应用于将短期字节数组转换为memory segments.
*/
public static MemorySegment wrap(byte[] buffer) {  
    return new MemorySegment(buffer, null);  
}

除了将已有的byte[]数组空间转换成MemorySegment之外，在MemorySegmentFactory中同时提供了通过分配堆内存空间创建MemorySegment的方法。如代码所示，在MemorySegmentFactory.allocateUnpooledSegment()方法中通过指定参数size申请固定数量的byte[]数组，这里new byte[size]的操作实际上就是从堆内存申请内存空间。

//分配一些非池化内存并创建一个代表该内存的新内存段。
//此方法类似于allocateUnpooledSegment(int, Object) ，但内存段的所有者为 null。
public static MemorySegment allocateUnpooledSegment(int size) {  
    return allocateUnpooledSegment(size, null);  
}

在MemorySegment构造器中，提供了对byte[] buffer堆内存进行初始化的逻辑，在方法中首先将buffer赋值给heapMemory，然后将address设定为BYTE_ARRAY_BASE_OFFSET，表示byte[]数组内容的起始部分，然后根据数组对象和偏移量获取元素值（getObject）。

设定offHeapBuffer和cleaner为空。offHeapBuffer和cleaner主要在OffHeap中使用，owner参数表示当前的所有者，通常情况下设定为空。

//创建一个新的memory segment，表示字节数组的内存。
//由于字节数组由堆内存支持，因此该内存段将其数据保存在堆上。缓冲区的大小必须至少为 8 字节。
//memory segment引用给定的owner
MemorySegment(@Nonnull byte[] buffer, @Nullable Object owner) {  
    this.heapMemory = buffer;  
    this.offHeapBuffer = null;  
    this.size = buffer.length;  
    this.address = BYTE_ARRAY_BASE_OFFSET;  
    this.addressLimit = this.address + this.size;  
    this.owner = owner;  
    this.allowWrap = true;  
    this.cleaner = null;  
    this.isFreedAtomic = new AtomicBoolean(false);  
}

2.基于MemorySegment管理堆外内存

在MemorySegment中通过ByteBuffer.allocateDirect(numBytes)方法申请堆外内存，然后用sun.misc.Unsafe对象操作堆外内存。

如下代码，在MemorySegmentFactory.allocateOffHeapUnsafeMemory()方法中，

调用MemoryUtils.allocateUnsafe(size)方法获取堆外内存空间的地址，
然后调用MemoryUtils.wrapUnsafeMemoryWithByteBuffer()方法从给定的内存地址中申请内存空间，并转换成ByteBuffer，
最后通过HybridMemorySegment对象封装ByteBuffer，并返回给使用方进行使用。

/**
分配堆外不安全内存并创建一个新的内存段来表示该内存。
该段的创建会在其 java 包装对象即将被垃圾回收时调度其内存释放操作，类似于java.nio.DirectByteBuffer.DirectByteBuffer(int) 。不同之处在于，此内存分配不受选项 -XX:MaxDirectMemorySize 限制。
**/
public static MemorySegment allocateOffHeapUnsafeMemory(  
        int size, Object owner, Runnable customCleanupAction) {  
    long address = MemoryUtils.allocateUnsafe(size);  
    ByteBuffer offHeapBuffer = MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);  
    Runnable cleaner = MemoryUtils.createMemoryCleaner(address, customCleanupAction);  
    return new MemorySegment(offHeapBuffer, owner, false, cleaner);  
}

如代码清单8-6所示，在MemoryUtils.wrapUnsafeMemoryWithByteBuffer()方法中，

首先调用DIRECT_BUFFER_CONSTRUCTOR.newInstance(address, size)方法从堆外内存中申请内存空间并转换为ByteBuffer对象。
对于DIRECT_BUFFER_CONSTRUCTOR变量，实际上就是创建DirectByteBuffer对应的私有构造器，调用allocateInstance()方法就能直接从堆外内存中创建DirectByteBuffer对象。

//用ByteBuffer包装不安全的native memory
static ByteBuffer wrapUnsafeMemoryWithByteBuffer(long address, int size) {  
    //noinspection OverlyBroadCatchBlock  
    try {  
        ByteBuffer buffer = (ByteBuffer) UNSAFE.allocateInstance(DIRECT_BYTE_BUFFER_CLASS);  
        UNSAFE.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);  
        UNSAFE.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);  
        buffer.clear();  
        return buffer;  
    } catch (Throwable t) {  
        throw new Error("Failed to wrap unsafe off-heap memory with ByteBuffer", t);  
    }  
}

在MemoryUtils中，DIRECT_BUFFER_CONSTRUCTOR通过反射获取。通过java.nio.DirectByteBuffer构造器创建ByteBuffer内存对象，并将其封装在MemorySegment中。

private static Class<?> getClassByName(  
        @SuppressWarnings("SameParameterValue") String className) {  
    try {  
        return Class.forName(className);  
    } catch (ClassNotFoundException e) {  
        throw new Error("Could not find class '" + className + "' for unsafe operations.", e);  
    }  
}

接下来就可以通过MemorySegment使用申请到的堆外内存存储数据了，数据最终会以二进制的形式存储在指定地址的堆外内存空间中，再看下构造方法。

public static MemorySegment allocateOffHeapUnsafeMemory(  
        int size, Object owner, Runnable customCleanupAction) {  
    long address = MemoryUtils.allocateUnsafe(size);  
    ByteBuffer offHeapBuffer = MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);  
    Runnable cleaner = MemoryUtils.createMemoryCleaner(address, customCleanupAction);  
    return new MemorySegment(offHeapBuffer, owner, false, cleaner);  
}

3.基于Unsafe管理MemorySegment

使用unsafe操作内存
MemorySegment能够同时操作堆内存和堆外内存，得益于sun.misc.Unsafe类的实现，Unsafe类提供了一系列可以直接操作内存的方法。sun.misc.Unsafe目前也已经下沉到MemorySegment中实现。

需要注意的是，Unsafe并不是JDK的标准实现，而是Sun的内部实现，存在于sun.misc包中，在Oracle发行的JDK中并不包含其源代码。

虽然我们在一般的并发编程中不会直接用到Unsafe，但对于很多Java基础类库，如Netty、Cassandra和Kafka等高性能的框架，基本都会使用Unsafe操作内存空间。Unsafe在提升Java运行效率以及增强Java语言底层操作内存的能力方面起了很大作用。

借助unsafe可以操作堆内、堆外内存使用
如代码所示，MemorySegment中使用sun.misc.Unsafe实现了对内存空间的管理，MemorySegment将创建出来的Unsage对象存储至静态变量，供所有MemorySegment持有者操作堆内存和堆外内存使用。

/** The unsafe handle for transparent memory copied (heap / off-heap). */  
@SuppressWarnings("restriction")  
private static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;

MemorySegment中的Unsafe对象主要通过MemoryUtils创建，和其他框架使用Unsafe库一样，都是通过反射的方式创建的，MemorySegment中的Unsafe创建完毕后，可以通过Unsafe库操作和管理堆内存和堆外内存空间。

4.写入和读取内存数据

在MemorySegment中将int类型的数据写入内存中

在MemorySegment.putInt()方法中，需要传递：值+内存位置

（1）int value值以及（2）value对应的内存位置（index），其中int类型数据大小为4字节。
根据address以及index参数计算位置偏移量（pos），然后调用UNSAFE.putInt()方法将int数值写入指定的内存空间。

/**将给定的 int 值（32 位，4 个字节）写入系统本机字节顺序中的给定位置。
此方法为整数写入提供了最佳速度，除非需要特定的字节顺序，否则应使用此方法。
在大多数情况下，只要知道写入值的字节顺序与读取值的字节顺序相同就足够了
（例如内存中的瞬时存储，或者 I/O 和网络的序列化），使得此方法更好的选择。
**/
public void putInt(int index, int value) {  
    final long pos = address + index;  
    if (index >= 0 && pos <= addressLimit - 4) {  
        UNSAFE.putInt(heapMemory, pos, value);  
    } else if (address > addressLimit) {  
        throw new IllegalStateException("segment has been freed");  
    } else {  
        // index is in fact invalid  
        throw new IndexOutOfBoundsException();  
    }  
}

读取MemorySegment中int类型的数据

在MemorySegment.getInt()方法中需要指定：位置+偏移量

(1)读取int值的位置信息
(2)根据公式address+index计算位置偏移量（pos），address实际上就是前面创建的arrayBaseOffset指标，即基本类型数组的偏移量;
(3)通过数组对象和偏移量获取int元素值。

/**
按照系统的本机字节顺序从给定位置读取 int 值（32 位，4 字节）。
此方法提供了整数读取的最佳速度，除非需要特定的字节顺序，否则应使用此方法。
在大多数情况下，只要知道写入值的字节顺序与读取值的字节顺序相同就足够了
（例如内存中的瞬时存储，或者 I/O 和网络的序列化），使得此方法更好的选择
**/
public int getInt(int index) {  
    final long pos = address + index;  
    if (index >= 0 && pos <= addressLimit - 4) {  
        return UNSAFE.getInt(heapMemory, pos);  
    } else if (address > addressLimit) {  
        throw new IllegalStateException("segment has been freed");  
    } else {  
        // index is in fact invalid  
        throw new IndexOutOfBoundsException();  
    }  
}

5.创建MemoryCleaner垃圾清理器

内存泄漏的情况

通常情况下，直接申请堆外内存是不安全的，一旦JVM进程无法直接访问动态内存，即无法通过指针指向一个可访问的起始对象并访问该对象的内存空间时**，内存空间就会变为不可访问内存**，在使用人工内存管理中，系统不可访问内存最终会导致内存泄漏。

flink提供了MemoryCleaner

因此在创建堆外内存空间的同时，Flink也会调用MemoryUtils.createMemoryCleaner()方法创建内存垃圾清理器，专门用于回收和释放没有被正常释放的内存资源，防止内存泄漏。

createMemoryCleaner封装了两个逻辑：

清理内存：UNSAFE.freeMemory(address)
自定义清理策略：customCleanup

//MemorySegmentFactory. 创建MemoryCleaner
public static MemorySegment allocateOffHeapUnsafeMemory(  
        int size, Object owner, Runnable customCleanupAction) {  
    long address = MemoryUtils.allocateUnsafe(size);  
    ByteBuffer offHeapBuffer = MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);  
    Runnable cleaner = MemoryUtils.createMemoryCleaner(address, customCleanupAction);  
    return new MemorySegment(offHeapBuffer, owner, false, cleaner);  
}


//在MemoryUtils中，创建MemoryCleaner
/**
创建一个清理器来释放不安全的内存。
参数：
address – 要释放的不安全内存的地址 customCleanup – 清理 GC 的自定义操作
返回：
手动运行以释放不安全内存的操作
**/
static Runnable createMemoryCleaner(long address, Runnable customCleanup) {  
    return () -> {  
    
        releaseUnsafe(address);  
        customCleanup.run();  
    };  
}

private static void releaseUnsafe(long address) {  
    UNSAFE.freeMemory(address);  
}

customCleanup在flink内部定义了两种情况的清理逻辑：

内存大小来释放内存

package org.apache.flink.runtime.memory.UnsafeMemoryBudget.
//根据内存大小来释放内存
void releaseMemory(@Nonnegative long size) {  
    if (size == 0) {  
        return;  
    }  
    boolean released = false;  
    long currentAvailableMemorySize = 0L;  
    while (!released  
            && totalMemorySize  
                    >= (currentAvailableMemorySize = availableMemorySize.get()) + size) {  
        released =  
                availableMemorySize.compareAndSet(  
                        currentAvailableMemorySize, currentAvailableMemorySize + size);  
    }  
    if (!released) {  
        throw new IllegalStateException(  
                String.format(  
                        "Trying to release more managed memory (%d bytes) than has been allocated (%d bytes), the total size is %d bytes",  
                        size, currentAvailableMemorySize, totalMemorySize));  
    }  
}

没有清理策略，直接清理

private static final Runnable NO_OP = () -> {};