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架构描述

  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内部定义了两种情况的清理逻辑：

  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 = () -> {};

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/421271.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

IP劫持的危害分析及应对策略

IP劫持的危害分析及应对策略

在当今数字化时代&#xff0c;网络安全问题备受关注&#xff0c;其中IP劫持是一种常见而危险的威胁。本文将深入探讨IP劫持的危害&#xff0c;并提供一些有效的应对策略。 第一部分&#xff1a;IP劫持的定义 IP劫持是指黑客通过各种手段获取并篡改目标IP地址的控制权&#xf…
阅读更多...
解决Windows下Goland的Terminal设置为Git Bash失败

解决Windows下Goland的Terminal设置为Git Bash失败

路径不要选错了&#xff1a; 如果还是不行&#xff1a; 把bash路径加进去试试 goland设置Terminal
阅读更多...
【Python 元编程】自定义上下文管理器入门指南 ✨

【Python 元编程】自定义上下文管理器入门指南 ✨

【Python 元编程】自定义上下文管理器 ✨ 前言&#x1f4dc; with 语句想必大家都不陌生&#xff0c;在使用 Python 进行文件操作时候&#xff0c;都少不了它。 如以下&#xff1a; with open(file.txt, moder) as file:...但是我想大部分人只知其然&#xff0c;不知其所以…
阅读更多...
LINUX常用工具之VI编辑器

LINUX常用工具之VI编辑器

一、VI编辑器的用途 vi编辑器是所有Unix及Linux系统下标准的编辑器&#xff0c;它提供了一个视窗设备&#xff0c;通过它可以编辑文件它的强大不逊色于任何最新的文本编辑器。由于对Unix及Linux任系统的何版本&#xff0c;vi编辑器是完全相同的&#xff0c;因此您可以在其他任…
阅读更多...
三、Flask学习之BootSrap

三、Flask学习之BootSrap

三、Flask学习之BootSrap Bootstrap 是一款由Twitter团队开发的开源前端框架&#xff0c;它以响应式设计、移动端友好和丰富的组件为特色&#xff0c;为开发者提供了快速构建现代化网站和Web应用的工具。借助其灵活的栅格系统、丰富的UI组件和可定制的样式&#xff0c;Bootstr…
阅读更多...
class_15:虚函数

class_15:虚函数

#include <iostream> #include <string> using namespace std;//基类,父类 class Vehicle{ public:string type;string contry;string color;double price;int numOfWheel;virtual ~Vehicle(){};//类中有虚函数&#xff0c;析构函数一般也写成虚函数virtual voi…
阅读更多...
嵌入式软件工程师面试题——2025校招社招通用(计算机网络篇)（三十二）

嵌入式软件工程师面试题——2025校招社招通用(计算机网络篇)（三十二）

说明&#xff1a; 面试群&#xff0c;群号&#xff1a; 228447240面试题来源于网络书籍&#xff0c;公司题目以及博主原创或修改&#xff08;题目大部分来源于各种公司&#xff09;&#xff1b;文中很多题目&#xff0c;或许大家直接编译器写完&#xff0c;1分钟就出结果了。但…
阅读更多...
蓝桥杯官网填空题（奇怪的分式）

蓝桥杯官网填空题（奇怪的分式）

题目描述 本题为填空题&#xff0c;只需要算出结果后&#xff0c;在代码中使用输出语句将所填结果输出即可。 上小学的时候&#xff0c;小明经常自己发明新算法。一次&#xff0c;老师出的题目是&#xff1a;1/4乘以8/5 小明居然把分子拼接在一起&#xff0c;分母拼接在一起&…
阅读更多...
嵌入式软件工程师面试题——2025校招社招通用(计算机网络篇)（三十一）

嵌入式软件工程师面试题——2025校招社招通用(计算机网络篇)（三十一）

说明&#xff1a; 面试群&#xff0c;群号&#xff1a; 228447240面试题来源于网络书籍&#xff0c;公司题目以及博主原创或修改&#xff08;题目大部分来源于各种公司&#xff09;&#xff1b;文中很多题目&#xff0c;或许大家直接编译器写完&#xff0c;1分钟就出结果了。但…
阅读更多...
【目标检测】损失函数：不同损失函数概念及其代码实现

【目标检测】损失函数：不同损失函数概念及其代码实现

本篇文章介绍目标检测中不同的损失函数概念及其代码实现。目标检测主要任务为实现目标的分类与定位&#xff0c;其损失组成如下&#xff1a; 类别/置信度损失&#xff08;分类任务&#xff09;&#xff1a;BCE&#xff0c;FL&#xff0c;QFL&#xff0c;VFL 位置损失&#xff0…
阅读更多...
QCustomPlot开源库使用

QCustomPlot开源库使用

1.简介 QCustomPlot是用于绘图和数据可视化的Qt C 小部件。它没有进一步的依赖关系&#xff0c;并且有据可查。该绘图库专注于制作美观&#xff0c;出版质量的2D绘图&#xff0c;图形和图表&#xff0c;以及为实时可视化应用程序提供高性能。看一下“ 设置”和“ 基本绘图”教…
阅读更多...
银河麒麟桌面桌面操作系统v10保姆级安装

银河麒麟桌面桌面操作系统v10保姆级安装

目录 一、下载ISO映像文件 1.产品试用申请 2.试用版下载 二、虚拟机搭建 1.新建虚拟机 2. 选择虚拟机硬件兼容性 3.选择安装客户机操作系统 4.选择客户机操作系统 5.命名虚拟机 6.处理器配置 7.虚拟机内存 8.网络类型 9.硬件 10.指定磁盘容量 三、修改虚拟…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023