本文首发于 泊浮目的简书: https://www.jianshu.com/u/204...

版本	日期	备注
1.0	2021.12.20	文章首发

0. 前言 在最初接触到Flink时，是来自于业界里一些头部玩家的分享——大家会用其来处理海量数据。在这种场景下，如何避免JVM GC带来StopTheWorld带来的副作用这样的问题一直盘绕在我心头。直到用了Flink以后，阅读了相关的源码（以1.14.0为基准），终于有了一些答案。在这篇文章里也是会分享给大家。
1. JVM内存管理的不足 除了上述提到的StopTheWorld，JVM的内存管理还会带来以下问题：

• 内存浪费：一个Java对象在内存中存储时会分为三个部分：对象头、实例数据、对其填充部分。首先，32位和64位的实现中，对象头分别要占用32bit和64bit。而为了提供整体的使用效率，JVM内存中的数据不是连续存储的，而是按照8byte的整数倍进行存储。哪怕你只有1byte，会自动padding7byte。
• 缓存未命中：大家都知道CPU是有L1、2、3级缓存的，当CPU去读取内存中的数据时，会将内存中邻近的数据读到缓存中——这是程序局部性原理的一种实践手段。最近被CPU访问的数据，短期内CPU还要访问（时间）；被CPU访问的数据附近的数据，CPU短期内还要访问（空间）。但我们前面提到，Java对象在堆上存储的时候并不是连续的，所以CPU去读取JVM上的对象时，缓存的邻近内存区域数据往往不是CPU下一步计算所需要的。这时CPU只能空转等待从内存里读取数据（两者的速度不是一个量级）。如果数据恰好被swap到硬盘里，那就是难上加难了。

2. Flink的演进方案 在v0.10之前，Flink使用了堆上内存的实现。简单来说就是通过Unsafe来分配内存，并用byte数组的方式将其引用起来，应用层自己维护类型信息来获取相应的数据。但这样仍然会有问题：

• 在堆内内存过大的情况下，JVM启动时间会很长，而且Full GC会到达分钟级。
• IO效率低：堆上内存写磁盘或网络至少需要1次内存复制。

因此在v0.10后，Flink引入了堆外内存管理功能。见Jira：Add an off-heap variant of the managed memory。除了解决堆内内存的问题，还会带来一些好处：

• 堆外内存可以做成进程之间共享。这意味Flink可以做此一些方便的故障恢复。

当然，凡事都是有双面性的，缺点是：

• 分配短生命周期的对象，比起堆上内存，在堆外内存上分配开销更高。
• 堆外内存出错时排错更为复杂。

这种实现在Spark中也可以找到，它叫做MemoryPool，同时支持堆内和堆外的内存方式，具体见MemoryMode.scala；Kafka也有类似的思路——通过Java NIO的ByteBuffer来保存它的消息。
3. 源码分析 总的来说，Flink在这一块的实现是比较清晰的——和操作系统一样有内存段，也有内存页这样的数据结构。
3.1 内存段 主要实现为MemorySegment。在v1.12前MemorySegment
仅仅为一个接口，它的实现有两个HybridMemorySegment和HeapMemorySegment。在之后的发展中，大家发现HeapMemorySegment基本都没有人用了，而是都用HybridMemorySegment了，为了优化性能——避免运行时每次都去查函数表确认调用的函数，去掉了HeapMemorySegment，并将HybridMemorySegment移到了MemorySegment中——这会见带来近2.7倍的调用速度优化。：Off-heap Memory in Apache Flink and the curious JIT compiler以及Jira：Don't explicitly use HeapMemorySegment in raw format serde。
MemorySegment主要负责引用内存段，并其中数据进行读写——它对基本类型支持的很好，而复杂类型则需要外部来做序列化。具体的实现还是比较简单的，从field的声明中就可以大致看出实现了。唯一需要讲一下的是LITTLE_ENDIAN：不同的CPU架构会才不同的存储顺序——PowerPC会采用Big Endian方式，低地址存放最低有效字节；而x86会采用Little Endian方式存储数据，低地址存放最高有效字节。

文章图片

说实话，读到这个代码的时候笔者还是略震惊的，因为写Java这么多年几乎对底层的硬件是无感知的。没想到Java代码还要考虑兼容CPU架构的逻辑。

这个时候就会有同学问了，那这个MemorySegments是如何在Flink中运作的呢？我们可以看个测试用例：BinaryRowDataTest里的testPagesSer：
先是有MemorySegments，通过对应的BinaryRowWriter写入数据到RowData，再用BinaryRowDataSerializer写RowData到RandomAccessOutputView：

@Test public void testPagesSer() throws IOException { MemorySegment[] memorySegments = new MemorySegment[5]; ArrayList memorySegmentList = new ArrayList<>(); for (int i = 0; i < 5; i++) { memorySegments[i] = MemorySegmentFactory.wrap(new byte[64]); memorySegmentList.add(memorySegments[i]); }{ // multi memorySegments String str = "啦啦啦啦啦我是快乐的粉刷匠，啦啦啦啦啦我是快乐的粉刷匠，" + "啦啦啦啦啦我是快乐的粉刷匠。"; BinaryRowData row = new BinaryRowData(1); BinaryRowWriter writer = new BinaryRowWriter(row); writer.writeString(0, fromString(str)); writer.complete(); RandomAccessOutputView out = new RandomAccessOutputView(memorySegments, 64); BinaryRowDataSerializer serializer = new BinaryRowDataSerializer(1); serializer.serializeToPages(row, out); BinaryRowData mapRow = serializer.createInstance(); mapRow = serializer.mapFromPages( mapRow, new RandomAccessInputView(memorySegmentList, 64)); writer.reset(); writer.writeString(0, mapRow.getString(0)); writer.complete(); assertEquals(str, row.getString(0).toString()); BinaryRowData deserRow = serializer.deserializeFromPages( new RandomAccessInputView(memorySegmentList, 64)); writer.reset(); writer.writeString(0, deserRow.getString(0)); writer.complete(); assertEquals(str, row.getString(0).toString()); } // ignore some code }

3.2 内存页 一个MemorySegment默认对应了32KB大小的内存块。在流处理中，很容易出现超过32KB的数据，这时就需要跨MemorySegment。那么对于编写相应逻辑的人就需要持有多个MemorySegment，因此Flink提供了内存页的实现，它会持有多个MemorySegment实例，方便框架的开发人员来快速的编写Memory相关的代码，而无需关注一个个的MemorySegment。
其抽象为DataInputView和DataOutputView，分别对了数据读取和数据写入。
接下来，还是关联实际的代码看一下。我们以我们最常见的KafkaProducer使用为例：

|-- KafkaProducer#invoke //在这里指定了serializedValue \-- KeyedSerializationSchema#serializeValue //序列化record 的value

我们挑一个实现看看，以TypeInformationKeyValueSerializationSchema为例：

|-- TypeInformationKeyValueSerializationSchema#deserialize //KeyedSerializationSchema的实现类 |-- DataInputDeserializer#setBuffer // 这是DataInputView的实现，用内部的byte数组存储数据。这里很奇怪的是并没有使用MemorySegement。 |-- TypeSerializer#deserialize// 它的实现会针对不同的类型，从DataInputView里读出数据返回

其实这里的例子不太恰当。因为KeyedSerializationSchema已经被标记为了废弃。社区更建议我们使用KafkaSerializationSchema。第一个原因是因为KeyedSerializationSchema的抽象并不合适Kafka，当Kafka在Record新加字段时，是很难抽象当这个接口里的——这个接口仅仅关注了key、value以及topic。

以KafkaSerializationSchema展开的话，我们可以看典型的实现——KafkaSerializationSchemaWrapper，我们关心的地方很容找到：

@Override public ProducerRecord serialize(T element, @Nullable Long timestamp) { byte[] serialized = serializationSchema.serialize(element); final Integer partition; if (partitioner != null) { partition = partitioner.partition(element, null, serialized, topic, partitions); } else { partition = null; }final Long timestampToWrite; if (writeTimestamp) { timestampToWrite = timestamp; } else { timestampToWrite = null; }return new ProducerRecord<>(topic, partition, timestampToWrite, null, serialized); }

这个serializationSchema的声明是一个名为SerializationSchema的接口。可以看到它有大量的实现，其中很多对应了DataStream还有SQL API中的format。我们以TypeInformationSerializationSchema为例继续跟踪：

@Public public class TypeInformationSerializationSchema implements DeserializationSchema, SerializationSchema {//ignore some filed/** The serializer for the actual de-/serialization. */ private final TypeSerializer serializer; ....

又看到我们熟悉的接口TypeSerializer了。就像上面说的，它的实现会针对不同的类型，从DataInputView、DataOutputView进行互动，提供序列化和反序列化的能力。在它的方法签名中也是可以看到的：

/** * Serializes the given record to the given target output view. * * @param record The record to serialize. * @param target The output view to write the serialized data to. * @throws IOException Thrown, if the serialization encountered an I/O related error. Typically *raised by the output view, which may have an underlying I/O channel to which it *delegates. */ public abstract void serialize(T record, DataOutputView target) throws IOException; /** * De-serializes a record from the given source input view. * * @param source The input view from which to read the data. * @return The deserialized element. * @throws IOException Thrown, if the de-serialization encountered an I/O related error. *Typically raised by the input view, which may have an underlying I/O channel from which *it reads. */ public abstract T deserialize(DataInputView source) throws IOException; /** * De-serializes a record from the given source input view into the given reuse record instance * if mutable. * * @param reuse The record instance into which to de-serialize the data. * @param source The input view from which to read the data. * @return The deserialized element. * @throws IOException Thrown, if the de-serialization encountered an I/O related error. *Typically raised by the input view, which may have an underlying I/O channel from which *it reads. */ public abstract T deserialize(T reuse, DataInputView source) throws IOException; /** * Copies exactly one record from the source input view to the target output view. Whether this * operation works on binary data or partially de-serializes the record to determine its length * (such as for records of variable length) is up to the implementer. Binary copies are * typically faster. A copy of a record containing two integer numbers (8 bytes total) is most * efficiently implemented as {@code target.write(source, 8); }. * * @param source The input view from which to read the record. * @param target The target output view to which to write the record. * @throws IOException Thrown if any of the two views raises an exception. */ public abstract void copy(DataInputView source, DataOutputView target) throws IOException;

那么TypeSerializer#deserialize到底是怎么被调用到的呢？这些细节并不是这篇文章需要关心的。在这里我们展示一下调用链，有兴趣的读者可以沿着这个调用链看一下具体的代码：

|-- TypeSerializer#deserialize |-- StreamElementSerializer#deserialize |-- TypeInformationKeyValueSerializationSchema#deserialize |-- KafkaDeserializationSchema#deserialize |-- KafkaFetcher#partitionConsumerRecordsHandler //到这里已经很清楚了，这里是由FlinkKafkaConsumer new出来的对象

3.3 缓冲池 还有一个比较有意思的类是LocalBufferPool，封装了MemorySegment。一般用于网络缓冲器（NetworkBuffer），NetworkBuffer是网络交换数据的包装，当结果分区（ResultParition）开始写出数据的时候，需要向LocalBufferPool申请Buffer资源。
写入逻辑：

|-- Task#constructor //构造任务 |-- NettyShuffleEnvironment#createResultPartitionWriters // 创建用于写入结果的结果分区 |-- ResultPartitionFactory#create \-- ResultPartitionFactory#createBufferPoolFactory //在这里创建了一个简单的BufferPoolFactory |-- PipelinedResultPartition#constructor |-- BufferWritingResultPartition#constructor |-- SortMergeResultPartition#constructor or BufferWritingResultPartition#constructor |-- ResultPartition#constructor \-- ResultPartition#steup // 注册缓冲池到这个结果分区中

另外，NetworkBuffer实现了Netty的AbstractReferenceCountedByteBuf。这意味着这里采用了经典的引用计数算法，当Buffer不再被需要时，会被回收。
4. 其他 4.1 相关Flink Jira 【读Flink源码谈设计（有效管理内存之道）】以下是我在写本文时参考过的Jira列表：

• Add an off-heap variant of the managed memory：https://issues.apache.org/jir...
• Separate type specific memory segments.：https://issues.apache.org/jir...
• Investigate potential out-of-memory problems due to managed unsafe memory allocation：https://issues.apache.org/jir...
• Adjust GC Cleaner for unsafe memory and Java 11：https://issues.apache.org/jir...
• FLIP-49 Unified Memory Configuration for TaskExecutors：https://issues.apache.org/jir...
• Don't explicitly use HeapMemorySegment in raw format serde：https://issues.apache.org/jir...
• Refactor HybridMemorySegment：https://issues.apache.org/jir...
• use flink's buffers in netty：https://issues.apache.org/jir...
• Add copyToUnsafe, copyFromUnsafe and equalTo to MemorySegment：https://issues.apache.org/jir...

推荐阅读

• 

  丑小鸭的故事告诉我们什么道理？丑小鸭的故事简短版
• 

  为什么眼药水的有效期是24个月,药店老板却说只能用一个月？
• 

  mysql 事务sql mysql数据库中的事件
• 

  新冠疫苗加强针是指第三针吗
• 

  wifi5和wifi6网速差别详细介绍
• 

  Go 专栏｜接口 interface
• 

  英孚英语每日e课 英孚英语
• 

  分手后有哪些治愈你的歌？
• 

  《都挺好》感悟
• 

  这里的爱心妈咪小屋，邀您有空来坐坐！ 有空来坐坐
• 

  犯了非法低价出让国有土地使用权应当怎样处罚
• 

  南昌公积金贷款额度计算公式
• 

  连云港2023秦山岛公开水域游泳邀请赛比赛项目+年龄分组
• 

  微信如何视频会议 微信怎样视频会议
• 

  索尼n3ap属什么级别（索尼n3ap级别是多少）
• 

  饿了么APP账号忘记怎么办？详细解决方法
• 

  我来教你迅雷极速版怎么设置下载完成后关机 迅雷极速版设置下载完自动关机教程
• 

  乳腺疾病需要做哪些检查 乳腺有哪些检查
• 

  正宗烤鱼的制作方法 怎么做烤鱼
• 

  摄影大赛介绍语 摄影大赛介绍