一、案例
序列化代码截图.png
zk-providers.png
报错截图.png
- 从图一可以看出provider端提供了两个服务，并且
JsonDemoService指定序列化协议为fastjson；DemoService指定序列化协议为hessian2； - 从图二可以看出两个服务注册到ZK上的URL中序列化协议都是对的，证明通过
@DubboService(parameters = {"serialization", "xxx"})这种方式指定序列化协议是生效的； - 从图三可以看出调用
JsonDemoService报错了，原因是使用了hessian2协议进行序列化，但是JsonDemoRequest没有实现Serializable；
到这里就发现一个问题，provider中JsonDemoService明明指定序列化协议为fastjson，并且从注册到ZK上的URL貌似也能印证这一点，但是为什么consumer调用时却使用了hessian2协议？？？带着这个问题我们来分析下dubbo provider和consumer之间交互时序列化协议的实现原理。
二、解析
涉及序列化的环节 (1).png
下面对这六个节点分别进行分析
1、服务发布
服务导出.png
以上是服务端启动服务导出时涉及到序列化的粗略流程
- dubbo启动的入口
DubboBootstrap.start中会先初始化一些基础组件，例如，配置中心相关组件、事件监听、元数据相关组件等等，这些组件最终都会被缓存到ConfigManager.configsCache中(Dubbo中凡是实现AbstractConfig接口的类，在被Spring初始化之后，最终都会被缓存到configManager中)，其中就包含ServiceBean，@DubboService(parameters = {"serialization", "xxx"})中parameters也会保存在ServiceBean.parameters中。
image.png -
exportServices()方法，它是服务发布核心逻辑的入口，会遍历ConfigManager中每一个ServiceBean进行导出。 -
ServiceConfig.doExportUrlsFor1Protocol主要分为两部分，一部分是组装服务的 URL，另一部分就是服务发布。其中组装服务URL时会把配置的ServiceBean.parameters拼接到URL中，即&serialization=fastjson。服务发布的部分是"registry://" 协议，会通过RegistryProtocol.export实现。
// JsonDemoService服务URL
dubbo://192.168.1.15:20880/org.apache.dubbo.demo.JsonDemoService?anyhost=true
&application=dubbo-demo-annotation-provider
&bind.ip=192.168.1.15
&bind.port=20880
&deprecated=false
&dubbo=2.0.2
&dynamic=true
&generic=false
&interface=org.apache.dubbo.demo.JsonDemoService
&methods=sayByeBye
&pid=39221
&release=
&serialization=fastjson
&side=provider
×tamp=1655450646368
-
RegistryProtocol.export主要也是分为两部分，一部分是通过具体的DubboProtocol创建服务；另一部分就是通过SPI接口RegistryFactory获取到具体的Registry把服务注册到注册中心(ZK或Nacos等)。 -
DubboProtocol.openServer创建服务传入的参数只有一个URL，最终会创建Netty ServerBootstrap并经过层层包装成ProtocolServer，然后URL会被封装到Netty ChannelHandler并注册到Netty ChannelPipeline上(这部分下面会讲到)。创建的Server会缓存在serverMap中，其中key是当前服务的IP+dubbo协议端口号192.168.1.15:20880 -> {DubboProtocolServer@3845}。也就是说同一个dubbo端口只会创建一个Server，并且Server对应的URL为该端口上第一个被扫描到的服务的URL。
/**
* <host:port, ProtocolServer>
* 记录了全部的 ProtocolServer 实例，其中的 Key 是 host 和 port 组成的字符串，
* Value 是监听该地址的 ProtocolServer。ProtocolServer 就是对 RemotingServer 的一层简单封装，表示一个服务端
*/
protected final Map<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();
private void openServer(URL url) {
String key = url.getAddress();
boolean isServer = url.getParameter(IS_SERVER_KEY, true);
// 只有服务端才能创建 ProtocolServer 并对外服务
if (isServer) {
// 检查是否已有 ProtocolServer 在监听 URL 指定的地址
ProtocolServer server = serverMap.get(key);
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) {
serverMap.put(key, createServer(url));
}
}
} else {
// 如果已有ProtocolServer实例，则尝试根据URL信息重置ProtocolServer
server.reset(url);
}
}
}
- 服务注册在
register方法中，其中参数registeredProviderUrl就是最终注册到注册中心的dubbo协议URL，最终调用ZK zkClient.create方法向 Zookeeper 注册服务;
private void register(URL registryUrl, URL registeredProviderUrl) {
Registry registry = registryFactory.getRegistry(registryUrl);
registry.register(registeredProviderUrl);
}
public void doRegister(URL url) {
try {
zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
2、客户端引用
服务引用.png
以上是服务引用时涉及到序列化的粗略流程
DubboBootstrap.start()方法除了会调用exportServices()方法完成服务发布之外，还会调用referServices()方法完成服务引用，其中会遍历扫描到的所有ReferenceConfig进行服务引用(主要就是创建代理类)；RegistryProtocol.doRefer中，首先会根据 URL 初始化RegistryDirectory实例，然后生成 consumer协议 Subscribe URL 并进行注册，之后会把URL作为参数通过具体的Registry订阅服务；
consumer://192.168.1.15/org.apache.dubbo.demo.DemoService?application=dubbo-demo-annotation-consumer
&cluster=failfast
&dubbo=2.0.2
&init=false
&interface=org.apache.dubbo.demo.DemoService
&methods=sayHello,sayHelloAsync
&pid=47044&side=consumer
&sticky=false
×tamp=1655733277885
-
ZookeeperRegistry.doSubscribe这里以ZK为例，首先根据consumer URL生成路径，这里有3个路径providers、routers、configurators；然后针对每个路径注册子节点监听器，同时返回子节点(providers子节点就是当前引用服务的所有服务端URL)；最后因为是首次引用，会触发一次监听，其中providers监听便会创建对应的客户端NettyClient。
public void doSubscribe(final URL url, final NotifyListener listener) {
...
List<URL> urls = new ArrayList<>();
// 针对每一种category路径进行监听(providers、routers、configurators)
for (String path : toCategoriesPath(url)) {
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
// 封装监听逻辑
ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, k, toUrlsWithEmpty(url, parentPath, currentChilds)));
zkClient.create(path, false);
// 添加监听，并返回该目录当前所有的子节点，providers时为所有服务端URL
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
// 如果有子节点直接触发一次监听，urls就是服务端URL
notify(url, listener, urls);
}
-
DubboProtocol.getSharedClient创建Client传入的只有URL和连接数两个参数，最终会创建Netty Bootstrap并经过层层包装成ReferenceCountExchangeClient，然后URL会被封装到Netty ChannelHandler并注册到Netty ChannelPipeline上。
另外还一个比较重要的是创建Client分为共享连接和独享连接两种模式：
当使用独享连接的时候，对每个 Service 建立固定数量的 Client，每个 Client 维护一个底层连接。如下图所示，就是针对每个 Service 都启动了两个独享连接：
独享连接.png
当使用共享连接的时候，会区分不同的网络地址（host:port），一个地址只建立固定数量的共享连接。如下图所示，Provider 1 暴露了多个服务，Consumer 引用了 Provider 1 中的多个服务，共享连接是说 Consumer 调用 Provider 1 中的多个服务时，是通过固定数量的共享 TCP 长连接进行数据传输，这样就可以达到减少服务端连接数的目的。
共享连接.png
那怎么去创建共享连接呢？创建共享连接的实现细节是在
getSharedClient()方法中，它首先从referenceClientMap缓存（Map<String, List<ReferenceCountExchangeClient>> 类型）中查询 Key（host 和 port 拼接成的字符串）对应的共享 Client 集合，如果查找到的 Client 集合全部可用，则直接使用这些缓存的 Client，否则要创建新的 Client 来补充替换缓存中不可用的 Client。
private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
String key = url.getAddress();
List<ReferenceCountExchangeClient> clients = referenceClientMap.get(key);
...
locks.putIfAbsent(key, new Object());
synchronized (locks.get(key)) {
clients = referenceClientMap.get(key);
...
// If the clients is empty, then the first initialization is
if (CollectionUtils.isEmpty(clients)) {
clients = buildReferenceCountExchangeClientList(url, connectNum);
referenceClientMap.put(key, clients);
} else {
for (int i = 0; i < clients.size(); i++) {
ReferenceCountExchangeClient referenceCountExchangeClient = clients.get(i);
// If there is a client in the list that is no longer available, create a new one to replace him.
if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
clients.set(i, buildReferenceCountExchangeClient(url));
continue;
}
referenceCountExchangeClient.incrementAndGetCount();
}
}
locks.remove(key);
return clients;
}
}
dubbo默认如果没有设置connections参数，则使用共享连接，也就是说引用同一个host+port的所有服务默认只会创建一个Client，并且Client对应的服务端URL为第一个被扫描到的Referance对应服务的URL，该Client后续与服务端交互使用的序列化协议都会由这个服务端URL上的serialization决定。
3、客户端请求
前面两部分已经解析了服务注册、NettyServer、NettyClient上绑定的URL的由来以及序列化协议的设置。到这里已经能够解释开头案例中选错序列化协议的问题了，但是还有一个疑问就是同一个服务序列化协议是如何在服务端和客户端之间保证一致的？带着这个疑问继续分析下面几个部分。
下面几部分再分析下服务端与消费端交互过程中序列化协议是如何传递的。
消费端请求过程比较长，这里简单概括下DubboInvoker之前的流程：
上层业务 Bean 会被封装成Invoker对象，然后传入DubboProtocol.export()方法中，该 Invoker 被封装成DubboExporter，并保存到exporterMap集合中缓存。
在DubboProtocol暴露的ProtocolServer收到请求时，经过一系列解码处理，最终会到达DubboProtocol.requestHandler这个ExchangeHandler对象中，该ExchangeHandler对象会从exporterMap集合中取出请求的 Invoker，并调用其 invoke() 方法处理请求。
DubboProtocol.protocolBindingRefer()方法则会将底层的ExchangeClient集合封装成DubboInvoker，然后由上层逻辑封装成代理对象，这样业务层就可以像调用本地 Bean 一样，完成远程调用。
客户端请求.png
-
DubboInvoke.doInvoke方法中会委托ExchangeClient执行Channel的request操作，这个ExchangeClient就是服务引用时创建的ReferenceCountExchangeClient，它里面包装了服务URL和底层进行RPC通信的NettyClient。
protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
...
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
int timeout = calculateTimeout(invocation, methodName);
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
return AsyncRpcResult.newDefaultAsyncResult(invocation);
} else {
ExecutorService executor = getCallbackExecutor(getUrl(), inv);
CompletableFuture<AppResponse> appResponseFuture =
currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj);
// save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
FutureContext.getContext().setCompatibleFuture(appResponseFuture);
AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv);
result.setExecutor(executor);
return result;
}
} ...
}
-
HeaderExchangeChannel.request这里会用Request模型把请求内容RpcInvocation装饰起来，然后发送一个Request类型的消息。
public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor) throws RemotingException {
Request req = new Request();
req.setVersion(Version.getProtocolVersion());
req.setTwoWay(true);
req.setData(request);
// 创建DefaultFuture对象，可以从future中主动获得请求对应的响应信息
DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout, executor);
try {
channel.send(req);
} catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
-
NettyChannel.send这里会委托给Netty 框架中的 Channel(与当前的 Dubbo Channel 对象一一对应)执行writeAndFlush方法，看过Netty代码的话就知道这里会执行Netty的ChannelPipeline。
ChannelFuture future = channel.writeAndFlush(message);
- ChannelPipeline是在打开Netty客户端时构建的，这里会注册用于编码的
ChannelHandler->InternalEncoder(实现了Netty的MessageToByteEncoder接口)
protected void doOpen() throws Throwable {
// 创建NettyClientHandler
final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
bootstrap = new Bootstrap();
bootstrap.group(NIO_EVENT_LOOP_GROUP)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
//.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
.channel(socketChannelClass());
// 设置连接超时时间，这里使用到AbstractEndpoint中的connectTimeout字段
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(3000, getConnectTimeout()));
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 心跳请求的时间间隔
int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
ch.pipeline().addLast("negotiation", SslHandlerInitializer.sslClientHandler(getUrl(), nettyClientHandler));
}
// 通过NettyCodecAdapter创建Netty中的编解码器
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
// 注册ChannelHandler
ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
.addLast("decoder", adapter.getDecoder())
.addLast("encoder", adapter.getEncoder())
.addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
.addLast("handler", nettyClientHandler);
....
}
});
}
- InternalEncoder最终会通过SPI接口调用到
ExchangeCodec.encodeRequest方法对Request进行序列化，这里可以看到通过Channel上绑定的服务URL中设置的序列化协议获取Serialization，再获取ContentType的ID值，是一个byte类型的值，唯一确定一个算法。最后再通过位运算把这个ID值写到dubbo传输协议头的16-23位上。
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
Serialization serialization = getSerialization(channel);
// header.
// 创建16字节的字节数组
byte[] header = new byte[HEADER_LENGTH];
// set magic number.
// 设置前16位数据，也就是设置header[0]和header[1]的数据为Magic High和Magic Low
Bytes.short2bytes(MAGIC, header);
// set request and serialization flag.
// 16-23位为serialization编号，用到或运算10000000|serialization编号，例如serialization编号为11111，则为00011111
header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
...
}
4、服务端接收
服务端接收.png
-
InternalDecoder.decode跟客户端请求时候流程类似，服务端服务发布创建Netty ServerBootstrap时也会向ChannelPipeline注册用于解码的ChannelHandler->InternalDecoder(实现了Netty的MessageToByteDecoder接口)
protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) throws Exception {
// 将ByteBuf封装成统一的ChannelBuffer
ChannelBuffer message = new NettyBackedChannelBuffer(input);
// 拿到关联的Channel
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
do {
// 记录当前readerIndex的位置
int saveReaderIndex = message.readerIndex();
// 委托给Codec2进行解码
Object msg = codec.decode(channel, message);
....
} while (message.readable());
}
}
-
ExchangeCodec.decode可以看到通过位运算从客户端Request头中取出序列化协议ID，然后跟NettyChannel、data字节流一起包装成DecodeableRpcInvocation放到Request中data中，后续通过DecodeableRpcInvocation对请求数据进行反序列化。
protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
// get request id.
long id = Bytes.bytes2long(header, 4);
.....
// decode request.
Request req = new Request(id);
req.setVersion(Version.getProtocolVersion());
req.setTwoWay((flag & FLAG_TWOWAY) != 0);
if ((flag & FLAG_EVENT) != 0) {
req.setEvent(true);
}
try {
Object data;
DecodeableRpcInvocation inv;
if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) {
inv = new DecodeableRpcInvocation(channel, req, is, proto);
inv.decode();
} else {
inv = new DecodeableRpcInvocation(channel, req,
new UnsafeByteArrayInputStream(readMessageData(is)), proto);
}
data = inv;
req.setData(data);
} catch (Throwable t) {
if (log.isWarnEnabled()) {
log.warn("Decode request failed: " + t.getMessage(), t);
}
// bad request
req.setBroken(true);
req.setData(t);
}
return req;
}
-
DecodeableRpcInvocation.decode这里通过客户端传入的serializationType获取Serialization对请求数据进行反序列化。
ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
.deserialize(channel.getUrl(), input);
到这里就可以解释同一个服务序列化协议是如何在服务端和客户端之间保证一致的？这个疑问了，就是服务端并不会从绑定的NettyServer URL中获取，也不会从服务端注册到注册中心的URL中获取，而是从客户端请求中获取。
三、总结
再回到文章开头的案例，
首先总结下选错序列化协议的原因
dubbo消费端创建底层通信Client时默认使用共享连接模式，也就是说引用同一个host+port的所有服务默认只会创建一个Client，并且Client对应的服务端URL为第一个被扫描到的Referance对应服务的URL，该Client后续与服务端交互使用的序列化协议都会由这个服务端URL上的serialization决定。
针对开头的案例，由于先扫描到了DemoService(指定hessian2序列化)，就会创建一个绑定了DemoService URL的共享Client，后续调用这个host+port上所有的服务都会使用这个共享Client，也就是都会使用hessian2序列化。
然后提供两个解决方案
- 配置多个
dubbo:protocol指定不同的端口号和序列化协议，然后在注册服务时指定具体的dubbo:protocol。
<!-- 新增hessian2协议，绑定20881端口 -->
<dubbo:protocol id="dubbo2" name="dubbo" port="20881" default="false" serialization="hessian2"/>
<!-- 注册服务指定使用dubbo2协议 -->
@DubboService(protocol="dubbo2")
- 如果只有个别服务需要使用特殊的序列化协议，也可以通过配置
connections参数，让创建Client时使用独享连接模式。
@DubboService(parameters = {"serialization", "hessian2"}, connections=1)
