Flink 1.12.2 源码浅析 : 内存管理相关

逃离我推掉我的手 2023-01-20 03:00 23阅读 0赞

### . ###

*  一 .前言
 *  二 .MemorySegment \[内存页\]
 *   *  2.1. 属性
     *  2.2. 构造方法
     *  2.3. 方法
     *  2.4. 实现类HybridMemorySegment
     *   *  2.4.1. 属性
         *  2.4.2. 构造方法
         *  2.4.3. free
         *  2.4.4. warp
         *  2.4.5. get/put相关
 *  三. 内存页
 *   *  3.1. DataInputView
     *   *  3.1.1. 前言
         *  3.1.2. AbstractPagedInputView
         *   *  3.1.2.1 属性
             *  3.1.2.2 构造方法
             *  3.1.2.3 Page 管理相关
             *  3.1.2.4 读取相关方法
     *  3.2. DataOutputView
     *  3.3. 内存页的使用

# 一 .前言 #

flink 从一开始就选择了使用自主的内存管理,

原因如下:

1.  Java对象存储密度低。Java的对象在内存中存储包含3个主要部分：对象头，实例数据，拆分填充部分。例如，一个只包含boolean属性的对象占16byte：对象头占8byte，boolean属性占1字节，为了对齐达到8的倍数额外占7字节。而实际上只需要一个bit（1/8字节）就够了。
2.  Full GC会极大地影响性能。尤其是为了处理大量数据而开了很大的内存空间的JVM来说，GC会达到秒级甚至分钟级。
3.  OOM问题影响稳定性。OutOfMemoryError是分布式计算框架经常会遇到的问题，当JVM中所有对象大小超过分配给JVM的内存大小时，就会发生OutOfMemory Error错误，导致JVM崩溃，分散框架的健壮性和性能都会受到影响。
4.  缓存未命中中问题。CPU进行计算的时候，是从CPU缓存中获取数据。这样就会大大降低CacheMiss。使得CPU集中处理业务，而不是空转。（Java的对象在堆上存储的时候并不是连续的，所以从内存中读取的Java对象时，缓存的邻近的内存区域的数据往往不是CPU后续计算所需要的，这就是缓存未命中。此时CPU需要空转等待从内存中重新读取数据。）

MemorySegment 是一段固定长度的内存，也是 Flink 中最小的内存分配单元,默认大小为32k。支持堆上内存和堆外内存两种.  
堆上的话,内部存储为一个java baty数组. 在堆外的话, 是一个ByteBuffer (基于 Netty 的 DirectByteBufferr).每条记录都会以序列化的形式存储在一个或多个MemorySegment中

*  堆内内存的劣势

1.  启动很大堆内存（100s of GBytes heap memory）的 JVM 需要很长时间，GC 停留时间也会很长（分钟级）。使用堆外内存的话，JVM 只需要分配较少的堆内存（只需要分配 Remaining Heap 那一块）。
2.  堆外内存在写磁盘或网络传输时是可以利用 zero-copy 特性，I/O 和网络传输的效率更高。
3.  堆外内存是进程间共享的，也就是说，即使 JVM 进程崩溃也不会丢失数据。这可以用来做故障恢复。

*  堆外内存劣势

1.  堆内存可以很方便地进行监控和分析，相较而言堆外内存则更加难以控制；
2.  Flink 有时可能需要短生命周期的 MemorySegment，在堆上申请开销会更小；
3.  一些操作在堆内存上会更快一些

# 二 .MemorySegment \[内存页\] #

MemorySegment ，是 Flink 的内存抽象的最小分配单元 默认情况下，一个 MemorySegmcnt 对应着一个 32KB 大小的内存块。 这块内存既可以是堆上内存( Java 的 byte 数组) .也可以是堆外内存(基于 Netty 的 DirectByteBufferr ) .

## 2.1. 属性 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

*  静态属性

/** * unsafe 类处理/转换 （堆/堆外) 内存 . * The unsafe handle for transparent memory copied (heap / off-heap). * * */
        @SuppressWarnings("restriction")
        protected static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
    
        /** * * 二进制字节数组的起始索引，相对于字节数组对象而言 。 * * The beginning of the byte array contents, relative to the byte array object. * */
        @SuppressWarnings("restriction")
        protected static final long BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
    
        /** * 标记字节顺序的常量 * 判断是否为 LittleEndian 模式的字节存储顺序，若不是，就是 BigEndian 模式 。 * 因为这是一个布尔常量，JIT编译器可以很好地利用它来积极地消除不适用的代码路径 * * Constant that flags the byte order. * * Because this is a boolean constant, the JIT compiler can * use this well to aggressively eliminate the non-applicable code paths. */
        private static final boolean LITTLE_ENDIAN =
                (ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN);

*  基础属性

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>sun.misc.Unsafe UNSAFE</td> 
   <td>unsafe 类处理/转换 （堆/堆外) 内存 .</td> 
  </tr> 
  <tr> 
   <td>BYTE_ARRAY_BASE_OFFSET</td> 
   <td>二进制字节数组的起始索引，相对于字节数组对象而言 。</td> 
  </tr> 
  <tr> 
   <td>boolean LITTLE_ENDIAN</td> 
   <td>标记字节顺序的常量</td> 
  </tr> 
  <tr> 
   <td>byte[] heapMemory</td> 
   <td>堆内存引用</td> 
  </tr> 
  <tr> 
   <td>long address</td> 
   <td>堆外内存地址</td> 
  </tr> 
  <tr> 
   <td>long addressLimit</td> 
   <td>标识地址结束位置 (address +size)</td> 
  </tr> 
  <tr> 
   <td>int size</td> 
   <td>内存段的字节数。 : 32M = 32 * 1024 = 32768</td> 
  </tr> 
  <tr> 
   <td>Object owner</td> 
   <td>memory segment的所有人</td> 
  </tr> 
 </tbody> 
</table>

/** * 堆内存引用 * * 如果 MemeorySegment 使用堆上内存，则表示一个堆上的字节数组 (byle[] ) . * 如果 MemeorySegment 使用堆外内存，则为null 。 * * * * The heap byte array object relative to which we access the memory. * * <p>Is non-<tt>null</tt> if the memory is on the heap, and is <tt>null</tt>, if the memory is * off the heap. * * If we have this buffer, we must never void this reference, or the memory * segment will point to undefined addresses outside the heap and may in out-of-order execution * cases cause segmentation faults. */
        protected final byte[] heapMemory;
    
        /** * 堆外内存地址 * 如果 heapMemory 为null , 这就代表一个堆外内存的绝对地址. * * The address to the data, relative to the heap memory byte array. * If the heap memory byte array is <tt>null</tt>, this becomes an absolute memory address outside the heap. */
        protected long address;
    
        /** * 标识地址结束位置 (address +size) * * The address one byte after the last addressable byte, i.e. <tt>address + size</tt> while the * segment is not disposed. */
        protected final long addressLimit;
    
        /** * 内存段的字节数。 : 32M = 32 * 1024 = 32768 * * The size in bytes of the memory segment. * */
        protected final int size;
    
        /** * memory segment的所有人 * Optional owner of the memory segment. * */
        private final Object owner;

## 2.2. 构造方法 ##

有两种构造方法的实现. 基于堆上内存和堆外内存.

*  基于堆内存创建MemorySegment

/** * Creates a new memory segment that represents the memory of the byte array. * * <p>Since the byte array is backed by on-heap memory, this memory segment holds its * data on heap. The buffer must be at least of size 8 bytes. * * @param buffer The byte array whose memory is represented by this memory segment. */
    	MemorySegment(byte[] buffer, Object owner) { 
    		if (buffer == null) { 
    			throw new NullPointerException("buffer");
    		}
    
    		this.heapMemory = buffer;
    		this.address = BYTE_ARRAY_BASE_OFFSET;
    		this.size = buffer.length;
    		this.addressLimit = this.address + this.size;
    		this.owner = owner;
    	}

*  基于堆外内存创建MemorySegment

/** * Creates a new memory segment that represents the memory at the absolute address given * by the pointer. * * @param offHeapAddress The address of the memory represented by this memory segment. * @param size The size of this memory segment. */
    	MemorySegment(long offHeapAddress, int size, Object owner) { 
    		if (offHeapAddress <= 0) { 
    			throw new IllegalArgumentException("negative pointer or size");
    		}
    		if (offHeapAddress >= Long.MAX_VALUE - Integer.MAX_VALUE) { 
    			// this is necessary to make sure the collapsed checks are safe against numeric overflows
    			throw new IllegalArgumentException("Segment initialized with too large address: " + offHeapAddress
    					+ " ; Max allowed address is " + (Long.MAX_VALUE - Integer.MAX_VALUE - 1));
    		}
    
    		this.heapMemory = null;
    		this.address = offHeapAddress;
    		this.addressLimit = this.address + size;
    		this.size = size;
    		this.owner = owner;
    	}

## 2.3. 方法 ##

MemorySegment 同时也提供了对二进制数据进行读取和写人的方法。对于java基本数据类型，如char 、 int 、 long 等， MemorySegment 内置了方法，可以直按返回或者写人数据，对于其他类型 ， 读取二进制数组 byte \[\]后进行反序列化，序列化为二进制数组 byte \[ \]后写人。

这里就拿getLong 进行举例 :

根据操作系统的区别,分别调用getLongLittleEndian或者getLongBigEndian方法.

/** * Reads a long integer value (64bit, 8 bytes) from the given position, in little endian byte * order. This method's speed depends on the system's native byte order, and it is possibly * slower than {@link #getLong(int)}. For most cases (such as transient storage in memory or * serialization for I/O and network), it suffices to know that the byte order in which the * value is written is the same as the one in which it is read, and {@link #getLong(int)} is the * preferable choice. * * @param index The position from which the value will be read. * @return The long value at the given position. * @throws IndexOutOfBoundsException Thrown, if the index is negative, or larger than the * segment size minus 8. */
        public final long getLongLittleEndian(int index) { 
            if (LITTLE_ENDIAN) { 
                return getLong(index);
            } else { 
                return Long.reverseBytes(getLong(index));
            }
        }
    
        /** * Reads a long integer value (64bit, 8 bytes) from the given position, in big endian byte * order. This method's speed depends on the system's native byte order, and it is possibly * slower than {@link #getLong(int)}. For most cases (such as transient storage in memory or * serialization for I/O and network), it suffices to know that the byte order in which the * value is written is the same as the one in which it is read, and {@link #getLong(int)} is the * preferable choice. * * @param index The position from which the value will be read. * @return The long value at the given position. * @throws IndexOutOfBoundsException Thrown, if the index is negative, or larger than the * segment size minus 8. */
        public final long getLongBigEndian(int index) { 
            if (LITTLE_ENDIAN) { 
                return Long.reverseBytes(getLong(index));
            } else { 
                return getLong(index);
            }
        }

最终调用的是getLong方法

/** * Reads a long value (64bit, 8 bytes) from the given position, in the system's native byte * order. This method offers the best speed for long integer reading and should be used unless a * specific byte order is required. In most cases, it suffices to know that the byte order in * which the value is written is the same as the one in which it is read (such as transient * storage in memory, or serialization for I/O and network), making this method the preferable * choice. * * @param index The position from which the value will be read. * @return The long value at the given position. * @throws IndexOutOfBoundsException Thrown, if the index is negative, or larger than the * segment size minus 8. */
        public final long getLong(int index) { 
            final long pos = address + index;
            if (index >= 0 && pos <= addressLimit - 8) { 
                // 这是能够在一个实现中同时操作对内存和堆外内存的关键
                // sun.misc.Unsafe 的一些方法会根据对象引用表现出不同的行为
                // 在 reference 不为 null 的情况下，则会取该对象的地址，加上后面的 offset，从相对地址处取出 8 字节；
                // 而在 reference 为 null 的情况下，则 offset 就是要操作的绝对地址。
                // 所以，通过控制对象引用的值，就可以灵活地管理堆外内存和堆内存。
                //
                //既然 HybridMemorySegment 可以同时管理堆内存和堆外内存，为什么还需要 HeapMemorySegment 呢？
                // 这是因为假如所有的 MemorySegment 都是在堆上分配的，
                // 使用 HeapMemorySegment 相比于 HybridMemorySegment 会有更好的性能。
                // 但实际上，由于 Flink 中 Network buffer 使用的 MemorySegment 一定是在堆外分配的，
                // HeapMemorySegment 在 Flink 中已经不会再使用了，
                // 具体可以参考 FLINK-7310 always use the HybridMemorySegment。
                //
                //MemorySegment 通常不直接构造，而是通过 MemorySegmentFactory 来创建
                return UNSAFE.getLong(heapMemory, pos);
            } else if (address > addressLimit) { 
                throw new IllegalStateException("segment has been freed");
            } else { 
                // index is in fact invalid
                throw new IndexOutOfBoundsException();
            }
        }

## 2.4. 实现类HybridMemorySegment ##

MemorySegment有两个具体的子类：  
**HeapMemorySegment** 和 **HybridMemorySegment**。  
前者是用于分配堆内存，后者用来分配堆外内存和堆内存的。

在早期版本中，由于 MemorySegment 是只基于堆内存的，因而只需要提供一种类型的 MemorySegment 实现即可；  
而在引入对堆外内存的支持后，按一般的思路是应该在新增一个基于堆外内存的实现即可。但是，这里涉及到一个 JIT 优化的性能问题。在只有一种类型的 MemorySegment 的情况下，通过 Class Hierarchy Analysis (CHA)，JIT 编译器能够确定方法调用的具体实现，因而方法调用可以通过去虚化（de-virtualized）和内联（inlined）来提升性能。而一旦有了两种类型的实现，在同时使用两种类型的 MemorySegment 的情况下，JIT 编译器就无法进行优化，这大概会导致 2.7 倍的性能差异。

因而 Flink 做了这两种优化：  
1）确保只有一种 MemorySegment 的实现被加载；  
2）提供一种能同时处理管理堆内存和堆外内存的 MemorySegment 实现，从而保证频繁调用的 MemorySegment 能够被 JIT 优化。

详细的解释和性能的评测可以参考 Flink 团队的文章[Off-heap Memory in Apache Flink and the curious JIT compiler][]。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

HybridMemorySegment继承自MemorySegment , 是Flink唯一使用的MemorySegment内存管理类(HeapMemorySegment不适用). 内存可以是on-heap, off-heap direct or off-heap unsafe ，都是由此类处理的。这个类专门处理 堆内存的字节访问和字节复制调用，同时重用MemorySegment中的the multi-byte type accesses 和cross-segment 操作。

**MemorySegment通常不手动分配, 而是使用 MemorySegmentFactory进行管理**

### 2.4.1. 属性 ###

/** * 堆外内存的封装 * * 包装堆外内存的直接字节缓冲区。 * 此内存段保存对该缓冲区的引用，因此只要此内存段存在，内存就不会被释放。 * * The direct byte buffer that wraps the off-heap memory. * This memory segment holds a reference * to that buffer, so as long as this memory segment lives, the memory will not be released. */
        @Nullable private ByteBuffer offHeapBuffer;
    
        @Nullable private final Runnable cleaner;
    
        /** * 当underlying memory 为 unsafe 时，不允许包装。 * * 可以主动释放不安全的内存，而不进行引用计数。 * 因此，从包装好的缓冲区进行访问（可能没有意识到内存的释放）可能是有风险的。 * * * Wrapping is not allowed when the underlying memory is unsafe. * * Unsafe memory can be actively released, without reference counting. * * Therefore, access from wrapped buffers, * which may not be aware of the releasing of memory, could be risky. */
        private final boolean allowWrap;

### 2.4.2. 构造方法 ###

/** * Creates a new memory segment that represents the memory backing the given direct byte buffer. * * Note that the given ByteBuffer must be direct {@link * java.nio.ByteBuffer#allocateDirect(int)}, otherwise this method with throw an * IllegalArgumentException. * * <p>The memory segment references the given owner. * * @param buffer The byte buffer whose memory is represented by this memory segment. * @param owner The owner references by this memory segment. * @param allowWrap Whether wrapping {@link ByteBuffer}s from the segment is allowed. * @param cleaner The cleaner to be called on free segment. * @throws IllegalArgumentException Thrown, if the given ByteBuffer is not direct. */
        HybridMemorySegment(
                @Nonnull ByteBuffer buffer,
                @Nullable Object owner,
                boolean allowWrap,
                @Nullable Runnable cleaner) { 
            super(getByteBufferAddress(buffer), buffer.capacity(), owner);
            this.offHeapBuffer = buffer;
            this.allowWrap = allowWrap;
            this.cleaner = cleaner;
        }

### 2.4.3. free ###

@Override
        public void free() { 
            super.free();
            if (cleaner != null) { 
                cleaner.run();
            }
            offHeapBuffer = null; // to enable GC of unsafe memory
        }

### 2.4.4. warp ###

@Override
        public ByteBuffer wrap(int offset, int length) { 
            if (!allowWrap) { 
                throw new UnsupportedOperationException(
                        "Wrap is not supported by this segment. This usually indicates that the underlying memory is unsafe, thus transferring of ownership is not allowed.");
            }
            return wrapInternal(offset, length);
        }
    
    
    
       private ByteBuffer wrapInternal(int offset, int length) { 
            if (address <= addressLimit) { 
                
                if (heapMemory != null) { 
                    // 代表是堆上 内存
                    return ByteBuffer.wrap(heapMemory, offset, length);
                } else { 
                    try { 
                        //包装 堆外内存...
                        ByteBuffer wrapper = offHeapBuffer.duplicate();
                        wrapper.limit(offset + length);
                        wrapper.position(offset);
                        return wrapper;
                    } catch (IllegalArgumentException e) { 
                        throw new IndexOutOfBoundsException();
                    }
                }
            } else { 
                throw new IllegalStateException("segment has been freed");
            }
        }

### 2.4.5. get/put相关 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

# 三. 内存页 #

MemorySegment是Flink内存分配的最小单元， 对于跨MemorySegment保存的数据， 如果需要上层的使用者， 需要考虑所有的  
细节， 非常烦琐， 所以Flink又抽象了一层， 叫作内存页。

内存页是MemorySegment之上的数据访问视图， **数据读取抽象为DataInputView，数据写入抽象为DataOutputView**。 有了这一层， 上层使用者无须关心MemorySegment的细节， 该层会自动处理跨MemorySegment的读取和写入 .

## 3.1. DataInputView ##

### 3.1.1. 前言 ###

DataInputView 是一个接口, 这个接口定义了一些内存上的视图，可以用来顺序地读取内存的内容。  
该视图通常由一个或多个\{@link org.apache.flink.core.memory.MemorySegment\}支持。  
DataInputView继承了`java.io.DataInput`接口. 提供了从二进制流中读取不同数据类型的方法

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

### 3.1.2. AbstractPagedInputView ###

AbstractPagedInputView 是一个抽象类实现了DataInputView接口. 是由多个内存页支持的所有输入视图的基类。  
这个基类包含所有的解码方法来读取一个页面的数据并检测页面边界的交叉。  
具体的子类必须实现一旦越过边界就提供下一个内存页的方法。

#### 3.1.2.1 属性 ####

// 当前操作的MemorySegment
        private MemorySegment currentSegment;
    
        // 每segment开头要跳过的字节数
        // the number of bytes to skip at the beginning of each segment
        protected final int  headerLength;
    
        // 当前segment中的偏移量
        // the offset in the current segment
        private int positionInSegment;
    
        // 切换到下一个segment之前的当前segment中的limit
        // the limit in the current segment before switching to the next
        private int limitInSegment;
    
        // 可重用的字节缓冲区，用于utf-8解码
        // reusable byte buffer for utf-8 decoding
        private byte[] utfByteBuffer;
        // 可重用的char缓冲区，用于utf-8解码
        // reusable char buffer for utf-8 decoding
        private char[] utfCharBuffer;

#### 3.1.2.2 构造方法 ####

/** * 创建一个从给定segment开始的新视图。 * 输入直接在给定页面的header 之后开始。 * 如果header大小为零，则从头开始。 * 指定的初始限制描述了在视图必须前进到下一个segment之前，最多可以从当前segment读取哪个位置数据。 * * Creates a new view that starts with the given segment. * * The input starts directly after the header of the given page. * * If the header size is zero, it starts at the beginning. * * The specified initial limit describes up to which position data may be read from the current segment, before the view must advance to the next segment. * * @param initialSegment The memory segment to start reading from. * @param initialLimit The position one after the last valid byte in the initial segment. * @param headerLength The number of bytes to skip at the beginning of each segment for the * header. This length must be the same for all memory segments. */
        protected AbstractPagedInputView(
                MemorySegment initialSegment, int initialLimit, int headerLength) { 
            this.headerLength = headerLength;
            this.positionInSegment = headerLength;
            seekInput(initialSegment, headerLength, initialLimit);
        }
    
        /** * Creates a new view that is initially not bound to a memory segment. This constructor is * typically for views that always seek first. * * <p>WARNING: The view is not readable until the first call to either {@link #advance()}, or to * {@link #seekInput(MemorySegment, int, int)}. * * @param headerLength The number of bytes to skip at the beginning of each segment for the * header. */
        protected AbstractPagedInputView(int headerLength) { 
            this.headerLength = headerLength;
        }
    
        /** * Sets the internal state of the view such that the next bytes will be read from the given * memory segment, starting at the given position. The memory segment will provide bytes up to * the given limit position. * * @param segment The segment to read the next bytes from. * @param positionInSegment The position in the segment to start reading from. * @param limitInSegment The limit in the segment. When reached, the view will attempt to switch * to the next segment. */
        protected void seekInput(MemorySegment segment, int positionInSegment, int limitInSegment) { 
            this.currentSegment = segment;
            this.positionInSegment = positionInSegment;
            this.limitInSegment = limitInSegment;
        }

#### 3.1.2.3 Page 管理相关 ####

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>public MemorySegment getCurrentSegment()</td> 
   <td>获取当前MemorySegment</td> 
  </tr> 
  <tr> 
   <td>public int getCurrentPositionInSegment()</td> 
   <td>获取当前segment中的偏移量</td> 
  </tr> 
  <tr> 
   <td>public int getCurrentSegmentLimit()</td> 
   <td>获取切换到下一个segment之前的当前segment中的limit</td> 
  </tr> 
  <tr> 
   <td>protected abstract MemorySegment nextSegment(MemorySegment current)</td> 
   <td>具体子类实现页面交叉的方法。<br> 当前页已用完并且需要新页才能继续阅读时，将调用此方法。<br>如果没有更多page 可用，则此方法必须抛出{@link EOFException}。</td> 
  </tr> 
  <tr> 
   <td>protected abstract int getLimitForSegment(MemorySegment segment)</td> 
   <td>获取从给定内存segment读取字节的限制。<br>此方法必须返回给定内存segment中最后一个有效字节之后的字节位置。 <br> 当到达此方法返回的位置时，视图将尝试切换到下一个内存segment。</td> 
  </tr> 
  <tr> 
   <td>public void advance()</td> 
   <td>将视图转换到下一个内存segment。读数将在下一个segment的header之后继续。<br>此方法使用{@link #nextSegment（MemorySegment）}和 {@link #getLimitForSegment（MemorySegment）}来获取下一个segment并设置其限制。</td> 
  </tr> 
  <tr> 
   <td>public int getHeaderLength()</td> 
   <td>获取header的长度</td> 
  </tr> 
  <tr> 
   <td>protected void seekInput(MemorySegment segment, int positionInSegment, int limitInSegment)</td> 
   <td>初始化使用</td> 
  </tr> 
  <tr> 
   <td>protected void clear()</td> 
   <td>clear相关</td> 
  </tr> 
 </tbody> 
</table>

/** * Gets the memory segment that will be used to read the next bytes from. If the segment is * exactly exhausted, meaning that the last byte read was the last byte available in the * segment, then this segment will not serve the next bytes. The segment to serve the next bytes * will be obtained through the {@link #nextSegment(MemorySegment)} method. * * @return The current memory segment. */
        public MemorySegment getCurrentSegment() { 
            return this.currentSegment;
        }
    
        /** * Gets the position from which the next byte will be read. If that position is equal to the * current limit, then the next byte will be read from next segment. * * @return The position from which the next byte will be read. * @see #getCurrentSegmentLimit() */
        public int getCurrentPositionInSegment() { 
            return this.positionInSegment;
        }
    
        /** * Gets the current limit in the memory segment. This value points to the byte one after the * last valid byte in the memory segment. * * @return The current limit in the memory segment. * @see #getCurrentPositionInSegment() */
        public int getCurrentSegmentLimit() { 
            return this.limitInSegment;
        }
    
        /** * 具体子类实现页面交叉的方法。 * 当前页已用完并且需要新页才能继续阅读时，将调用此方法。 * 如果没有更多页面可用，则此方法必须抛出{@link EOFException}。 * * The method by which concrete subclasses realize page crossing. * * This method is invoked when the current page is exhausted and a new page is required to continue the reading. * * If no further page is available, this method must throw an {@link EOFException}. * * @param current The current page that was read to its limit. May be {@code null}, if this * method is invoked for the first time. * @return The next page from which the reading should continue. May not be {@code null}. If the * input is exhausted, an {@link EOFException} must be thrown instead. * @throws EOFException Thrown, if no further segment is available. * @throws IOException Thrown, if the method cannot provide the next page due to an I/O related * problem. */
        protected abstract MemorySegment nextSegment(MemorySegment current)
                throws EOFException, IOException;
    
        /** * 获取从给定内存segment读取字节的限制。 * 此方法必须返回给定内存segment中最后一个有效字节之后的字节位置。 * 当到达此方法返回的位置时，视图将尝试切换到下一个内存segment。 * * Gets the limit for reading bytes from the given memory segment. * * This method must return the position of the byte after the last valid byte in the given memory segment. * * When the position returned by this method is reached, the view will attempt to switch to the next memory segment. * * @param segment The segment to determine the limit for. * @return The limit for the given memory segment. */
        protected abstract int getLimitForSegment(MemorySegment segment);
    
        /** * 将视图转换到下一个内存segment。 * 读数将在下一个segment的header之后继续。 * 此方法使用{@link #nextSegment（MemorySegment）}和 * {@link #getLimitForSegment（MemorySegment）}来获取下一个段并设置其限制。 * * Advances the view to the next memory segment. * * The reading will continue after the header of the next segment. * * This method uses {@link #nextSegment(MemorySegment)} and {@link #getLimitForSegment(MemorySegment)} to get the next segment and set its limit. * * @throws IOException Thrown, if the next segment could not be obtained. * @see #nextSegment(MemorySegment) * @see #getLimitForSegment(MemorySegment) */
        public void advance() throws IOException { 
            doAdvance();
        }
    
        protected void doAdvance() throws IOException { 
            // 注意：
            // 此代码确保在EOF的情况下，我们保持在相同的位置，
            // 以便EOF是可复制的（如果nextSegment抛出可复制的EOFEException）
    
            // note: this code ensures that in case of EOF, we stay at the same position such that
            // EOF is reproducible (if nextSegment throws a reproducible EOFException)
            this.currentSegment = nextSegment(this.currentSegment);
            this.limitInSegment = getLimitForSegment(this.currentSegment);
            this.positionInSegment = this.headerLength;
        }
    
        /** @return header length. */
        public int getHeaderLength() { 
            return headerLength;
        }
    
        /** * Sets the internal state of the view such that the next bytes will be read from the given * memory segment, starting at the given position. The memory segment will provide bytes up to * the given limit position. * * @param segment The segment to read the next bytes from. * @param positionInSegment The position in the segment to start reading from. * @param limitInSegment The limit in the segment. When reached, the view will attempt to switch * to the next segment. */
        protected void seekInput(MemorySegment segment, int positionInSegment, int limitInSegment) { 
            this.currentSegment = segment;
            this.positionInSegment = positionInSegment;
            this.limitInSegment = limitInSegment;
        }
    
        /** * Clears the internal state of the view. After this call, all read attempts will fail, until * the {@link #advance()} or {@link #seekInput(MemorySegment, int, int)} method have been * invoked. */
        protected void clear() { 
            this.currentSegment = null;
            this.positionInSegment = this.headerLength;
            this.limitInSegment = headerLength;
        }

#### 3.1.2.4 读取相关方法 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]

## 3.2. DataOutputView ##

## 3.3. 内存页的使用 ##

adfdsf

参考资料:

Flink内核原理与实现 - 冯飞  
https://www.jianshu.com/p/dd7da2f1ac59

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221021/71b8a1cc4f934097a54e4677b3d220bb.png
[Off-heap Memory in Apache Flink and the curious JIT compiler]: https://flink.apache.org/news/2015/09/16/off-heap-memory.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221021/dd537074e8884ae49d7bdc60895f7b2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221021/ca177aa924c84e6098d801d5dcce6bbe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: /images/20221021/c0ff10a3c669482fb5e3c7ad3da50994.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]: /images/20221021/063c71c43e314aafb4e8d8ea1793ff4b.png