Kryo序列化在分布式处理情景下的问题

Kryo序列化在分布式处理情景下的问题

背景

最近在dubbo的issue列表里看到了这样一个issue,主要描述的是在某种情况下使用kryo作为序列化方式会报错,这个issue下面,另外一位来自阿里的Committer也给了问题出现的原因,这个问题还是非常有意思的,这里顺便看一下kryo的序列化方式,复现一下这个问题。

准备工作

issue下面其实已经贴出了一个demo可以用来复现上述的问题。但是在我的机器上一直不work,我就没有用他的demo了。这里我准备一个更简单的demo让大家更清楚的了解这个问题,先看看我们的类:

Server:启动的main函数代码可以从dubbo-demo下面找一个provider抄过来即可,主要看配置文件:

1
2
3
4
5
6
7
8
9
10
11
12
<!-- 这里使用kryo作为序列化方式 -->
<!-- 关于PhoneOptimizer这个类我们后面看 -->
<dubbo:protocol name="dubbo" port="20880" serialization="kryo" optimizer="dubbo.action.optimizer.PhoneOptimizer" heartbeat="10000000"/>

<dubbo:application name="demo-server"/>

<!-- 如果你要运行代码记得把zk地址改掉,或者用直连等方式 -->
<dubbo:registry address="zookeeper://xxx"/>

<!-- 暴露一个服务 -->
<bean id="phoneServiceImpl" class="dubbo.action.inter.PhoneServiceImpl"/>
<dubbo:service id="phoneService" interface="dubbo.action.inter.PhoneService" ref="phoneServiceImpl"/>

Client:代码就不看了,直接看配置:

1
2
3
4
5
6
7
8
9
10
<dubbo:application name="demo-consumer"/>

<!-- 一会再看这个optimizer,注意这里用的是UserOptimizer和上面使用的PhoneOptimizer不一样 -->
<dubbo:protocol name="dubbo" port="20881" serialization="kryo" optimizer="dubbo.action.optimizer.UserOptimizer" heartbeat="10000000"/>

<!-- 同样记得改改zk地址或者改成直连 -->
<dubbo:registry address="zookeeper://xxx"/>

<!-- 引用一个服务 -->
<dubbo:reference id="phoneService" interface="dubbo.action.inter.PhoneService" timeout="100000000"/>

PhoneService:这里只贴PhoneService的实现、Phone类还有User类,实在是太简单了:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public class PhoneServiceImpl implements PhoneService {

public Phone getPhone() {
return new Phone();
}
}

public class Phone implements Serializable {

private static final long serialVersionUID = -8578553481899481234L;

private String color = "black";

public String getColor() {
return color;
}

public void setColor(String color) {
this.color = color;
}
}

public class User implements Serializable {

private static final long serialVersionUID = 2345729265494289413L;

private String name = "xyh";

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}
}

Optimizer:这里把两个Optimizer都贴出来:

1
2
3
4
5
6
7
8
9
10
11
12
13
public class UserOptimizer implements SerializationOptimizer {

public Collection<Class> getSerializableClasses() {
return Collections.singletonList(User.class);
}
}

public class PhoneOptimizer implements SerializationOptimizer {

public Collection<Class> getSerializableClasses() {
return Collections.singletonList(Phone.class);
}
}

代码特别简单,就是简单的引用和暴露,唯一特别的地方就是我们设置了我们的optimizer,服务端用PhoneOptimizer而客户端使用UserOptimizer,并且我们使用Kryo作为序列化方式。

启动一下程序,启动Server没什么特别的,正常启动。启动Client的时候报错:

1
2
3
4
5
6
7
PhoneService ps = (PhoneService) context.getBean("phoneService");
Phone p = ps.getPhone();

错误:
Exception in thread "main" java.lang.ClassCastException: dubbo.action.domain.User cannot be cast to dubbo.action.domain.Phone
at com.alibaba.dubbo.common.bytecode.proxy0.getPhone(proxy0.java)
at dubbo.action.ServerAndConsumer.main(ServerAndConsumer.java:20)

我们明明在PhoneService里返回的是Phone类,怎么就cast exception了呢。带着这个问题我们看下源码。

源码分析——Optimizer的作用

要解决这个问题,我们要先看看Optimizer是怎么起作用的,这个部分的代码是dubbo重启维护之后,从dubbox中合并过来的,代码在DubboProtocol#optimizeSerialization

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 加载我们配置中的 optimizer="xxx.PhoneOptimezer" 这个类
Class clazz = Thread.currentThread().getContextClassLoader().loadClass(className);
if (!SerializationOptimizer.class.isAssignableFrom(clazz)) {
// throw exception...
}
SerializationOptimizer optimizer = (SerializationOptimizer) clazz.newInstance();
if (optimizer.getSerializableClasses() == null) {
return;
}
// 把我们PhoneOptimezer中的类注册到SerializableClassRegistry中
for (Class c : optimizer.getSerializableClasses()) {
SerializableClassRegistry.registerClass(c);
}
optimizers.add(className);

SerializableClassRegistry的的内容非常简单,不多说了,看下SerializableClassRegistrygetRegisteredClasses调用方,在AbstractKryoFactory里(注意Fst序列化方式中也用到了类似的地方,我们先不看,只看kryo相关):

1
2
3
for (Class clazz : SerializableClassRegistry.getRegisteredClasses()) {
kryo.register(clazz);
}

其实就是把我们的PhoneUser两个domain类注册到了kryo里。

源码分析——kryo序列化

现在我们知道了这个Optimizer的作用之后,就要看看为什么会报错了。这里我们要从客户端和服务端两个端看起,先看看服务端写出去了什么东西,再看看客户端收到了什么东西。

这里我需要跟大家说明一下kryo的两种写class信息的序列化方式:

  1. 写入的是class name,默认情况下就是这种写class的方式,性能不好,因为写入的数据比较大,无论是传输还是写入都慢。
  2. 写入的是id,对于我们注册过的类(调用过kryo.register(Class type)),我们写入的class信息是一个id,性能好,写入的数据量小,写入和传输都快。

我们下面看看我们如何出的错。

我会带着大家一起跟踪整个过程,先以debug的方式运行服务端,再以普通方式运行客户端。我们看看我们到底写出去个什么东西,我们既然要看写出去个什么东西,当然要看我们encode相关的部分,这里我们encode的是response,代码入口在DubboCountCodec#encode,我们跳过一些信息,看kryo相关的内容,入口在KryoObjectOutput#writeObject,追进去:

可以看到我们写出去的确实是一个Phone对象,我们看看writeClass方法,这个主要是把类的信息写到流里,最终走到DefaultClassResolver#writeClass里:

这里要注意,我们发现我们走到了registration.getId() != NAME这个分支里,结果我们实际上并没有向流里写什么信息,只是写了一个idoutput.writeVarInt),这是为什么,我们要追溯一下Kryo#getRegistration(java.lang.Class)这个方法,这个方法底层是从classResolverclassToRegistration这个map中获取一个Registration,那我们就看看这个classToRegistration中的值是从哪来的,看下使用这个字段的地方,我们找到了线索:
DefaultClassResolver#register方法中使用了put。那我们可以很轻松的联想到我们之前在启动dubbo的时候,注册了我们的Phone.class到kryo中。看下这个register方法:

1
2
3
4
5
6
7
8
9
10
if (registration.getId() != NAME) {
// 如果id不是-1,就把我们的registration放到一个map里,键就是id
idToRegistration.put(registration.getId(), registration);
} else if (TRACE) {
// sth...
}
classToRegistration.put(registration.getType(), registration);
if (registration.getType().isPrimitive())
classToRegistration.put(getWrapperClass(registration.getType()), registration);
return registration;

我们在使用kryo.register(Class type)这个方法时,其实默认调用的是这样的:return classResolver.register(new Registration(type, serializer, getNextRegistrationId()));,注意这里getNextRegistrationId,实际上就是从idToRegistration这个字段里取最大的id,然后加1后返回。kryo默认会注册五十多个类,都是java原生的类,比如BooleanInteger

我们写完了class信息之后,就是实际写入我们的对象的具体信息了,比如我们的color字段。这里不说了,跟我们的问题关系不大。

其实说到这里,基本上很多朋友也能猜到问题了,我们在服务端注册的Phone.class,但是在客户端,我们注册的User.class,而二者对应的id是同一个,这种情况下就会出现问题。为什么id是同一个,这里我说一下,之前我们说过,kryo会默认注册五十多个jdk原生的类,那么客户端和服务端的kryo绝对是两个对象,在这两个kryo都没有注册其他类的情况下,PhoneUser这两个类在两个kryo中的位置是一样的。

问题大揭秘

我们猜完了,就要实际看看效果了。这里我们用普通方式运行服务端,debug方式运行客户端,看看我们客户端收到数据以后,为什么报错。

上面是encode这里就是decode了,decode的就是响应信息,最终和我们kryo相关的地方在KryoObjectInput#readObject(),调用的是kryo的readClassAndObject方法:

我们看看读取出来的是什么:

果然是一个User对象。至此我们就知道为什么会class cast exception了。

后记

这里可能有人会说,你并没有复现issue中提到的场景啊,issue里说报出来的是另外一个错误。这里其实本质是一样的,如果你看得懂这篇文章,应该会明白。说白了,kryo中有个注册表,标记id - class的关系。

但是在分布式系统中,两端的这个id - class表关系一旦错误,就会出现各种问题,这是kryo在分布式系统中的局限性,下面借用一下解决这个issue的这位大神的图,一目了然:

图上所示的就是一种典型的,两个端id - class错乱的情况。另外一篇文章也提到了kryo在分布式场景中的问题,就提到了issue中的场景。

kryo初始化类的小细节

这个部分与我们的问题无关,有兴趣的可以看下,没兴趣的直接跳过即可。

kryo初始化对象很特别,使用的是ConstructorAccess这个类。我们看下ConstructorAccess#get方法,先看头部:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Class enclosingType = type.getEnclosingClass();
boolean isNonStaticMemberClass = enclosingType != null && type.isMemberClass() && !Modifier.isStatic(type.getModifiers());

String className = type.getName();
// 拼装类名,我们的场景中就是 xxx.PhoneConstructorAccess
String accessClassName = className + "ConstructorAccess";
if (accessClassName.startsWith("java."))
accessClassName = "reflectasm." + accessClassName;
Class accessClass;
AccessClassLoader loader = AccessClassLoader.get(type);
try {
// 尝试加载xxx.PhoneConstructorAccess
accessClass = loader.loadClass(accessClassName);
catch(xxx) {
xxx
}

很奇怪,居然尝试去加载一个不存在的类,我们每个类进来,都会尝试加载ClassNameConstructorAccess这个类。我们的例子中,当然不存在这个PhoneConstructorAccess类,继续看如果捕捉异常会怎样:

在捕捉到异常之后,使用ASM重新创建了一个ConstructorAccess子类,并且这个新类的newInstance方法,返回的就是我们的User(或者Phone)对象的实例。

这里我们可以看到,实际上kryo是通过一个额外的、通过ASM生成的类,来进行对象的初始化的。为什么用这种方式,不甚了解,有了解的朋友可以告知我一下。