之前我们谈到了Flink通过自主管理内存的方式来，避免了让JVM管理内存带来的一些问题。自主管理内存之后，JVM中原生的类型也就不适合使用了。因此Flink也对Java的类型进行了扩展，这就是我们本节关注的内容。

本节探讨的相关类主要位于包：org.apache.flink.types

类型的根Value

Value位于所有类型的继承链的最顶端，可以说是所有类型的根。它代指所有可被序列化为Flink二进制表示的类型。该接口本身并不提供任何接口方法，但它继承自两个接口。下图是它的继承关系图：

从上图可以看出任何实现了Value接口的特定类型，都需要满足Value继承的两个接口的契约：

• Serializable ：标记实现该接口的类可被序列化
• IOReadableWritable ：Flink核心IO包种的接口，实现该接口用于将类的实例序列化为二进制的表示形式

IOReadableWritable提供了读写数据的write/read方法，另外IOReadableWritable对接口的实现者的一个要求是其必须有一个默认的（无参）构造器。

容器类型ListValue和MapValue

在Value下，Flink直接提供了两个抽象的容器类型：ListValue和MapValue。它们都有几个共同点：

• 容器中存储的元素的类型都是Value类型（通过泛型类型约束）
• 容器类型自身也实现了Value，也即自身也可被序列化
• 都实现了JDK Java集合框架中各自的接口（List和Map）

关于上面的第三点，Flink其实采用的是装饰器模式。比如，我们拿MapValue来举例：

它内部有一个map字段，该字段的初始化可能来自从构造方法传入的外部被装饰的Map实例，也可能是从无参构造方法中直接实例化的Map实例。而MapValue中实现的Map接口的方法，大都通过调用map的实例方法实现。ListValue的做法类似，不再赘述。

值得一提的是，它们对IOReadableWritable的read/write方法的实现。

我们先来看一下read方法的实现：

public void read(final DataInputView in) throws IOException {
    int size = in.readInt();
    this.map.clear();

    try {
        for (; size > 0; size--) {
            final K key = this.keyClass.newInstance();
            final V val = this.valueClass.newInstance();
            key.read(in);
            val.read(in);
            this.map.put(key, val);
        }
    } catch (final InstantiationException | IllegalAccessException e) {
        throw new RuntimeException(e);
    }
}

它首先读取一个整型值size，该整型值表示的是元素对的个数。然后循环读取每个key和value同时反序列化之后将其加入内部的map中。

而write方法的实现，则是序列化每个元素的过程：

public void write(final DataOutputView out) throws IOException {
        out.writeInt(this.map.size());
        for (final Entry<K, V> entry : this.map.entrySet()) {
            entry.getKey().write(out);
            entry.getValue().write(out);
        }
    }

当然，也是将map的size先写入二进制结果的头部。结构示意如下图：

下面会我们来看所有具体的类型需要实现的三个接口。

基本类型实现的接口

ResettableValue接口

该接口提供了一个方法：

void setValue(T value);

用于将一个外部的value赋值给内部的同类型的对象。

CopyableValue接口

该接口提供一些拷贝方法以方便基本类型的拷贝。其类图如下：

接口方法中，值得关注的是三个copy相关的方法。前两个：copyTo和copy都必须提供深拷贝的实现。而最后一个copy方法，提供在Flink的二进制表示层面上的拷贝（等价于对IOReadableWritable的read以及write的先后调用，但这里copy方法的优势是，中间不需要进行反序列化的过程）。

NormalizableKey接口

该接口指定了实现规范化的键(normalizable key)需要满足的契约。先来解释一下什么叫作“规范化的键”，规范化的键指一种在二进制表示的方式下可以进行逐字节比较的键。而要使两个规范化的键能够比较，首先对于同一种类型，它们的最大字节长度要是相等的。对于这个条件，通过接口方法getMaxNormalizedKeyLen来定义。它针对一种类型通常都会返回一个常数值。比如对于32位的整型，它会返回常数值4。但一个规范化的键所占用的字节数不一定要跟该类型的最大字节数相等。当它比规定的最大的字节数小时，可以认为它只是该规范化键的一种“前缀”。

两个规范化的键进行比较，但满足两个条件的其中之一后会停止：

• 所有的字节都比较完成
• 两个相同位置的字节不相等

关于比较的结果，如果在相同的位置，两个字节的值不相等则值小的一个键被认为其整个键会小于另外一个键。

除此之外该接口还提供了将实现类型的值（规范化的键）写入给定的目标字节数组中去的方法。

void copyNormalizedKey(MemorySegment memory, int offset, int len);

对于该接口，值得一提的是，如果真正需要写入的字节数小于给定的len，那么它将会被填充一些特定的字符以进行补齐。

NormalizableKey接口直接继承自Key接口，Key用来使得一个类型可以作为键以建立跟值之间的关系。并且键Key要求是可被比较的，因为它实现了Comparable接口。目前Key接口被标记为“Deprecated”的，在未来的版本中可能会被废弃。

微信扫码关注公众号：Apache_Flink

QQ扫码关注QQ群：Apache Flink学习交流群（123414680）