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

版本 日期 备注
1.0 2021.12.20 文章首发
1.1 2021.12.22 错别字修正
1.2 2022.2.24 描述错误修正

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使用了堆上内存的实现。简单来说就是通过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方式存储数据，低地址存放最高有效字节。
读Flink源码谈设计：有效管理内存之道 - 图1

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

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

  1.     @Test
  2.     public void testPagesSer() throws IOException {
  3.         MemorySegment[] memorySegments = new MemorySegment[5];
  4.         ArrayList<MemorySegment> memorySegmentList = new ArrayList<>();
  5.         for (int i = 0; i < 5; i++) {
  6.             memorySegments[i] = MemorySegmentFactory.wrap(new byte[64]);
  7.             memorySegmentList.add(memorySegments[i]);
  8.         }
  10.         {
  11.             // multi memorySegments
  12.             String str = "啦啦啦啦啦我是快乐的粉刷匠，啦啦啦啦啦我是快乐的粉刷匠，" + "啦啦啦啦啦我是快乐的粉刷匠。";
  13.             BinaryRowData row = new BinaryRowData(1);
  14.             BinaryRowWriter writer = new BinaryRowWriter(row);
  15.             writer.writeString(0, fromString(str));
  16.             writer.complete();
  18.             RandomAccessOutputView out = new RandomAccessOutputView(memorySegments, 64);
  19.             BinaryRowDataSerializer serializer = new BinaryRowDataSerializer(1);
  20.             serializer.serializeToPages(row, out);
  22.             BinaryRowData mapRow = serializer.createInstance();
  23.             mapRow =
  24.                     serializer.mapFromPages(
  25.                             mapRow, new RandomAccessInputView(memorySegmentList, 64));
  26.             writer.reset();
  27.             writer.writeString(0, mapRow.getString(0));
  28.             writer.complete();
  29.             assertEquals(str, row.getString(0).toString());
  31.             BinaryRowData deserRow =
  32.                     serializer.deserializeFromPages(
  33.                             new RandomAccessInputView(memorySegmentList, 64));
  34.             writer.reset();
  35.             writer.writeString(0, deserRow.getString(0));
  36.             writer.complete();
  37.             assertEquals(str, row.getString(0).toString());
  38.         }
  39.      // ignore some code
  40.     }

3.2 内存页

一个MemorySegment默认对应了32KB大小的内存块。在流处理中，很容易出现超过32KB的数据，这时就需要跨MemorySegment。那么对于编写相应逻辑的人就需要持有多个MemorySegment，因此Flink提供了内存页的实现，它会持有多个MemorySegment实例，方便框架的开发人员来快速的编写Memory相关的代码，而无需关注一个个的MemorySegment。

其抽象为DataInputView和DataOutputView，分别对了数据读取和数据写入。

接下来，还是关联实际的代码看一下。我们以我们最常见的KafkaProducer使用为例：

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

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

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

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

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

  1.     @Override
  2.     public ProducerRecord<byte[], byte[]> serialize(T element, @Nullable Long timestamp) {
  3.         byte[] serialized = serializationSchema.serialize(element);
  4.         final Integer partition;
  5.         if (partitioner != null) {
  6.             partition = partitioner.partition(element, null, serialized, topic, partitions);
  7.         } else {
  8.             partition = null;
  9.         }
  11.         final Long timestampToWrite;
  12.         if (writeTimestamp) {
  13.             timestampToWrite = timestamp;
  14.         } else {
  15.             timestampToWrite = null;
  16.         }
  18.         return new ProducerRecord<>(topic, partition, timestampToWrite, null, serialized);
  19.     }

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

  1. @Public
  2. public class TypeInformationSerializationSchema<T>
  3.         implements DeserializationSchema<T>, SerializationSchema<T> {
  5.     //ignore some filed
  7.     /** The serializer for the actual de-/serialization. */
  8.     private final TypeSerializer<T> serializer;
  9. ....

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

  1.     /**
  2.      * Serializes the given record to the given target output view.
  3.      *
  4.      * @param record The record to serialize.
  5.      * @param target The output view to write the serialized data to.
  6.      * @throws IOException Thrown, if the serialization encountered an I/O related error. Typically
  7.      *     raised by the output view, which may have an underlying I/O channel to which it
  8.      *     delegates.
  9.      */
  10.     public abstract void serialize(T record, DataOutputView target) throws IOException;
  12.     /**
  13.      * De-serializes a record from the given source input view.
  14.      *
  15.      * @param source The input view from which to read the data.
  16.      * @return The deserialized element.
  17.      * @throws IOException Thrown, if the de-serialization encountered an I/O related error.
  18.      *     Typically raised by the input view, which may have an underlying I/O channel from which
  19.      *     it reads.
  20.      */
  21.     public abstract T deserialize(DataInputView source) throws IOException;
  23.     /**
  24.      * De-serializes a record from the given source input view into the given reuse record instance
  25.      * if mutable.
  26.      *
  27.      * @param reuse The record instance into which to de-serialize the data.
  28.      * @param source The input view from which to read the data.
  29.      * @return The deserialized element.
  30.      * @throws IOException Thrown, if the de-serialization encountered an I/O related error.
  31.      *     Typically raised by the input view, which may have an underlying I/O channel from which
  32.      *     it reads.
  33.      */
  34.     public abstract T deserialize(T reuse, DataInputView source) throws IOException;
  36.     /**
  37.      * Copies exactly one record from the source input view to the target output view. Whether this
  38.      * operation works on binary data or partially de-serializes the record to determine its length
  39.      * (such as for records of variable length) is up to the implementer. Binary copies are
  40.      * typically faster. A copy of a record containing two integer numbers (8 bytes total) is most
  41.      * efficiently implemented as {@code target.write(source, 8);}.
  42.      *
  43.      * @param source The input view from which to read the record.
  44.      * @param target The target output view to which to write the record.
  45.      * @throws IOException Thrown if any of the two views raises an exception.
  46.      */
  47.     public abstract void copy(DataInputView source, DataOutputView target) throws IOException;

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

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

3.3 缓冲池

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

写入逻辑：

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

另外，NetworkBuffer实现了Netty的AbstractReferenceCountedByteBuf。这意味着这里采用了经典的引用计数算法，当Buffer不再被需要时，会被回收。

4. 其他

4.1 相关Flink Jira

以下是我在写本文时参考过的Jira列表：