背景

这里所说的序列化器不是指实现TypeSerializer的状态序列化器，而是指flink在使用KryoSerializer序列化器时遇到kryo无法序列化的类型时，通过往kryo中注册某个序列化器类来让kryo可以序列化某个类的实例，所以这里严格意义上应该是说，怎么让Kryo支持某种自定义的类实例

Kryo序列化

我们知道flink当遇到Generic Type时，它自身内置的TypeSerializer序列化器没法序列化这些类，所以他会把这些类的处理工作都交给Kryo，也就是KryoSerializer，一般来说，Kryo是一个很强大的序列化器，他几乎可以序列化所有一般的类，但是也有一些类，比如Protobuf/Thrift等类，他是没法序列化和反序列化的，所以当你的实现类是使用了PB或者Thrift框架生成的类,抑或是Guava等集合框架类，你需要告诉Kryo如何序列化/反序列化这些类，只需要执行如下操作即可：

env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, MyCustomSerializer.class);env.getConfig().addDefaultKryoSerializer(MyCustomType.class, TBaseSerializer.class);kryo.register( Collections.singletonMap( "", "" ).getClass(), new CollectionsSingletonMapSerializer() );kryo.register( LocalDate.class, new JodaLocalDateSerializer() );

这样就可以解决Kryo遇到自定义的类时无法序列化和反序列化的问题了

彩蛋：

kryo支持绝大部分自定义类的序列化，比如Roaring64NavigableMap类，但是Kryo不会调用Roaring64NavigableMap中已有的Serializer/DeSerializer方法对Roaring64NavigableMap进行序列化和反序列化，而是对立面的字段比如long[],String[]等分别进行序列化，也就是说这种序列化/反序列化对比Roaring64NavigableMap已经提供的已有Serializer/DeSerializer方法效率和容量都要低好几倍，如果这里是一个大的性能问题，你如何做呢?

附：
kryo支持的特殊的serializer实现
https://github.com/magro/kryo-serializers