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

文章目录一. MemorySegment架构概览二. MemorySegment详解1.基于MemorySegment管理堆内存2.基于MemorySegment管理堆外内存3.基于Unsafe管理MemorySegment4.写入和读取内存数据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. */publicstaticMemorySegmentwrap(byte[]buffer){returnnewMemorySegment(buffer,null);}除了将已有的byte[]数组空间转换成MemorySegment之外在MemorySegmentFactory中同时提供了通过分配堆内存空间创建MemorySegment的方法。如代码所示在MemorySegmentFactory.allocateUnpooledSegment()方法中通过指定参数size申请固定数量的byte[]数组这里new byte[size]的操作实际上就是从堆内存申请内存空间。//分配一些非池化内存并创建一个代表该内存的新内存段。//此方法类似于allocateUnpooledSegment(int, Object) 但内存段的所有者为 null。publicstaticMemorySegmentallocateUnpooledSegment(intsize){returnallocateUnpooledSegment(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引用给定的ownerMemorySegment(Nonnullbyte[]buffer,NullableObjectowner){this.heapMemorybuffer;this.offHeapBuffernull;this.sizebuffer.length;this.addressBYTE_ARRAY_BASE_OFFSET;this.addressLimitthis.addressthis.size;this.ownerowner;this.allowWraptrue;this.cleanernull;this.isFreedAtomicnewAtomicBoolean(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 限制。 **/publicstaticMemorySegmentallocateOffHeapUnsafeMemory(intsize,Objectowner,RunnablecustomCleanupAction){longaddressMemoryUtils.allocateUnsafe(size);ByteBufferoffHeapBufferMemoryUtils.wrapUnsafeMemoryWithByteBuffer(address,size);RunnablecleanerMemoryUtils.createMemoryCleaner(address,customCleanupAction);returnnewMemorySegment(offHeapBuffer,owner,false,cleaner);}如代码清单8-6所示在MemoryUtils.wrapUnsafeMemoryWithByteBuffer()方法中首先调用DIRECT_BUFFER_CONSTRUCTOR.newInstance(address, size)方法从堆外内存中申请内存空间并转换为ByteBuffer对象。对于DIRECT_BUFFER_CONSTRUCTOR变量实际上就是创建DirectByteBuffer对应的私有构造器调用allocateInstance()方法就能直接从堆外内存中创建DirectByteBuffer对象。//用ByteBuffer包装不安全的native memorystaticByteBufferwrapUnsafeMemoryWithByteBuffer(longaddress,intsize){//noinspection OverlyBroadCatchBlocktry{ByteBufferbuffer(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();returnbuffer;}catch(Throwablet){thrownewError(Failed to wrap unsafe off-heap memory with ByteBuffer,t);}}在MemoryUtils中DIRECT_BUFFER_CONSTRUCTOR通过反射获取。通过java.nio.DirectByteBuffer构造器创建ByteBuffer内存对象并将其封装在MemorySegment中。privatestaticClass?getClassByName(SuppressWarnings(SameParameterValue)StringclassName){try{returnClass.forName(className);}catch(ClassNotFoundExceptione){thrownewError(Could not find class className for unsafe operations.,e);}}接下来就可以通过MemorySegment使用申请到的堆外内存存储数据了数据最终会以二进制的形式存储在指定地址的堆外内存空间中再看下构造方法。publicstaticMemorySegmentallocateOffHeapUnsafeMemory(intsize,Objectowner,RunnablecustomCleanupAction){longaddressMemoryUtils.allocateUnsafe(size);ByteBufferoffHeapBufferMemoryUtils.wrapUnsafeMemoryWithByteBuffer(address,size);RunnablecleanerMemoryUtils.createMemoryCleaner(address,customCleanupAction);returnnewMemorySegment(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)privatestaticfinalsun.misc.UnsafeUNSAFEMemoryUtils.UNSAFE;MemorySegment中的Unsafe对象主要通过MemoryUtils创建和其他框架使用Unsafe库一样都是通过反射的方式创建的MemorySegment中的Unsafe创建完毕后可以通过Unsafe库操作和管理堆内存和堆外内存空间。4.写入和读取内存数据在MemorySegment中将int类型的数据写入内存中在MemorySegment.putInt()方法中需要传递值内存位置1int value值以及2value对应的内存位置index其中int类型数据大小为4字节。根据address以及index参数计算位置偏移量pos然后调用UNSAFE.putInt()方法将int数值写入指定的内存空间。/**将给定的 int 值32 位4 个字节写入系统本机字节顺序中的给定位置。 此方法为整数写入提供了最佳速度除非需要特定的字节顺序否则应使用此方法。 在大多数情况下只要知道写入值的字节顺序与读取值的字节顺序相同就足够了 例如内存中的瞬时存储或者 I/O 和网络的序列化使得此方法更好的选择。 **/publicvoidputInt(intindex,intvalue){finallongposaddressindex;if(index0posaddressLimit-4){UNSAFE.putInt(heapMemory,pos,value);}elseif(addressaddressLimit){thrownewIllegalStateException(segment has been freed);}else{// index is in fact invalidthrownewIndexOutOfBoundsException();}}读取MemorySegment中int类型的数据在MemorySegment.getInt()方法中需要指定位置偏移量(1)读取int值的位置信息(2)根据公式addressindex计算位置偏移量posaddress实际上就是前面创建的arrayBaseOffset指标即基本类型数组的偏移量;(3)通过数组对象和偏移量获取int元素值。/** 按照系统的本机字节顺序从给定位置读取 int 值32 位4 字节。 此方法提供了整数读取的最佳速度除非需要特定的字节顺序否则应使用此方法。 在大多数情况下只要知道写入值的字节顺序与读取值的字节顺序相同就足够了 例如内存中的瞬时存储或者 I/O 和网络的序列化使得此方法更好的选择 **/publicintgetInt(intindex){finallongposaddressindex;if(index0posaddressLimit-4){returnUNSAFE.getInt(heapMemory,pos);}elseif(addressaddressLimit){thrownewIllegalStateException(segment has been freed);}else{// index is in fact invalidthrownewIndexOutOfBoundsException();}}5.创建MemoryCleaner垃圾清理器内存泄漏的情况通常情况下直接申请堆外内存是不安全的一旦JVM进程无法直接访问动态内存即无法通过指针指向一个可访问的起始对象并访问该对象的内存空间时**内存空间就会变为不可访问内存**在使用人工内存管理中系统不可访问内存最终会导致内存泄漏。flink提供了MemoryCleaner因此在创建堆外内存空间的同时Flink也会调用MemoryUtils.createMemoryCleaner()方法创建内存垃圾清理器专门用于回收和释放没有被正常释放的内存资源防止内存泄漏。createMemoryCleaner封装了两个逻辑清理内存UNSAFE.freeMemory(address)自定义清理策略customCleanup//MemorySegmentFactory. 创建MemoryCleanerpublicstaticMemorySegmentallocateOffHeapUnsafeMemory(intsize,Objectowner,RunnablecustomCleanupAction){longaddressMemoryUtils.allocateUnsafe(size);ByteBufferoffHeapBufferMemoryUtils.wrapUnsafeMemoryWithByteBuffer(address,size);RunnablecleanerMemoryUtils.createMemoryCleaner(address,customCleanupAction);returnnewMemorySegment(offHeapBuffer,owner,false,cleaner);}//在MemoryUtils中创建MemoryCleaner/** 创建一个清理器来释放不安全的内存。 参数 address – 要释放的不安全内存的地址 customCleanup – 清理 GC 的自定义操作 返回 手动运行以释放不安全内存的操作 **/staticRunnablecreateMemoryCleaner(longaddress,RunnablecustomCleanup){return()-{releaseUnsafe(address);customCleanup.run();};}privatestaticvoidreleaseUnsafe(longaddress){UNSAFE.freeMemory(address);}customCleanup在flink内部定义了两种情况的清理逻辑内存大小来释放内存packageorg.apache.flink.runtime.memory.UnsafeMemoryBudget.//根据内存大小来释放内存voidreleaseMemory(Nonnegativelongsize){if(size0){return;}booleanreleasedfalse;longcurrentAvailableMemorySize0L;while(!releasedtotalMemorySize(currentAvailableMemorySizeavailableMemorySize.get())size){releasedavailableMemorySize.compareAndSet(currentAvailableMemorySize,currentAvailableMemorySizesize);}if(!released){thrownewIllegalStateException(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));}}没有清理策略直接清理privatestaticfinalRunnableNO_OP()-{};