在大数据面前，JVM的内存结构和GC机制往往会成为掣肘

1. 对象开销：在HotSpot中，每个对象占用的内存空间必须是8的倍数。这就导致每个对象占据的内存空间包含三部分：头信息（header）、类的field所占空间、对齐需要的空间（padding，由于前两个大小不够8导致），所以只有一个bool值的对象会占据16个字节，其中头占8字节，bool本身占用一个字节、padding占7个字节

2. GC：大数据量的垃圾回收，尤其是几十甚至上百G的内存应用，对jvm的gc机制是个挑战

3. OOM：OutOfMemoryError是分布式计算框架经常会遇到的问题，当JVM中所有对象大小超过分配给JVM的内存大小时，就会fOutOfMemoryError错误，JVM崩溃，分布式框架的健壮性和性能都会受到影响。

4.Cache Miss：现代体系的cpu会有多级缓存，而加载的时候是以Cache Line为单位加载。如果能够将对象连续存储，这样就会大大降低Cache Miss。使得cpu集中处理业务，而不是空转。

接下俩会从内存管理、序列化、数据结构和算法等角度来介绍Flink的一些做法。

一、内存管理

理论上Flink的内存管理分为三部分

1. Network Buffers：这个是在TaskManager启动的时候分配的，这是一组用于缓存网络数据的内存，每个块是32K，默认分配2048个，可以通过“taskmanager.network.numberOfBuffers”修改

2. Memory Manage pool：大量的Memory Segment块，用于运行时的算法（Sort/Join/Shufflt等），这部分启动的时候就会分配。下面这段代码，根据配置文件中的各种参数来计算内存的分配方法。（heap or off-heap，这个放到下节谈），内存的分配支持预分配和lazy load，默认懒加载的方式。

3. User Code，这部分是除了Memory Manager之外的内存用于User code和TaskManager本身的数据结构。

对GC的影响

1. 减少full gc时间：因为所有常用数据都在Memory Manager里，这部分内存的生命周期是伴随TaskManager而不会被回收。其他的常用数据对象都是用户定义的数据对象，这部分会快速的被Minor GC回收

2. 减少OOM：所有的运行时的内存应用都从池化的内存中获取，而且运行时的算法可以在内存不足的时候将数据写到堆外内存，例如NormalizedKeySorter、MutableHashTable

3. 节约空间：由于自定序列化/反序列化的方法，所有的对象都以二进制的形式存储，降低消耗

4. 高效的二进制操作和缓存友好：二进制数据以定义好的格式存储，可以高效地比较与操作。另外，该二进制形式可以把相关的值，以及hash值，键值和指针等相邻地放进内存中。这使得数据结构可以对高速缓存更友好，可以从 L1/L2/L3 缓存获得性能的提升

Flink的序列化

  目前 Java 生态圈提供了众多的序列化框架：Java serialization, Kryo, Apache Avro 等等。但是 Flink 实现了自己的序列化框架。因为在 Flink 中处理的数据流通常是同一类型，由于数据集对象的类型固定，对于数据集可以只保存一份对象Schema信息，节省大量的存储空间。同时，对于固定大小的类型，也可通过固定的偏移位置存取。当我们需要访问某个对象成员变量的时候，通过定制的序列化工具，并不需要反序列化整个Java对象，而是可以直接通过偏移量，只是反序列化特定的对象成员变量。如果对象的成员变量较多时，能够大大减少Java对象的创建开销，以及内存数据的拷贝大小。

Flink支持任意的Java或是Scala类型。Flink 在数据类型上有很大的进步，不需要实现一个特定的接口（像Hadoop中的org.apache.hadoop.io.Writable），Flink 能够自动识别数据类型。Flink 通过 Java Reflection 框架分析基于 Java 的 Flink 程序 UDF (User Define Function)的返回类型的类型信息，通过 Scala Compiler 分析基于 Scala 的 Flink 程序 UDF 的返回类型的类型信息。类型信息由TypeInformation类表示，TypeInformation 支持以下几种类型：

BasicTypeInfo: 任意Java 基本类型（装箱的）或 String 类型。

BasicArrayTypeInfo: 任意Java基本类型数组（装箱的）或 String 数组。

WritableTypeInfo: 任意 Hadoop Writable 接口的实现类。

TupleTypeInfo: 任意的 Flink Tuple 类型(支持Tuple1 to Tuple25)。Flink tuples 是固定长度固定类型的Java Tuple实现。

CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples)。

PojoTypeInfo: 任意的 POJO (Java or Scala)，例如，Java对象的所有成员变量，要么是 public 修饰符定义，要么有 getter/setter 方法。

GenericTypeInfo: 任意无法匹配之前几种类型的类。

  前六种数据类型基本上可以满足绝大部分的Flink程序，针对前六种类型数据集，Flink皆可以自动生成对应的TypeSerializer，能非常高效地对数据集进行序列化和反序列化。对于最后一种数据类型，Flink会使用Kryo进行序列化和反序列化。每个TypeInformation中，都包含了serializer，类型会自动通过serializer进行序列化，然后用Java Unsafe接口写入MemorySegments。对于可以用作key的数据类型，Flink还同时自动生成TypeComparator，用来辅助直接对序列化后的二进制数据进行compare、hash等操作。对于 Tuple、CaseClass、POJO 等组合类型，其TypeSerializer和TypeComparator也是组合的，序列化和比较时会委托给对应的serializers和comparators。如下图展示 一个内嵌型的Tuple3对象的序列化过程。

Flink如何操作二进制数据

Flink会分配一块内催给对应的operator，内存会分为两部分，一部分存储真实的数据，一部分存储object的指针。

以Sort为例：

每个需要比较的对象都分为两部分pointer+key 和object，其中key是定长的key，如果不是定长（比如String）那么就取它的top n character。

   将实际的数据和指针加定长key分开存放有两个目的。第一，交换定长块（key+pointer）更高效，不用交换真实的数据也不用移动其他key和pointer。第二，这样做是缓存友好的，因为key都是连续存储在内存中的，可以大大减少 cache miss。

在交换过程中，只需要比较key就可以完成sort的过程，只有key1 == key2的情况，才需要反序列化拿出实际的对象做比较，而比较之后只需要交换对应的key而不需要交换实际的对象

MemoryManager

MemoryManager提供了两个内部类HeapMemoryPool和HybridOffHeapMemoryPool，这两个类是根据不同的内存类型，将内存池化的两个类。MemoryManager在初始化的时候引入了PageSize的概念，PageSize本质上和MemorySegment没有什么不同，其作用只是为了保持每一个Segment大小是相同的

HeapMemoryPool和Hybrid都会根据Allocate  page size预分配一些内存（预分配可能是0）

HeapMemoryPool分配的对象是byte[]， 而Hybrid分配的是ByteBuffer。在分配的时候都通过allocate方法进行分配。在allocate方法中，由于担心多线程的问题，会用一把锁做线程同步

这里要注意的是如果为预分配的模式，则从池子里面取，否则allocate new segments

MemoryPool的方法如下；

参考：

1. http://www.javamex.com/tutorials/memory/object_memory_usage.shtml

2. http://www.36dsj.com/archives/33650

3. https://flink.apache.org/news/2015/05/11/Juggling-with-Bits-and-Bytes.html

4. http://wuchong.me/blog/2016/04/29/flink-internals-memory-manage/