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

这篇具有很好参考价值的文章主要介绍了flink内存管理(二):MemorySegment的设计与实现:(1)架构、(2)管理堆内/外内存、(3)写入/读取内存、(4)垃圾清理器。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

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

一. MemorySegment架构概览

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

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

flink内存管理(二):MemorySegment的设计与实现:(1)架构、(2)管理堆内/外内存、(3)写入/读取内存、(4)垃圾清理器,# flink 实战,# flink源码,flink,架构,大数据

MemorySegment架构描述

  1. 为了尽可能避免直接实例化MemorySegment对象,Flink通过MemorySegmentFactory工厂类创建了MemorySegment。这是因为使用工厂模式控制类的创建,能够帮助JIT执行虚化(de-virtualized)和内联(inlined)的性能优化。
  2. DataOutputView接口扩展了java.io.DataOutput接口,提供了对一个或多个MemorySegment执行写入操作的视图。使用DataOutputView提供的方法可以灵活高效地将数据按顺序写入连续的MemorySegment内存块中
  3. DataInputView接口扩展了java.io.DataInput接口,提供了对一个或多个MemorySegment执行读取操作的视图。使用DataInputView提供的方法可以灵活高效地按顺序读取MemorySegment中的内存数据
  4. MemoryManager主要用于管理排序、哈希和缓存等操作对应的内存空间,且这些操作主要集中在离线计算场景中。
  5. 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内部定义了两种情况的清理逻辑:文章来源地址https://www.toymoban.com/news/detail-814035.html

  1. 内存大小来释放内存
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));  
    }  
}
  1. 没有清理策略,直接清理
private static final Runnable NO_OP = () -> {};

到了这里,关于flink内存管理(二):MemorySegment的设计与实现:(1)架构、(2)管理堆内/外内存、(3)写入/读取内存、(4)垃圾清理器的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 智慧图书管理系统架构设计与实现

    随着数字化时代的到来,智慧图书管理系统在图书馆和机构中扮演着重要的角色。一个优秀的图书管理系统不仅需要满足基本的借阅管理需求,还需要具备高效的性能、良好的扩展性和稳定的安全性。本文将讨论智慧图书管理系统的架构设计与实现,以满足现代图书管理的多

    2024年02月20日
    浏览(49)
  • 【大数据】Flink 内存管理(四):TaskManager 内存分配(实战篇)

    《 Flink 内存管理 》系列(已完结),共包含以下 4 篇文章: Flink 内存管理(一):设置 Flink 进程内存 Flink 内存管理(二):JobManager 内存分配(含实际计算案例) Flink 内存管理(三):TaskManager 内存分配(理论篇) Flink 内存管理(四):TaskManager 内存分配(实战篇) 😊

    2024年03月13日
    浏览(54)
  • 学校图书馆管理系统的架构设计与实现

    随着大学生越来越多,学校图书馆的管理变得愈发复杂。为了更好地管理和服务于学生和教职工,学校需要建立一个高效的图书馆管理系统。本文将介绍学校图书馆管理系统的架构设计与实现。 1. 架构设计 学校图书馆管理系统的架构设计主要分为三个部分: 展示层:展示层

    2024年02月22日
    浏览(53)
  • Flink理论—Flink架构设计

    Flink 是一个分布式系统,需要有效分配和管理计算资源才能执行流应用程序。它集成了所有常见的集群资源管理器,例如Hadoop YARN,但也可以设置作为独立集群甚至库运行,例如Spark 的 Standalone Mode 本节概述了 Flink 架构,并且描述了其主要组件如何交互以执行应用程序和从故障

    2024年02月20日
    浏览(51)
  • 【微服务架构设计和实现】4.5 服务发现、注册和配置管理

    往期回顾: 第一章:【云原生概念和技术】 第二章:【容器化应用程序设计和开发】 第三章:【基于容器的部署、管理和扩展】 第四章:【4.1 微服务架构概述和设计原则】 第四章:【4.2 服务边界的定义和划分】 第四章:【4.3 服务之间的通信和API设计】 第四章:【4.4 数

    2024年02月10日
    浏览(39)
  • 【大数据】Flink 架构(四):状态管理

    《 Flink 架构 》系列(已完结),共包含以下 6 篇文章: Flink 架构(一):系统架构 Flink 架构(二):数据传输 Flink 架构(三):事件时间处理 Flink 架构(四):状态管理 Flink 架构(五):检查点 Checkpoint(看完即懂) Flink 架构(六):保存点 Savepoint 😊 如果您觉得这篇

    2024年02月19日
    浏览(43)
  • SQLServer的内存管理架构

    地址空间的已提交区域由 Windows 虚拟内存管理器 (VMM) 映射到可用的物理内存。 虚拟内存系统允许物理内存的过度使用,因此虚拟内存与物理内存的比率可以超过 1:1。因此,较大的程序可以在具有各种物理内存配置的计算机上运行。但是,使用比所有进程的总平均工作集

    2023年04月16日
    浏览(29)
  • 深入理解 Flink(一)Flink 架构设计原理

    深入理解 Flink 系列文章已完结,总共八篇文章,直达链接: 深入理解 Flink (一)Flink 架构设计原理 深入理解 Flink (二)Flink StateBackend 和 Checkpoint 容错深入分析 深入理解 Flink (三)Flink 内核基础设施源码级原理详解 深入理解 Flink (四)Flink Time+WaterMark+Window 深入分析 深入

    2024年02月02日
    浏览(43)
  • PowerDotNet平台化软件架构设计与实现系列(17):PCRM个人用户管理平台

    个人用户管理是业务系统中非常基础且重要的一个公共服务系统,我们写的绝大多数应用都和个人用户或会员有关,用户(会员)数据安全无小事,必须有一个完备的用户管理平台系统。 因为不同公司的主业务不同,个人用户管理的侧重点也会有不同,PowerDotNet这里介绍的个人

    2024年02月03日
    浏览(38)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包