轻松读懂 Dubbo,手写2W+文字 dubbo 原理解析(下篇)
作者:互联网
整体duubo的服务消费原理
Dubbo 框架做服务消费也分为两大部分 , 第一步通过持有远程服务实例生成Invoker,这个Invoker 在客户端是核心的远程代理对象 。 第二步会把Invoker 通过动态代理转换成实现用户接口的动态代理引用 。
服务消费方引用服务的蓝色初始化链,时序图
1.2 源码分析应用引用入口:ReferenceBean 的getObject 方法,该方法定义在Spring 的FactoryBean 接口中,ReferenceBean 实现了这个方法。
public Object getObject() throws Exception { return get(); } public synchronized T get() { // 检测 ref 是否为空,为空则通过 init 方法创建 if (ref == null) { // init 方法主要用于处理配置,以及调用 createProxy 生成代理类 init(); } return ref; }
Dubbo 提供了丰富的配置,用于调整和优化框架行为,性能等。Dubbo 在引用或导出服务时,首先会对这些配置进行检查和处理,以保证配置的正确性。
private void init() { // 创建代理类 ref = createProxy(map); }
此方法代码很长,主要完成的配置加载,检查,以及创建引用的代理对象。这里要从createProxy 开始看起。从字面意思上来看,createProxy 似乎只是用于创建代理对象的。但实际上并非如此,该方法还会调用其他方法构建以及合并Invoker 实例。具体细节如下。
private T createProxy(Map<String, String> map) { URL tmpUrl = new URL("temp", "localhost", 0, map); ........... isDvmRefer = InjvmProtocol . getlnjvmProtocol( ) . islnjvmRefer(tmpUrl) // 本地引用略 if (isJvmRefer) { } else { // 点对点调用略 if (url != null && url.length() > 0) { } else { // 加载注册中心 url List<URL> us = loadRegistries(false); if (us != null && !us.isEmpty()) { for (URL u : us) { URL monitorUrl = loadMonitor(u); if (monitorUrl != null) { map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString())); } // 添加 refer 参数到 url 中,并将 url 添加到 urls 中 urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map))); } } } // 单个注册中心或服务提供者(服务直连,下同) if (urls.size() == 1) { // 调用 RegistryProtocol 的 refer 构建 Invoker 实例 invoker = refprotocol.refer(interfaceClass, urls.get(0)); // 多个注册中心或多个服务提供者,或者两者混合 } else { List<Invoker<?>> invokers = new ArrayList<Invoker<?>>(); URL registryURL = null; // 获取所有的 Invoker for (URL url : urls) { // 通过 refprotocol 调用 refer 构建 Invoker,refprotocol 会在运行时 // 根据 url 协议头加载指定的 Protocol 实例,并调用实例的 refer 方法 invokers.add(refprotocol.refer(interfaceClass, url)); if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { registryURL = url; } } if (registryURL != null) { // 如果注册中心链接不为空,则将使用 AvailableCluster URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME); // 创建 StaticDirectory 实例,并由 Cluster 对多个 Invoker 进行合并 invoker = cluster.join(new StaticDirectory(u, invokers)); } else { invoker = cluster.join(new StaticDirectory(invokers)); } } } //省略无关代码... // 生成代理类 return (T) proxyFactory.getProxy(invoker); }
上面代码很多,不过逻辑比较清晰。
1、如果是本地调用,直接jvm 协议从内存中获取实例
2、如果只有一个注册中心,直接通过Protocol 自适应拓展类构建Invoker 实例接口
3、如果有多个注册中心,此时先根据url 构建Invoker。然后再通过Cluster 合并多个Invoker,最后调用ProxyFactory 生成代理类
(1)创建客户端
在服务消费方,Invoker 用于执行远程调用。Invoker 是由Protocol 实现类构建而来。Protocol 实现类有很多,这里分析DubboProtocol
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException { optimizeSerialization(url); // 创建 DubboInvoker DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers); invokers.add(invoker); return invoker; }
上面方法看起来比较简单,创建一个DubboInvoker。通过构造方法传入远程调用的client对象。默认情况下,Dubbo 使用NettyClient 进行通信。接下来,我们简单看一下getClients 方法的逻辑。
private ExchangeClient[] getClients(URL url) { // 是否共享连接 boolean service_share_connect = false; // 获取连接数,默认为0,表示未配置 int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0); // 如果未配置 connections,则共享连接 if (connections == 0) { service_share_connect = true; connections = 1; } ExchangeClient[] clients = new ExchangeClient[connections]; for (int i = 0; i < clients.length; i++) { if (service_share_connect) { // 获取共享客户端 clients[i] = getSharedClient(url); } else { // 初始化新的客户端 clients[i] = initClient(url); } } return clients; }
这里根据connections 数量决定是获取共享客户端还是创建新的客户端实例,getSharedClient 方法中也会调用initClient 方法,因此下面我们一起看一下这个方法。
private ExchangeClient initClient(URL url) { // 获取客户端类型,默认为 netty String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT)); //省略无关代码 ExchangeClient client; try { // 获取 lazy 配置,并根据配置值决定创建的客户端类型 if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) { // 创建懒加载 ExchangeClient 实例 client = new LazyConnectExchangeClient(url, requestHandler); } else { // 创建普通 ExchangeClient 实例 client = Exchangers.connect(url, requestHandler); } } catch (RemotingException e) { throw new RpcException("Fail to create remoting client for service..."); } return client; }
initClient 方法首先获取用户配置的客户端类型,默认为netty。下面我们分析一下Exchangers 的connect 方法。
public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { // 获取 Exchanger 实例,默认为 HeaderExchangeClient return getExchanger(url).connect(url, handler); }
如上,getExchanger 会通过SPI 加载HeaderExchangeClient 实例,这个方法比较简单,大家自己看一下吧。接下来分析HeaderExchangeClient 的实现。
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { // 这里包含了多个调用,分别如下: // 1. 创建 HeaderExchangeHandler 对象 // 2. 创建 DecodeHandler 对象 // 3. 通过 Transporters 构建 Client 实例 // 4. 创建 HeaderExchangeClient 对象 return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true); }
这里的调用比较多,我们这里重点看一下Transporters 的connect 方法。如下:
public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException { if (url == null) { throw new IllegalArgumentException("url == null"); } ChannelHandler handler; if (handlers == null || handlers.length == 0) { handler = new ChannelHandlerAdapter(); } else if (handlers.length == 1) { handler = handlers[0]; } else { // 如果 handler 数量大于1,则创建一个 ChannelHandler 分发器 handler = new ChannelHandlerDispatcher(handlers); } // 获取 Transporter 自适应拓展类,并调用 connect 方法生成 Client 实例 return getTransporter().connect(url, handler); }
如上,getTransporter 方法返回的是自适应拓展类,该类会在运行时根据客户端类型加载指定的Transporter 实现类。若用户未配置客户端类型,则默认加载NettyTransporter,并调用该类的connect 方法。如下:
public Client connect(URL url, ChannelHandler listener) throws RemotingException { // 创建 NettyClient 对象 return new NettyClient(url, listener); }
(2)注册
这里就已经创建好了NettyClient对象。关于DubboProtocol 的refer 方法就分析完了。接下来,继续分析RegistryProtocol 的refer 方法逻辑。
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { // 取 registry 参数值,并将其设置为协议头 url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY); // 获取注册中心实例 Registry registry = registryFactory.getRegistry(url); if (RegistryService.class.equals(type)) { return proxyFactory.getInvoker((T) registry, type, url); } // 将 url 查询字符串转为 Map Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY)); // 获取 group 配置 String group = qs.get(Constants.GROUP_KEY); if (group != null && group.length() > 0) { if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) { // 通过 SPI 加载 MergeableCluster 实例,并调用 doRefer 继续执行服务引用逻辑 return doRefer(getMergeableCluster(), registry, type, url); } } // 调用 doRefer 继续执行服务引用逻辑 return doRefer(cluster, registry, type, url); }
上面代码首先为url 设置协议头,然后根据url 参数加载注册中心实例。然后获取group 配置,根据group 配置决定doRefer 第一个参数的类型。这里的重点是doRefer 方法,如下:
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { // 创建 RegistryDirectory 实例 RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url); // 设置注册中心和协议 directory.setRegistry(registry); directory.setProtocol(protocol); Map<String, String> parameters = new HashMap<String, String> (directory.getUrl().getParameters()); // 生成服务消费者链接 URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters); // 注册服务消费者,在 consumers 目录下新节点 if (!Constants.ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(Constants.REGISTER_KEY, true)) { registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY, Constants.CHECK_KEY, String.valueOf(false))); } // 订阅 providers、configurators、routers 等节点数据 directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, Constants.PROVIDERS_CATEGORY + "," + Constants.CONFIGURATORS_CATEGORY + "," + Constants.ROUTERS_CATEGORY)); // 一个注册中心可能有多个服务提供者,因此这里需要将多个服务提供者合并为一个 Invoker invoker = cluster.join(directory); ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory); return invoker; }
如上,doRefer 方法创建一个RegistryDirectory 实例,然后生成服务者消费者链接,并向注册中心进行注册。注册完毕后,紧接着订阅providers、configurators、routers 等节点下的数据。完成订阅后,RegistryDirectory 会收到这几个节点下的子节点信息。由于一个服务可能部署在多台服务器上,这样就会在providers 产生多个节点,这个时候就需要Cluster 将多个服务节点合并为一个,并生成一个Invoker。
(3)创建代理对象
Invoker 创建完毕后,接下来要做的事情是为服务接口生成代理对象。有了代理对象,即可进行远程调用。代理对象生成的入口方法为ProxyFactory 的getProxy,接下来进行分析。
public <T> T getProxy(Invoker<T> invoker) throws RpcException { // 调用重载方法 return getProxy(invoker, false); } public <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException { Class<?>[] interfaces = null; // 获取接口列表 String config = invoker.getUrl().getParameter("interfaces"); if (config != null && config.length() > 0) { // 切分接口列表 String[] types = Constants.COMMA_SPLIT_PATTERN.split(config); if (types != null && types.length > 0) { interfaces = new Class<?>[types.length + 2]; // 设置服务接口类和 EchoService.class 到 interfaces 中 interfaces[0] = invoker.getInterface(); interfaces[1] = EchoService.class; for (int i = 0; i < types.length; i++) { // 加载接口类 interfaces[i + 1] = ReflectUtils.forName(types[i]); } } } if (interfaces == null) { interfaces = new Class<?>[]{invoker.getInterface(), EchoService.class}; } // 为 http 和 hessian 协议提供泛化调用支持,参考 pull request #1827 if (!invoker.getInterface().equals(GenericService.class) && generic) { int len = interfaces.length; Class<?>[] temp = interfaces; // 创建新的 interfaces 数组 interfaces = new Class<?>[len + 1]; System.arraycopy(temp, 0, interfaces, 0, len); // 设置 GenericService.class 到数组中 interfaces[len] = GenericService.class; } // 调用重载方法 return getProxy(invoker, interfaces); } public abstract <T> T getProxy(Invoker<T> invoker, Class<?>[] types);
如上,上面大段代码都是用来获取interfaces 数组的,我们继续往下看。getProxy(Invoker, Class<?>[]) 这个方法是一个抽象方法,下面我们到JavassistProxyFactory 类中看一下该方法的实现代码。
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) { // 生成 Proxy 子类(Proxy 是抽象类)。并调用 Proxy 子类的 newInstance 方法创建Proxy 实例 return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker)); }
上面代码并不多,首先是通过Proxy 的getProxy 方法获取Proxy 子类,然后创建InvokerInvocationHandler 对象,并将该对象传给newInstance 生成Proxy 实例。InvokerInvocationHandler 实现JDK 的InvocationHandler 接口,具体的用途是拦截接口类调用。下面以org.apache.dubbo.demo.DemoService 这个接口为例,来看一下该接口代理类代码大致是怎样的(忽略EchoService 接口)。
package org.apache.dubbo.common.bytecode; public class proxy0 implements org.apache.dubbo.demo.DemoService { public static java.lang.reflect.Method[] methods; private java.lang.reflect.InvocationHandler handler; public proxy0() { } public proxy0(java.lang.reflect.InvocationHandler arg0) { handler = $1; } public java.lang.String sayHello(java.lang.String arg0) { Object[] args = new Object[1]; args[0] = ($w) $1; Object ret = handler.invoke(this, methods[0], args); return (java.lang.String) ret; } }
好了,到这里代理类生成逻辑就分析完了。整个过程比较复杂,大家需要耐心看一下。
1.3 总结- 从注册中心发现引用服务:在有注册中心,通过注册中心发现提供者地址的情况下,ReferenceConfig 解析出的URL 格式为: registry://registryhost:/org.apache.registry.RegistryService?refer=URL.encode("conumerhost/com.foo.FooService?version=1.0.0") 。
- 通过URL 的registry://协议头识别,就会调用RegistryProtocol#refer()方法
- 查询提供者URL,如 dubbo://service-host/com.foo.FooService?version=1.0.0 ,来获取注册中心
- 创建一个RegistryDirectory 实例并设置注册中心和协议
- 生成conusmer 连接,在consumer 目录下创建节点,向注册中心注册
- 注册完毕后,订阅providers,configurators,routers 等节点的数据
- 通过URL 的 dubbo:// 协议头识别,调用 DubboProtocol#refer() 方法,创建一个
ExchangeClient 客户端并返回DubboInvoker 实例 - 由于一个服务可能会部署在多台服务器上,这样就会在providers 产生多个节点,这样也就会得到多个DubboInvoker 实例,就需要RegistryProtocol 调用Cluster 将多个服务提供者节点伪装成一个节点,并返回一个Invoker
- Invoker 创建完毕后,调用ProxyFactory 为服务接口生成代理对象,返回提供者引用
在之前的内容中,我们分析了消费者端服务发现与提供者端服务暴露的相关内容,同时也知道消费者端通过内置的负载均衡算法获取合适的调用invoker进行远程调用。接下来我们再研究下远程调用过程即网络通信。
网络通信位于Remoting模块:
- Remoting 实现是Dubbo 协议的实现,如果你选择RMI 协议,整个Remoting 都不会用上;
- Remoting 内部再划为 Transport 传输层 和 Exchange 信息交换层 ;
- Transport 层只负责单向消息传输,是对Mina, Netty, Grizzly 的抽象,它也可以扩展UDP 传输;
- Exchange 层是在传输层之上封装了Request-Response 语义;
网络通信的问题:
- 客户端与服务端连通性问题
- 粘包拆包问题
- 异步多线程数据一致问题
dubbo内置,dubbo协议 ,rmi协议,hessian协议,http协议,webservice协议,thrift协议,rest协议,grpc协议,memcached协议,redis协议等10种通讯协议。各个协议特点如下
dubbo协议
Dubbo 缺省协议采用单一长连接和NIO 异步通讯,适合于小数据量大并发的服务调用,以及服务消费者机器数远大于服务提供者机器数的情况。
- 缺省协议,使用基于mina 1.1.7 和hessian 3.2.1 的tbremoting 交互。
- 连接个数:单连接
- 连接方式:长连接
- 传输协议:TCP
- 传输方式:NIO 异步传输
- 序列化:Hessian 二进制序列化
- 适用范围:传入传出参数数据包较小(建议小于100K),消费者比提供者个数多,单一消费者无法压满提供者,尽量不要用dubbo 协议传输大文件或超大字符串。
- 适用场景:常规远程服务方法调用
rmi协议
RMI 协议采用JDK 标准的 java.rmi.* 实现,采用阻塞式短连接和JDK 标准序列化方式。
- 连接个数:多连接
- 连接方式:短连接
- 传输协议:TCP
- 传输方式:同步传输
- 序列化:Java 标准二进制序列化
- 适用范围:传入传出参数数据包大小混合,消费者与提供者个数差不多,可传文件。
- 适用场景:常规远程服务方法调用,与原生RMI服务互操作
hessian协议
Hessian 协议用于集成Hessian 的服务,Hessian 底层采用Http 通讯,采用Servlet 暴露服务,
Dubbo 缺省内嵌Jetty 作为服务器实现。
Dubbo 的Hessian 协议可以和原生Hessian 服务互操作,即:提供者用Dubbo 的Hessian 协议暴露服务,消费者直接用标准Hessian 接口调用或者提供方用标准Hessian 暴露服务,消费方用Dubbo 的Hessian 协议调用。
- 连接个数:多连接
- 连接方式:短连接
- 传输协议:HTTP
- 传输方式:同步传输
- 序列化:Hessian二进制序列化
- 适用范围:传入传出参数数据包较大,提供者比消费者个数多,提供者压力较大,可传文件。
- 适用场景:页面传输,文件传输,或与原生hessian服务互操作
http协议
基于HTTP 表单的远程调用协议,采用Spring 的HttpInvoker 实现
- 连接个数:多连接
- 连接方式:短连接
- 传输协议:HTTP
- 传输方式:同步传输
- 序列化:表单序列化
- 适用范围:传入传出参数数据包大小混合,提供者比消费者个数多,可用浏览器查看,可用表单或URL传入参数,暂不支持传文件。
- 适用场景:需同时给应用程序和浏览器JS 使用的服务。
webservice协议
基于WebService 的远程调用协议,基于Apache CXF 实现](http://dubbo.apache.org/zh-cn/docs/user/references/protocol/webservice.html#fn2)。
可以和原生WebService 服务互操作,即:提供者用Dubbo 的WebService 协议暴露服务,消费者直接用标准WebService 接口调用,或者提供方用标准WebService 暴露服务,消费方用Dubbo 的WebService 协议调用。
- 连接个数:多连接
- 连接方式:短连接
- 传输协议:HTTP
- 传输方式:同步传输
- 序列化:SOAP 文本序列化(http + xml)
- 适用场景:系统集成,跨语言调用
thrift协议
当前dubbo 支持[1]的thrift 协议是对thrift 原生协议[2] 的扩展,在原生协议的基础上添加了一些额外的头信息,比如service name,magic number 等。
rest协议
基于标准的Java REST API——JAX-RS 2.0(Java API for RESTful Web Services的简写)实现的REST调用支持
grpc协议
Dubbo 自2.7.5 版本开始支持gRPC 协议,对于计划使用HTTP/2 通信,或者想利用gRPC 带来的Stream、反压、Reactive 编程等能力的开发者来说, 都可以考虑启用gRPC 协议。
为期望使用gRPC 协议的用户带来服务治理能力,方便接入Dubbo 体系用户可以使用Dubbo 风格的,基于接口的编程风格来定义和使用远程服务
memcached协议
基于memcached实现的RPC 协议
redis协议
基于Redis 实现的RPC 协议
2.2 序列化序列化就是将对象转成字节流,用于网络传输,以及将字节流转为对象,用于在收到字节流数据后还原成对象。序列化的优势有很多,例如安全性更好、可跨平台等。我们知道dubbo基于netty进行网络通讯,在NettyClient.doOpen() 方法中可以看到Netty的相关类
bootstrap.setPipelineFactory(new ChannelPipelineFactory() { public ChannelPipeline getPipeline() { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this); ChannelPipeline pipeline = Channels.pipeline(); pipeline.addLast("decoder", adapter.getDecoder()); pipeline.addLast("encoder", adapter.getEncoder()); pipeline.addLast("handler", nettyHandler); return pipeline; } });
然后去看NettyCodecAdapter 类最后进入ExchangeCodec类的encodeRequest方法,如下:
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException { Serialization serialization = getSerialization(channel); // header. byte[] header = new byte[HEADER_LENGTH];
是的,就是Serialization接口,默认是Hessian2Serialization序列化接口。
Dubbo序列化支持java、compactedjava、nativejava、fastjson、dubbo、fst、hessian2、kryo,protostuff其中默认hessian2。其中java、compactedjava、nativejava属于原生java的序列化。
dubbo序列化:阿里尚未开发成熟的高效java序列化实现,阿里不建议在生产环境使用它。
hessian2序列化:hessian是一种跨语言的高效二进制序列化方式。但这里实际不是原生的hessian2序列化,而是阿里修改过的,它是dubbo RPC默认启用的序列化方式。
json序列化:目前有两种实现,一种是采用的阿里的fastjson库,另一种是采用dubbo中自己实现的简单json库,但其实现都不是特别成熟,而且json这种文本序列化性能一般不如上面两种二进制序列化。
java序列化:主要是采用JDK自带的Java序列化实现,性能很不理想。
(1)Dubbo中的数据格式
解决socket中数据粘包拆包问题,一般有三种方式
定长协议(数据包长度一致)
定长的协议是指协议内容的长度是固定的,比如协议byte长度是50,当从网络上读取50个byte后,就进行decode解码操作。定长协议在读取或者写入时,效率比较高,因为数据缓存的大小基本都确定了,就好比数组一样,缺陷就是适应性不足,以RPC场景为例,很难估计出定长的长度是多少。
特殊结束符(数据尾:通过特殊的字符标识#)
相比定长协议,如果能够定义一个特殊字符作为每个协议单元结束的标示,就能够以变长的方式进行通信,从而在数据传输和高效之间取得平衡,比如用特殊字符 \n 。特殊结束符方式的问题是过于简单的思考了协议传输的过程,对于一个协议单元必须要全部读入才能够进行处理,除此之外必须要防止用户传输的数据不能同结束符相同,否则就会出现紊乱。
变长协议(协议头+payload模式)
这种一般是自定义协议,会以定长加不定长的部分组成,其中定长的部分需要描述不定长的内容长度。dubbo就是使用这种形式的数据传输格式
Dubbo 数据包分为消息头和消息体,消息头用于存储一些元信息,比如魔数(Magic),数据包类型(Request/Response),消息体长度(Data Length)等。消息体中用于存储具体的调用消息,比如方法名称,参数列表等。下面简单列举一下消息头的内容。
偏移量(Bit) 字段 取值
0 ~ 7 魔数高位 0xda00
8 ~ 15 魔数低位 0xbb
16 数据包类型 0 - Response, 1 - Request
17 调用方式 仅在第16位被设为1的情况下有效,0 - 单向调用,1 - 双向调用
18 事件标 识 0 - 当前数据包是请求或响应包,1 - 当前数据包是心跳包
19 ~23 序列化器编号 2 - Hessian2Serialization
3 - JavaSerialization
4 - CompactedJavaSerialization
6 - FastJsonSerialization
7 - NativeJavaSerialization
8 - KryoSerialization
9 - FstSerialization
24 ~31 状态 20 - OK 30 - CLIENT_TIMEOUT 31 - SERVER_TIMEOUT 40 -BAD_REQUEST 50 - BAD_RESPONSE ......
32 ~95 请求编号 共8字节,运行时生成
96 ~127 消息体长度 运行时计算
(2)消费端发送请求
/** *proxy0#sayHello(String) *—> InvokerInvocationHandler#invoke(Object, Method, Object[]) * —> MockClusterInvoker#invoke(Invocation) * —> AbstractClusterInvoker#invoke(Invocation) * —> FailoverClusterInvoker#doInvoke(Invocation, List<Invoker<T>>,LoadBalance) * —> Filter#invoke(Invoker, Invocation) // 包含多个 Filter 调用 * —> ListenerInvokerWrapper#invoke(Invocation) * —> AbstractInvoker#invoke(Invocation) * —> DubboInvoker#doInvoke(Invocation) * —> ReferenceCountExchangeClient#request(Object, int) * —> HeaderExchangeClient#request(Object, int) * —> HeaderExchangeChannel#request(Object, int) * —> AbstractPeer#send(Object) * —> AbstractClient#send(Object, boolean) * —> NettyChannel#send(Object, boolean) * —> NioClientSocketChannel#write(Object) */
dubbo消费方,自动生成代码对象如下
public class proxy0 implements ClassGenerator.DC, EchoService, DemoService { private InvocationHandler handler; public String sayHello(String string) { // 将参数存储到 Object 数组中 Object[] arrobject = new Object[]{string}; // 调用 InvocationHandler 实现类的 invoke 方法得到调用结果 Object object = this.handler.invoke(this, methods[0], arrobject); // 返回调用结果 return (String)object; } }
InvokerInvocationHandler 中的invoker 成员变量类型为MockClusterInvoker,MockClusterInvoker内部封装了服务降级逻辑。下面简单看一下:
public Result invoke(Invocation invocation) throws RpcException { Result result = null; // 获取 mock 配置值 String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), MOCK_KEY, Boolean.FALSE.toString()).trim(); if (value.length() == 0 || value.equalsIgnoreCase("false")) { // 无 mock 逻辑,直接调用其他 Invoker 对象的 invoke 方法, // 比如 FailoverClusterInvoker result = this.invoker.invoke(invocation); } else if (value.startsWith("force")) { // force:xxx 直接执行 mock 逻辑,不发起远程调用 result = doMockInvoke(invocation, null); } else { // fail:xxx 表示消费方对调用服务失败后,再执行 mock 逻辑,不抛出异常 try { result = this.invoker.invoke(invocation); } catch (RpcException e) { // 调用失败,执行 mock 逻辑 result = doMockInvoke(invocation, e); } } return result; }
考虑到前文已经详细分析过FailoverClusterInvoker,因此本节略过FailoverClusterInvoker,直接分析DubboInvoker。
public abstract class AbstractInvoker<T> implements Invoker<T> { public Result invoke(Invocation inv) throws RpcException { if (destroyed.get()) { throw new RpcException("Rpc invoker for service ..."); } RpcInvocation invocation = (RpcInvocation) inv; // 设置 Invoker invocation.setInvoker(this); if (attachment != null && attachment.size() > 0) { // 设置 attachment invocation.addAttachmentsIfAbsent(attachment); } Map<String, String> contextAttachments = RpcContext.getContext().getAttachments(); if (contextAttachments != null && contextAttachments.size() != 0) { // 添加 contextAttachments 到 RpcInvocation#attachment 变量中 invocation.addAttachments(contextAttachments); } if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) { // 设置异步信息到 RpcInvocation#attachment 中 invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString()); } RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation); try { // 抽象方法,由子类实现 return doInvoke(invocation); } catch (InvocationTargetException e) { // ... } catch (RpcException e) { // ... } catch (Throwable e) { return new RpcResult(e); } } protected abstract Result doInvoke(Invocation invocation) throws Throwable; // 省略其他方法 }
上面的代码来自AbstractInvoker 类,其中大部分代码用于添加信息到RpcInvocation#attachment 变量中,添加完毕后,调用doInvoke 执行后续的调用。doInvoke 是一个抽象方法,需要由子类实现,下面到DubboInvoker 中看一下。
@Override protected Result doInvoke(final Invocation invocation) throws Throwable { RpcInvocation inv = (RpcInvocation) invocation; final String methodName = RpcUtils.getMethodName(invocation); //将目标方法以及版本号作为参数放入到Invocation中 inv.setAttachment(PATH_KEY, getUrl().getPath()); inv.setAttachment(VERSION_KEY, version); //获得客户端连接 ExchangeClient currentClient; //初始化invoker的时候,构建的一个远程通信连接 if (clients.length == 1) { //默认 currentClient = clients[0]; } else { //通过取模获得其中一个连接 currentClient = clients[index.getAndIncrement() % clients.length]; } try { //表示当前的方法是否存在返回值 boolean isOneway = RpcUtils.isOneway(getUrl(), invocation); int timeout = getUrl().getMethodParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT); //isOneway 为 true,表示“单向”通信 if (isOneway) {//异步无返回值 boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false); currentClient.send(inv, isSent); RpcContext.getContext().setFuture(null); return AsyncRpcResult.newDefaultAsyncResult(invocation); } else { //存在返回值 //是否采用异步 AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv); CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout); responseFuture.whenComplete((obj, t) -> { if (t != null) { asyncRpcResult.completeExceptionally(t); } else { asyncRpcResult.complete((AppResponse) obj); } }); RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult)); return asyncRpcResult; } } //省略无关代码 }
最终进入到HeaderExchangeChannel#request方法,拼装Request并将请求发送出去
public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException { if (closed) { throw new RemotingException(this.getLocalAddress(), null, "Failed tosend request " + request + ", cause: The channel " + this + " is closed!"); } // 创建请求对象 Request req = new Request(); req.setVersion(Version.getProtocolVersion()); req.setTwoWay(true); req.setData(request); DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout); try { //NettyClient channel.send(req); } catch (RemotingException e) { future.cancel(); throw e; } return future; }
请求编码如何做的?
在netty启动时,我们设置了编解码器,其中通过ExchangeCodec完成编解码工作如下:
public class ExchangeCodec extends TelnetCodec { // 消息头长度 protected static final int HEADER_LENGTH = 16; // 魔数内容 protected static final short MAGIC = (short) 0xdabb; protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0]; protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1]; protected static final byte FLAG_REQUEST = (byte) 0x80; protected static final byte FLAG_TWOWAY = (byte) 0x40; protected static final byte FLAG_EVENT = (byte) 0x20; protected static final int SERIALIZATION_MASK = 0x1f; private static final Logger logger = LoggerFactory.getLogger(ExchangeCodec.class); public Short getMagicCode() { return MAGIC; } @Override public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException { if (msg instanceof Request) { // 对 Request 对象进行编码 encodeRequest(channel, buffer, (Request) msg); } else if (msg instanceof Response) { // 对 Response 对象进行编码,后面分析 encodeResponse(channel, buffer, (Response) msg); } else { super.encode(channel, buffer, msg); } } protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException { Serialization serialization = getSerialization(channel); // 创建消息头字节数组,长度为 16 byte[] header = new byte[HEADER_LENGTH]; // 设置魔数 Bytes.short2bytes(MAGIC, header); // 设置数据包类型(Request/Response)和序列化器编号 header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId()); // 设置通信方式(单向/双向) if (req.isTwoWay()) { header[2] |= FLAG_TWOWAY; } // 设置事件标识 if (req.isEvent()) { header[2] |= FLAG_EVENT; } // 设置请求编号,8个字节,从第4个字节开始设置 Bytes.long2bytes(req.getId(), header, 4); // 获取 buffer 当前的写位置 int savedWriteIndex = buffer.writerIndex(); // 更新 writerIndex,为消息头预留 16 个字节的空间 buffer.writerIndex(savedWriteIndex + HEADER_LENGTH); ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer); // 创建序列化器,比如 Hessian2ObjectOutput ObjectOutput out = serialization.serialize(channel.getUrl(), bos); if (req.isEvent()) { // 对事件数据进行序列化操作 encodeEventData(channel, out, req.getData()); } else { // 对请求数据进行序列化操作 encodeRequestData(channel, out, req.getData(), req.getVersion()); } out.flushBuffer(); if (out instanceof Cleanable) { ((Cleanable) out).cleanup(); } bos.flush(); bos.close(); // 获取写入的字节数,也就是消息体长度 int len = bos.writtenBytes(); checkPayload(channel, len); // 将消息体长度写入到消息头中 Bytes.int2bytes(len, header, 12); // 将 buffer 指针移动到 savedWriteIndex,为写消息头做准备 buffer.writerIndex(savedWriteIndex); // 从 savedWriteIndex 下标处写入消息头 buffer.writeBytes(header); // 设置新的 writerIndex,writerIndex = 原写下标 + 消息头长度 + 消息体长度 buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len); } // 省略其他方法 }
以上就是请求对象的编码过程,该过程首先会通过位运算将消息头写入到header 数组中。然后对Request 对象的data 字段执行序列化操作,序列化后的数据最终会存储到ChannelBuffer 中。序列化操作执行完后,可得到数据序列化后的长度len,紧接着将len 写入到header 指定位置处。最后再将消息头字节数组header 写入到ChannelBuffer 中,整个编码过程就结束了。本节的最后,我们再来看一下Request 对象的data 字段序列化过程,也就是encodeRequestData 方法的逻辑,如下:
public class DubboCodec extends ExchangeCodec implements Codec2 { protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException { RpcInvocation inv = (RpcInvocation) data; // 依次序列化 dubbo version、path、version out.writeUTF(version); out.writeUTF(inv.getAttachment(Constants.PATH_KEY)); out.writeUTF(inv.getAttachment(Constants.VERSION_KEY)); // 序列化调用方法名 out.writeUTF(inv.getMethodName()); // 将参数类型转换为字符串,并进行序列化 out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes())); Object[] args = inv.getArguments(); if (args != null) for (int i = 0; i < args.length; i++) { // 对运行时参数进行序列化 out.writeObject(encodeInvocationArgument(channel, inv, i)); } // 序列化 attachments out.writeObject(inv.getAttachments()); } }
至此,关于服务消费方发送请求的过程就分析完了,接下来我们来看一下服务提供方是如何接收请求的。
(3)提供方接受请求
请求如何解码?
这里直接分析请求数据的解码逻辑,忽略中间过程,如下:
public class ExchangeCodec extends TelnetCodec { @Override public Object decode(Channel channel, ChannelBuffer buffer) throws IOException { int readable = buffer.readableBytes(); // 创建消息头字节数组 byte[] header = new byte[Math.min(readable, HEADER_LENGTH)]; // 读取消息头数据 buffer.readBytes(header); // 调用重载方法进行后续解码工作 return decode(channel, buffer, readable, header); } @Override protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException { // 检查魔数是否相等 if (readable > 0 && header[0] != MAGIC_HIGH || readable > 1 && header[1] != MAGIC_LOW) { int length = header.length; if (header.length < readable) { header = Bytes.copyOf(header, readable); buffer.readBytes(header, length, readable - length); } for (int i = 1; i < header.length - 1; i++) { if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) { buffer.readerIndex(buffer.readerIndex() - header.length + i); header = Bytes.copyOf(header, i); break; } } // 通过 telnet 命令行发送的数据包不包含消息头,所以这里 // 调用 TelnetCodec 的 decode 方法对数据包进行解码 return super.decode(channel, buffer, readable, header); } // 检测可读数据量是否少于消息头长度,若小于则立即返回 DecodeResult.NEED_MORE_INPUT if (readable < HEADER_LENGTH) { return DecodeResult.NEED_MORE_INPUT; } // 从消息头中获取消息体长度 int len = Bytes.bytes2int(header, 12); // 检测消息体长度是否超出限制,超出则抛出异常 checkPayload(channel, len); int tt = len + HEADER_LENGTH; // 检测可读的字节数是否小于实际的字节数 if (readable < tt) { return DecodeResult.NEED_MORE_INPUT; } ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len); try { // 继续进行解码工作 return decodeBody(channel, is, header); } finally { if (is.available() > 0) { try { StreamUtils.skipUnusedStream(is); } catch (IOException e) { logger.warn(e.getMessage(), e); } } } } }
上面方法通过检测消息头中的魔数是否与规定的魔数相等,提前拦截掉非常规数据包,比如通过telnet命令行发出的数据包。接着再对消息体长度,以及可读字节数进行检测。最后调用decodeBody 方法进行后续的解码工作,ExchangeCodec 中实现了decodeBody 方法,但因其子类DubboCodec 覆写了该方法,所以在运行时DubboCodec 中的decodeBody 方法会被调用。下面我们来看一下该方法的代码。
public class DubboCodec extends ExchangeCodec implements Codec2 { @Override protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException { // 获取消息头中的第三个字节,并通过逻辑与运算得到序列化器编号byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK); Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto); // 获取调用编号 long id = Bytes.bytes2long(header, 4); // 通过逻辑与运算得到调用类型,0 - Response,1 - Request if ((flag & FLAG_REQUEST) == 0) { // 对响应结果进行解码,得到 Response 对象。这个非本节内容,后面再分析 // ... } else { // 创建 Request 对象 Request req = new Request(id); req.setVersion(Version.getProtocolVersion()); // 通过逻辑与运算得到通信方式,并设置到 Request 对象中 req.setTwoWay((flag & FLAG_TWOWAY) != 0); // 通过位运算检测数据包是否为事件类型 if ((flag & FLAG_EVENT) != 0) { // 设置心跳事件到 Request 对象中 req.setEvent(Request.HEARTBEAT_EVENT); } try { Object data; if (req.isHeartbeat()) { // 对心跳包进行解码,该方法已被标注为废弃 data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is)); } else if (req.isEvent()) { // 对事件数据进行解码 data = decodeEventData(channel, deserialize(s, channel.getUrl(), is)); } else { DecodeableRpcInvocation inv; // 根据 url 参数判断是否在 IO 线程上对消息体进行解码 if (channel.getUrl().getParameter( Constants.DECODE_IN_IO_THREAD_KEY, Constants.DEFAULT_DECODE_IN_IO_THREAD)) { inv = new DecodeableRpcInvocation(channel, req, is, proto); // 在当前线程,也就是 IO 线程上进行后续的解码工作。此工作完成后, 可将 // 调用方法名、attachment、以及调用参数解析出来 inv.decode(); } else { // 仅创建 DecodeableRpcInvocation 对象,但不在当前线程上执行解 码逻辑 inv = new DecodeableRpcInvocation(channel, req, new UnsafeByteArrayInputStream(readMessageData(is)), proto); } data = inv; } // 设置 data 到 Request 对象中 req.setData(data); } catch (Throwable t) { // 若解码过程中出现异常,则将 broken 字段设为 true, // 并将异常对象设置到 Reqeust 对象中 req.setBroken(true); req.setData(t); } return req; } } }
如上,decodeBody 对部分字段进行了解码,并将解码得到的字段封装到Request 中。随后会调用DecodeableRpcInvocation 的decode 方法进行后续的解码工作。此工作完成后,可将调用方法名、attachment、以及调用参数解析出来。
调用服务
解码器将数据包解析成Request 对象后,NettyHandler 的messageReceived 方法紧接着会收到这个对象,并将这个对象继续向下传递。整个调用栈如下:
NettyServerHandler#channelRead(ChannelHandlerContext, MessageEvent) —> AbstractPeer#received(Channel, Object) —> MultiMessageHandler#received(Channel, Object) —> HeartbeatHandler#received(Channel, Object) —> AllChannelHandler#received(Channel, Object) —> ExecutorService#execute(Runnable) // 由线程池执行后续的调用逻辑=
这里我们直接分析调用栈中的分析第一个和最后一个调用方法逻辑。如下:
考虑到篇幅,以及很多中间调用的逻辑并非十分重要,所以这里就不对调用栈中的每个方法都进行分析了。这里我们直接分析最后一个调用方法逻辑。如下:
public class ChannelEventRunnable implements Runnable { private final ChannelHandler handler; private final Channel channel; private final ChannelState state; private final Throwable exception; private final Object message; @Override public void run() { // 检测通道状态,对于请求或响应消息,此时 state = RECEIVED if (state == ChannelState.RECEIVED) { try { // 将 channel 和 message 传给 ChannelHandler 对象,进行后续的调用 handler.received(channel, message); } catch (Exception e) { logger.warn("... operation error, channel is ... message is ..."); } } // 其他消息类型通过 switch 进行处理 else { switch (state) { case CONNECTED: try { handler.connected(channel); } catch (Exception e) { logger.warn("... operation error, channel is ..."); } break; case DISCONNECTED: // ... case SENT: // ... case CAUGHT: // ... default: logger.warn("unknown state: " + state + ", message is " + message); } } } }
如上,请求和响应消息出现频率明显比其他类型消息高,所以这里对该类型的消息进行了针对性判断ChannelEventRunnable 仅是一个中转站,它的run 方法中并不包含具体的调用逻辑,仅用于将参数传给其他ChannelHandler 对象进行处理,该对象类型为DecodeHandler
public class DecodeHandler extends AbstractChannelHandlerDelegate { public DecodeHandler(ChannelHandler handler) { super(handler); } @Override public void received(Channel channel, Object message) throws RemotingException { if (message instanceof Decodeable) { // 对 Decodeable 接口实现类对象进行解码 decode(message); } if (message instanceof Request) { // 对 Request 的 data 字段进行解码 decode(((Request) message).getData()); } if (message instanceof Response) { // 对 Request 的 result 字段进行解码 decode(((Response) message).getResult()); } // 执行后续逻辑 handler.received(channel, message); } private void decode(Object message) { // Decodeable 接口目前有两个实现类, // 分别为 DecodeableRpcInvocation 和 DecodeableRpcResult if (message != null && message instanceof Decodeable) { try { // 执行解码逻辑 ((Decodeable) message).decode(); } catch (Throwable e) { if (log.isWarnEnabled()) { log.warn("Call Decodeable.decode failed: " + e.getMessage(), e); } } } } }
DecodeHandler 主要是包含了一些解码逻辑,完全解码后的Request 对象会继续向后传递
public class DubboProtocol extends AbstractProtocol { public static final String NAME = "dubbo"; private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() { @Override public Object reply(ExchangeChannel channel, Object message) throws RemotingException { if (message instanceof Invocation) { Invocation inv = (Invocation) message; // 获取 Invoker 实例 Invoker<?> invoker = getInvoker(channel, inv); if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INV OKE))) { // 回调相关,忽略 } RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress()); // 通过 Invoker 调用具体的服务 return invoker.invoke(inv); } throw new RemotingException(channel, "Unsupported request: ..."); } // 忽略其他方法 } Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException { // 忽略回调和本地存根相关逻辑 // ... int port = channel.getLocalAddress().getPort(); // 计算 service key,格式为 groupName/serviceName:serviceVersion:port。比 如: // dubbo/com.alibaba.dubbo.demo.DemoService:1.0.0:20880 String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY)); // 从 exporterMap 查找与 serviceKey 相对应的 DubboExporter 对象, // 服务导出过程中会将 <serviceKey, DubboExporter> 映射关系存储到 exporterMap 集合中 DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey); if (exporter == null) throw new RemotingException(channel, "Not found exported service ..."); // 获取 Invoker 对象,并返回 return exporter.getInvoker(); } // 忽略其他方法 }
在之前课程中介绍过,服务全部暴露完成之后保存到exporterMap中。这里就是通过serviceKey获取exporter之后获取Invoker,并通过Invoker 的invoke 方法调用服务逻辑
public abstract class AbstractProxyInvoker<T> implements Invoker<T> { @Override public Result invoke(Invocation invocation) throws RpcException { try { // 调用 doInvoke 执行后续的调用,并将调用结果封装到 RpcResult 中,并 return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments())); } catch (InvocationTargetException e) { return new RpcResult(e.getTargetException()); } catch (Throwable e) { throw new RpcException("Failed to invoke remote proxy method ..."); } } protected abstract Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable; }
如上,doInvoke 是一个抽象方法,这个需要由具体的Invoker 实例实现。Invoker 实例是在运行时通过JavassistProxyFactory 创建的,创建逻辑如下:
public class JavassistProxyFactory extends AbstractProxyFactory { // 省略其他方法 @Override public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) { final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type); // 创建匿名类对象 return new AbstractProxyInvoker<T>(proxy, type, url) { @Override protected Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable { // 调用 invokeMethod 方法进行后续的调用 return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments); } }; } }
Wrapper 是一个抽象类,其中invokeMethod 是一个抽象方法。Dubbo 会在运行时通过Javassist 框架为Wrapper 生成实现类,并实现invokeMethod 方法,该方法最终会根据调用信息调用具体的服务。以DemoServiceImpl 为例,Javassist 为其生成的代理类如下。
/** Wrapper0 是在运行时生成的,大家可使用 Arthas 进行反编译 */ public class Wrapper0 extends Wrapper implements ClassGenerator.DC { public static String[] pns; public static Map pts; public static String[] mns; public static String[] dmns; public static Class[] mts0; // 省略其他方法 public Object invokeMethod(Object object, String string, Class[] arrclass, Object[] arrobject) throws InvocationTargetException { DemoService demoService; try { // 类型转换 demoService = (DemoService)object; } catch (Throwable throwable) { throw new IllegalArgumentException(throwable); } try { // 根据方法名调用指定的方法 if ("sayHello".equals(string) && arrclass.length == 1) { return demoService.sayHello((String)arrobject[0]); } } catch (Throwable throwable) { throw new InvocationTargetException(throwable); } throw new NoSuchMethodException(new StringBuffer().append("Not found method \"").append(string).append("\" in class com.alibaba.dubbo.demo.DemoService.").toString()); } }
到这里,整个服务调用过程就分析完了。最后把调用过程贴出来,如下:
ChannelEventRunnable#run() —> DecodeHandler#received(Channel, Object) —> HeaderExchangeHandler#received(Channel, Object) —> HeaderExchangeHandler#handleRequest(ExchangeChannel, Request) —> DubboProtocol.requestHandler#reply(ExchangeChannel, Object) —> Filter#invoke(Invoker, Invocation) —> AbstractProxyInvoker#invoke(Invocation) —> Wrapper0#invokeMethod(Object, String, Class[], Object[]) —> DemoServiceImpl#sayHello(String)
(4)提供方返回调用结果
服务提供方调用指定服务后,会将调用结果封装到Response 对象中,并将该对象返回给服务消费方。服务提供方也是通过NettyChannel 的send 方法将Response 对象返回,这里就不在重复分析了。
(5)消费方接收调用结果
服务消费方在收到响应数据后,首先要做的事情是对响应数据进行解码,得到Response 对象。然后再将该对象传递给下一个入站处理器,这个入站处理器就是NettyHandler。接下来NettyHandler 会将这个对象继续向下传递,最后AllChannelHandler 的received 方法会收到这个对象,并将这个对象派发到线程池中。这个过程和服务提供方接收请求的过程是一样的,因此这里就不重复分析了
03 小结至此整个dubbo的核心流程原理及其源码,我们就分析完毕了,整体流程思路不复杂,但是细节很多,要先理解其思想,还是得多花时间再仔细撸一遍。
标签:Dubbo,2W,dubbo,url,Object,调用,new,序列化,public 来源: https://blog.51cto.com/u_15152535/2833582