技术文章内容聚合第一站
互联网技术教程 - 开发者的网上家园

轻松读懂 Dubbo,手写2W+文字 dubbo 原理解析(下篇)

作者:java 输入输出流关   发布日期:2021-06-04 22:06:50   浏览: 000次   来源: 编程语言  

本文由java 输入输出流关闭于2021年06月04日 22时06分50秒编辑 本文给大家主要讲解的是:轻松读懂 Dubbo,手写2W+文字 dubbo 原理解析(下篇)希望大家可以从中学习总结自己的一套方法和理解,更多关于方法的知识请关注互联网技术教程 - 蒋昊然的技术网站

e515d4f796d701107dcc7b5ac17c64b7.jpeg

01 服务发现1.1 服务发现流程

整体duubo的服务消费原理

36bb32987e4efaf393a8dd112f276f2d.jpeg

Dubbo 框架做服务消费也分为两大部分 。 第一步通过持有远程服务实例生成Invoker,这个Invoker 在客户端是核心的远程代理对象 。 第二步会把Invoker 通过动态代理转换成实现用户接口的动态代理引用 。

服务消费方引用服务的蓝色初始化链。时序图

9ff070e246d67bd5ba37996b2e112360.jpeg

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 总结
  1. 从注册中心发现引用服务:在有注册中心。通过注册中心发现提供者地址的情况下。ReferenceConfig 解析出的URL 格式为: registry://registryhost:/org.apache.registry.RegistryService?refer=URL.encode("conumerhost/com.foo.FooService?version=1.0.0") 。
  2. 通过URL 的registry://协议头识别。就会调用RegistryProtocol#refer()方法
  3. 查询提供者URL。如 dubbo://service-host/com.foo.FooService?version=1.0.0 。来获取注册中心
  4. 创建一个RegistryDirectory 实例并设置注册中心和协议
  5. 生成conusmer 连接。在consumer 目录下创建节点。向注册中心注册
  6. 注册完毕后。订阅providers。configurators。routers 等节点的数据
  7. 通过URL 的 dubbo:// 协议头识别。调用 DubboProtocol#refer() 方法。创建一个
    ExchangeClient 客户端并返回DubboInvoker 实例
  8. 由于一个服务可能会部署在多台服务器上。这样就会在providers 产生多个节点。这样也就会得到多个DubboInvoker 实例。就需要RegistryProtocol 调用Cluster 将多个服务提供者节点伪装成一个节点。并返回一个Invoker
  9. Invoker 创建完毕后。调用ProxyFactory 为服务接口生成代理对象。返回提供者引用
02 网络通信

在之前的内容中。我们分析了消费者端服务发现与提供者端服务暴露的相关内容。同时也知道消费者端通过内置的负载均衡算法获取合适的调用invoker进行远程调用。接下来我们再研究下远程调用过程即网络通信。

网络通信位于Remoting模块

  • Remoting 实现是Dubbo 协议的实现。如果你选择RMI 协议。整个Remoting 都不会用上;
  • Remoting 内部再划为 Transport 传输层 和 Exchange 信息交换层 ;
  • Transport 层只负责单向消息传输。是对Mina, Netty, Grizzly 的抽象。它也可以扩展UDP 传输;
  • Exchange 层是在传输层之上封装了Request-Response 语义;

网络通信的问题:

  • 客户端与服务端连通性问题
  • 粘包拆包问题
  • 异步多线程数据一致问题
2.1 通信协议

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序列化实现。性能很不理想。

2.3 网络通信

(1)Dubbo中的数据格式

解决socket中数据粘包拆包问题。一般有三种方式

定长协议(数据包长度一致)

定长的协议是指协议内容的长度是固定的。比如协议byte长度是50。当从网络上读取50个byte后。就进行decode解码操作。定长协议在读取或者写入时。效率比较高。因为数据缓存的大小基本都确定了。就好比数组一样。缺陷就是适应性不足。以RPC场景为例。很难估计出定长的长度是多少。

特殊结束符(数据尾:通过特殊的字符标识#)

相比定长协议。如果能够定义一个特殊字符作为每个协议单元结束的标示。就能够以变长的方式进行通信。从而在数据传输和高效之间取得平衡。比如用特殊字符 \n 。特殊结束符方式的问题是过于简单的思考了协议传输的过程。对于一个协议单元必须要全部读入才能够进行处理。除此之外必须要防止用户传输的数据不能同结束符相同。否则就会出现紊乱。

变长协议(协议头+payload模式

这种一般是自定义协议。会以定长加不定长的部分组成。其中定长的部分需要描述不定长的内容长度。dubbo就是使用这种形式的数据传输格式

ef8fa9e822e4b982c1c4b0692b11fc9c.jpeg

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的核心流程原理及其源码,我们就分析完毕了,整体流程思路不复杂,但是细节很多,要先理解其思想,还是得多花时间再仔细撸一遍。



版权保护: 本文由互联网技术教程原创,转载请保留链接: https://jhrtech.cn/java/203317.html

说点什么吧
  • 全部评论(0
    还没有评论,快来抢沙发吧!