• Dubbo原理解析，彻底搞懂dubbo (下)

    服务发现

    服务发现流程

    整体duubo的服务消费原理

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

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

    源码分析应用

    引用入口:ReferenceBean 的getObject 方法，该方法定义在Spring 的FactoryBean 接口中，ReferenceBean 实现了这个方法。

    1. public Object getObject() throws Exception {
    2.    return get();
    3. }
    4. public synchronized T get() {
    5.    // 检测 ref 是否为空，为空则通过 init 方法创建
    6.    if (ref == null) {
    7.        // init 方法主要用于处理配置，以及调用 createProxy 生成代理类
    8.        init();
    9.    }
    10.    return ref;
    11. }

    Dubbo 提供了丰富的配置，用于调整和优化框架行为，性能等。Dubbo 在引用或导出服务时，首先会对这些配置进行检查和处理，以保证配置的正确性。

    1. private void init() {
    2.    // 创建代理类
    3.    ref = createProxy(map);
    4. }

    此方法代码很长，主要完成的配置加载，检查，以及创建引用的代理对象。这里要从createProxy 开始看起。从字面意思上来看，createProxy 似乎只是用于创建代理对象的。但实际上并非如此，该方法还会调用其他方法构建以及合并Invoker 实例。具体细节如下。

    1. private T createProxy(Map map) {
    2.    URL tmpUrl = new URL("temp", "localhost", 0, map);
    3. ...........
    4. isDvmRefer = InjvmProtocol . getlnjvmProtocol( ) . islnjvmRefer(tmpUrl)
    5.    // 本地引用略
    6.    if (isJvmRefer) {
    7.    } else {
    8.        // 点对点调用略
    9.        if (url != null && url.length() > 0) {
    10.            
    11.        } else {
    12.            // 加载注册中心 url
    13.            List us = loadRegistries(false);
    14.            if (us != null && !us.isEmpty()) {
    15.                for (URL u : us) {
    16.                    URL monitorUrl = loadMonitor(u);
    17.                    if (monitorUrl != null) {
    18.                        map.put(Constants.MONITOR_KEY,
    19. URL.encode(monitorUrl.toFullString()));
    20.                    }
    21.                    // 添加 refer 参数到 url 中，并将 url 添加到 urls 中
    22.                    urls.add(u.addParameterAndEncoded(Constants.REFER_KEY,
    23. StringUtils.toQueryString(map)));
    24.                }
    25.            }
    26.        }
    27.        // 单个注册中心或服务提供者(服务直连，下同)
    28.        if (urls.size() == 1) {
    29.            // 调用 RegistryProtocol 的 refer 构建 Invoker 实例
    30.            invoker = refprotocol.refer(interfaceClass, urls.get(0));
    31.        // 多个注册中心或多个服务提供者，或者两者混合
    32.        } else {
    33.            List> invokers = new ArrayList>();
    34.            URL registryURL = null;
    35.            // 获取所有的 Invoker
    36.            for (URL url : urls) {
    37.                // 通过 refprotocol 调用 refer 构建 Invoker，refprotocol 会在运行时
    38.                // 根据 url 协议头加载指定的 Protocol 实例，并调用实例的 refer 方法
    39.                invokers.add(refprotocol.refer(interfaceClass, url));
    40.                if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    41.                    registryURL = url;
    42.                }
    43.            }
    44.            if (registryURL != null) {
    45.                // 如果注册中心链接不为空，则将使用 AvailableCluster
    46.                URL u = registryURL.addParameter(Constants.CLUSTER_KEY,
    47. AvailableCluster.NAME);
    48.                // 创建 StaticDirectory 实例，并由 Cluster 对多个 Invoker 进行合并
    49.                invoker = cluster.join(new StaticDirectory(u, invokers));
    50.            } else {
    51.                invoker = cluster.join(new StaticDirectory(invokers));
    52.            }
    53.        }
    54.    }
    55.     //省略无关代码...
    56.     // 生成代理类
    57.    return (T) proxyFactory.getProxy(invoker);
    58. }

    折叠

    上面代码很多，不过逻辑比较清晰。
    1、如果是本地调用，直接jvm 协议从内存中获取实例
    2、如果只有一个注册中心，直接通过Protocol 自适应拓展类构建Invoker 实例接口
    3、如果有多个注册中心，此时先根据url 构建Invoker。然后再通过Cluster 合并多个Invoker，最后调用ProxyFactory 生成代理类

    创建客户端

    在服务消费方，Invoker 用于执行远程调用。Invoker 是由Protocol 实现类构建而来。Protocol 实现类有很多，这里分析DubboProtocol

    1. public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
    2.    optimizeSerialization(url);
    3.    // 创建 DubboInvoker
    4.    DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url,
    5. getClients(url), invokers);
    6.    invokers.add(invoker);
    7.    return invoker;
    8. }

    上面方法看起来比较简单，创建一个DubboInvoker。通过构造方法传入远程调用的client对象。默认情况下，Dubbo 使用NettyClient 进行通信。接下来，我们简单看一下getClients 方法的逻辑。

    1. private ExchangeClient[] getClients(URL url) {
    2.    // 是否共享连接
    3.    boolean service_share_connect = false;
    4. // 获取连接数，默认为0，表示未配置
    5.    int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
    6.    // 如果未配置 connections，则共享连接
    7.    if (connections == 0) {
    8.        service_share_connect = true;
    9.        connections = 1;
    10.    }
    11.    ExchangeClient[] clients = new ExchangeClient[connections];
    12.    for (int i = 0; i < clients.length; i++) {
    13.        if (service_share_connect) {
    14.            // 获取共享客户端
    15.            clients[i] = getSharedClient(url);
    16.        } else {
    17.            // 初始化新的客户端
    18.            clients[i] = initClient(url);
    19.        }
    20.    }
    21.    return clients;
    22. }

    这里根据connections 数量决定是获取共享客户端还是创建新的客户端实例，getSharedClient 方法中也会调用initClient 方法，因此下面我们一起看一下这个方法。

    1. private ExchangeClient initClient(URL url) {
    2.    // 获取客户端类型，默认为 netty
    3.    String str = url.getParameter(Constants.CLIENT_KEY,
    4. url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
    5.    //省略无关代码
    6.    ExchangeClient client;
    7.    try {
    8.        // 获取 lazy 配置，并根据配置值决定创建的客户端类型
    9.        if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
    10.            // 创建懒加载 ExchangeClient 实例
    11.            client = new LazyConnectExchangeClient(url, requestHandler);
    12.        } else {
    13.            // 创建普通 ExchangeClient 实例
    14.            client = Exchangers.connect(url, requestHandler);
    15.        }
    16.    } catch (RemotingException e) {
    17.        throw new RpcException("Fail to create remoting client for service...");
    18.    }
    19.    return client;
    20. }

    initClient 方法首先获取用户配置的客户端类型，默认为netty。下面我们分析一下Exchangers 的connect 方法。

    1. public static ExchangeClient connect(URL url, ExchangeHandler handler) throws
    2. RemotingException {
    3.    // 获取 Exchanger 实例，默认为 HeaderExchangeClient
    4.    return getExchanger(url).connect(url, handler);
    5. }

    如上，getExchanger 会通过SPI 加载HeaderExchangeClient 实例，这个方法比较简单，大家自己看一下吧。接下来分析HeaderExchangeClient 的实现。

    1. public ExchangeClient connect(URL url, ExchangeHandler handler) throws
    2. RemotingException {
    3.    // 这里包含了多个调用，分别如下：
    4.    // 1. 创建 HeaderExchangeHandler 对象
    5.    // 2. 创建 DecodeHandler 对象
    6.    // 3. 通过 Transporters 构建 Client 实例
    7.    // 4. 创建 HeaderExchangeClient 对象
    8.    return new HeaderExchangeClient(Transporters.connect(url, new
    9. DecodeHandler(new HeaderExchangeHandler(handler))), true);
    10. }

    这里的调用比较多，我们这里重点看一下Transporters 的connect 方法。如下：

    1. public static Client connect(URL url, ChannelHandler... handlers) throws
    2. RemotingException {
    3.    if (url == null) {
    4.        throw new IllegalArgumentException("url == null");
    5.        }
    6.    ChannelHandler handler;
    7.    if (handlers == null || handlers.length == 0) {
    8.        handler = new ChannelHandlerAdapter();
    9.    } else if (handlers.length == 1) {
    10.        handler = handlers[0];
    11.    } else {
    12.        // 如果 handler 数量大于1，则创建一个 ChannelHandler 分发器
    13.        handler = new ChannelHandlerDispatcher(handlers);
    14.    }
    15.    
    16.    // 获取 Transporter 自适应拓展类，并调用 connect 方法生成 Client 实例
    17.    return getTransporter().connect(url, handler);
    18. }

    如上，getTransporter 方法返回的是自适应拓展类，该类会在运行时根据客户端类型加载指定的Transporter 实现类。若用户未配置客户端类型，则默认加载NettyTransporter，并调用该类的connect 方法。如下：

    1. public Client connect(URL url, ChannelHandler listener) throws RemotingException
    2. {
    3.    // 创建 NettyClient 对象
    4.    return new NettyClient(url, listener);
    5. }

    注册

    这里就已经创建好了NettyClient对象。关于DubboProtocol 的refer 方法就分析完了。接下来，继续分析RegistryProtocol 的refer 方法逻辑。

    1. public  Invoker refer(Class type, URL url) throws RpcException {
    2.    // 取 registry 参数值，并将其设置为协议头
    3.    url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY,
    4. Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
    5.    // 获取注册中心实例
    6.    Registry registry = registryFactory.getRegistry(url);
    7.    if (RegistryService.class.equals(type)) {
    8.        return proxyFactory.getInvoker((T) registry, type, url);
    9.    }
    10.    // 将 url 查询字符串转为 Map
    11.    Map qs =
    12. StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
    13.    // 获取 group 配置
    14.    String group = qs.get(Constants.GROUP_KEY);
    15.    if (group != null && group.length() > 0) {
    16.        if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
    17.                || "*".equals(group)) {
    18.            // 通过 SPI 加载 MergeableCluster 实例，并调用 doRefer 继续执行服务引用逻辑
    19.            return doRefer(getMergeableCluster(), registry, type, url);
    20.        }
    21.    }
    22.    
    23.    // 调用 doRefer 继续执行服务引用逻辑
    24.    return doRefer(cluster, registry, type, url);
    25. }

    上面代码首先为url 设置协议头，然后根据url 参数加载注册中心实例。然后获取group 配置，根据group 配置决定doRefer 第一个参数的类型。这里的重点是doRefer 方法，如下：

    1. private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T>
    2. type, URL url) {
    3.    // 创建 RegistryDirectory 实例
    4.    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    5.    // 设置注册中心和协议
    6.    directory.setRegistry(registry);
    7.    directory.setProtocol(protocol);
    8.    Map<String, String> parameters = new HashMap<String, String>
    9. (directory.getUrl().getParameters());
    10.    // 生成服务消费者链接
    11.    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL,
    12. parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
    13.    // 注册服务消费者，在 consumers 目录下新节点
    14.    if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
    15.            && url.getParameter(Constants.REGISTER_KEY, true)) {
    16.        registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY,
    17. Constants.CONSUMERS_CATEGORY,
    18.                Constants.CHECK_KEY, String.valueOf(false)));
    19.    }
    20.    // 订阅 providers、configurators、routers 等节点数据
    21.    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
    22.            Constants.PROVIDERS_CATEGORY
    23.                    + "," + Constants.CONFIGURATORS_CATEGORY
    24.                    + "," + Constants.ROUTERS_CATEGORY));
    25.    // 一个注册中心可能有多个服务提供者，因此这里需要将多个服务提供者合并为一个
    26.    Invoker invoker = cluster.join(directory);
    27.    ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl,
    28. directory);
    29.    return invoker;
    30. }

    如上，doRefer 方法创建一个RegistryDirectory 实例，然后生成服务者消费者链接，并向注册中心进行注册。注册完毕后，紧接着订阅providers、configurators、routers 等节点下的数据。完成订阅后，RegistryDirectory 会收到这几个节点下的子节点信息。由于一个服务可能部署在多台服务器上，这样就会在providers 产生多个节点，这个时候就需要Cluster 将多个服务节点合并为一个，并生成一个Invoker。

    创建代理对象

    Invoker 创建完毕后，接下来要做的事情是为服务接口生成代理对象。有了代理对象，即可进行远程调用。代理对象生成的入口方法为ProxyFactory 的getProxy，接下来进行分析。

    1. public  T getProxy(Invoker invoker) throws RpcException {
    2.    // 调用重载方法
    3.    return getProxy(invoker, false);
    4. }
    5. public  T getProxy(Invoker invoker, boolean generic) throws RpcException {
    6.    Class[] interfaces = null;
    7.    // 获取接口列表
    8.     String config = invoker.getUrl().getParameter("interfaces");
    9.    if (config != null && config.length() > 0) {
    10.        // 切分接口列表
    11.        String[] types = Constants.COMMA_SPLIT_PATTERN.split(config);
    12.        if (types != null && types.length > 0) {
    13.            interfaces = new Class[types.length + 2];
    14.            // 设置服务接口类和 EchoService.class 到 interfaces 中
    15.            interfaces[0] = invoker.getInterface();
    16.            interfaces[1] = EchoService.class;
    17.            for (int i = 0; i < types.length; i++) {
    18.                // 加载接口类
    19.                interfaces[i + 1] = ReflectUtils.forName(types[i]);
    20.            }
    21.        }
    22.    }
    23.    if (interfaces == null) {
    24.        interfaces = new Class[]{invoker.getInterface(), EchoService.class};
    25.    }
    26.    // 为 http 和 hessian 协议提供泛化调用支持，参考 pull request #1827
    27.    if (!invoker.getInterface().equals(GenericService.class) && generic) {
    28.        int len = interfaces.length;
    29.        Class[] temp = interfaces;
    30.        // 创建新的 interfaces 数组
    31.        interfaces = new Class[len + 1];
    32.        System.arraycopy(temp, 0, interfaces, 0, len);
    33.        // 设置 GenericService.class 到数组中
    34.        interfaces[len] = GenericService.class;
    35.    }
    36.    // 调用重载方法
    37.    return getProxy(invoker, interfaces);
    38. }
    39. public abstract  T getProxy(Invoker invoker, Class[] types);

    如上，上面大段代码都是用来获取interfaces 数组的，我们继续往下看。getProxy(Invoker, Class[]) 这个方法是一个抽象方法，下面我们到JavassistProxyFactory 类中看一下该方法的实现代码。

    1. public  T getProxy(Invoker invoker, Class[] interfaces) {
    2.    // 生成 Proxy 子类（Proxy 是抽象类）。并调用 Proxy 子类的 newInstance 方法创建Proxy 实例
    3.    return (T) Proxy.getProxy(interfaces).newInstance(new
    4. InvokerInvocationHandler(invoker));
    5. }

    上面代码并不多，首先是通过Proxy 的getProxy 方法获取Proxy 子类，然后创建InvokerInvocationHandler 对象，并将该对象传给newInstance 生成Proxy 实例。InvokerInvocationHandler 实现JDK 的InvocationHandler 接口，具体的用途是拦截接口类调用。下面以org.apache.dubbo.demo.DemoService 这个接口为例，来看一下该接口代理类代码大致是怎样的（忽略EchoService 接口）。

    1. package org.apache.dubbo.common.bytecode;
    2. public class proxy0 implements org.apache.dubbo.demo.DemoService {
    3.    public static java.lang.reflect.Method[] methods;
    4.    private java.lang.reflect.InvocationHandler handler;
    5.    public proxy0() {
    6.    }
    7.    public proxy0(java.lang.reflect.InvocationHandler arg0) {
    8.        handler = $1;
    9.    }
    10.    public java.lang.String sayHello(java.lang.String arg0) {
    11.        Object[] args = new Object[1];
    12.        args[0] = ($w) $1;
    13.        Object ret = handler.invoke(this, methods[0], args);
    14.        return (java.lang.String) ret;
    15.    }
    16. }

    好了，到这里代理类生成逻辑就分析完了。整个过程比较复杂，大家需要耐心看一下。

    总结

    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 为服务接口生成代理对象，返回提供者引用

    网络通信

    在之前的内容中，我们分析了消费者端服务发现与提供者端服务暴露的相关内容，同时也知道消费者端通过内置的负载均衡算法获取合适的调用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/us
    er/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 协议

    序列化

    序列化就是将对象转成字节流，用于网络传输，以及将字节流转为对象，用于在收到字节流数据后还原成对象。序列化的优势有很多，例如安全性更好、可跨平台等。我们知道dubbo基于netty进行网络通讯，在NettyClient.doOpen() 方法中可以看到Netty的相关类

    1. bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    2.    public ChannelPipeline getPipeline() {
    3.        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(),
    4. NettyClient.this);
    5.        ChannelPipeline pipeline = Channels.pipeline();
    6.        pipeline.addLast("decoder", adapter.getDecoder());
    7.        pipeline.addLast("encoder", adapter.getEncoder());
    8.        pipeline.addLast("handler", nettyHandler);
    9.        return pipeline;
    10.    }
    11. });

    然后去看NettyCodecAdapter 类最后进入ExchangeCodec类的encodeRequest方法，如下：

    1. protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request
    2. req) throws IOException {
    3.        Serialization serialization = getSerialization(channel);
    4.        // header.
    5.        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序列化实现，性能很不理想。

    网络通信

    Dubbo中的数据格式

    解决socket中数据粘包拆包问题，一般有三种方式
    定长协议（数据包长度一致）
    定长的协议是指协议内容的长度是固定的，比如协议byte长度是50，当从网络上读取50个byte后，就进行decode解码操作。定长协议在读取或者写入时，效率比较高，因为数据缓存的大小基本都确定了，就好比数组一样，缺陷就是适应性不足，以RPC场景为例，很难估计出定长的长度是多少。
    特殊结束符（数据尾：通过特殊的字符标识#）
    相比定长协议，如果能够定义一个特殊字符作为每个协议单元结束的标示，就能够以变长的方式进行通信，从而在数据传输和高效之间取得平衡，比如用特殊字符 \n 。特殊结束符方式的问题是过于简单的思考了协议传输的过程，对于一个协议单元必须要全部读入才能够进行处理，除此之外必须要防止用户传输的数据不能同结束符相同，否则就会出现紊乱。
    变长协议（协议头+payload模式）
    这种一般是自定义协议，会以定长加不定长的部分组成，其中定长的部分需要描述不定长的内容长度。
    dubbo就是使用这种形式的数据传输格式

    dubbo源码解析
    docx 4星 超过85%的资源 2.94MB
    下载

    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 消息体长度 运行时计算

    消费端发送请求

    1. /**
    2. *proxy0#sayHello(String)
    3. *—> InvokerInvocationHandler#invoke(Object, Method, Object[])
    4. *   —> MockClusterInvoker#invoke(Invocation)
    5. *     —> AbstractClusterInvoker#invoke(Invocation)
    6. *       —> FailoverClusterInvoker#doInvoke(Invocation, List>,LoadBalance)
    7. *         —> Filter#invoke(Invoker, Invocation)  // 包含多个 Filter 调用
    8. *          —> ListenerInvokerWrapper#invoke(Invocation)
    9. *             —> AbstractInvoker#invoke(Invocation)
    10. *              —> DubboInvoker#doInvoke(Invocation)
    11. *                 —> ReferenceCountExchangeClient#request(Object, int)
    12. *                   —> HeaderExchangeClient#request(Object, int)
    13. *                     —> HeaderExchangeChannel#request(Object, int)
    14. *                       —> AbstractPeer#send(Object)
    15. *                         —> AbstractClient#send(Object, boolean)
    16. *                          —> NettyChannel#send(Object, boolean)
    17. *                             —> NioClientSocketChannel#write(Object)
    18. */

    dubbo消费方，自动生成代码对象如下

    1. public class proxy0 implements ClassGenerator.DC, EchoService, DemoService {
    2.    private InvocationHandler handler;
    3.    public String sayHello(String string) {
    4.        // 将参数存储到 Object 数组中
    5.        Object[] arrobject = new Object[]{string};
    6.        // 调用 InvocationHandler 实现类的 invoke 方法得到调用结果
    7.        Object object = this.handler.invoke(this, methods[0], arrobject);
    8.        // 返回调用结果
    9.        return (String)object;
    10.    }
    11. }

    InvokerInvocationHandler 中的invoker 成员变量类型为MockClusterInvoker，MockClusterInvoker内部封装了服务降级逻辑。下面简单看一下：

    1. public Result invoke(Invocation invocation) throws RpcException {
    2.        Result result = null;
    3. // 获取 mock 配置值
    4.        String value =
    5. directory.getUrl().getMethodParameter(invocation.getMethodName(), MOCK_KEY,
    6. Boolean.FALSE.toString()).trim();
    7.        if (value.length() == 0 || value.equalsIgnoreCase("false")) {
    8.             // 无 mock 逻辑，直接调用其他 Invoker 对象的 invoke 方法，
    9.            // 比如 FailoverClusterInvoker
    10.            result = this.invoker.invoke(invocation);
    11.        } else if (value.startsWith("force")) {
    12.            // force:xxx 直接执行 mock 逻辑，不发起远程调用
    13.            result = doMockInvoke(invocation, null);
    14.        } else {
    15.             // fail:xxx 表示消费方对调用服务失败后，再执行 mock 逻辑，不抛出异常
    16.            try {
    17.                result = this.invoker.invoke(invocation);
    18.            } catch (RpcException e) {
    19.                 // 调用失败，执行 mock 逻辑
    20.                result = doMockInvoke(invocation, e);
    21.            }
    22.        }
    23.        return result;
    24.    }

    考虑到前文已经详细分析过FailoverClusterInvoker，因此本节略过FailoverClusterInvoker，直接分析DubboInvoker。

    1. public abstract class AbstractInvoker<T> implements Invoker<T> {
    2.    
    3.    public Result invoke(Invocation inv) throws RpcException {
    4.        if (destroyed.get()) {
    5.            throw new RpcException("Rpc invoker for service ...");
    6.             }
    7.        RpcInvocation invocation = (RpcInvocation) inv;
    8.        // 设置 Invoker
    9.        invocation.setInvoker(this);
    10.        if (attachment != null && attachment.size() > 0) {
    11.            // 设置 attachment
    12.            invocation.addAttachmentsIfAbsent(attachment);
    13.        }
    14.        Map<String, String> contextAttachments =
    15. RpcContext.getContext().getAttachments();
    16.        if (contextAttachments != null && contextAttachments.size() != 0) {
    17.            // 添加 contextAttachments 到 RpcInvocation#attachment 变量中
    18.            invocation.addAttachments(contextAttachments);
    19.        }
    20.        if (getUrl().getMethodParameter(invocation.getMethodName(),
    21. Constants.ASYNC_KEY, false)) {
    22.            // 设置异步信息到 RpcInvocation#attachment 中
    23.            invocation.setAttachment(Constants.ASYNC_KEY,
    24. Boolean.TRUE.toString());
    25.        }
    26.        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    27.        try {
    28.            // 抽象方法，由子类实现
    29.            return doInvoke(invocation);
    30.        } catch (InvocationTargetException e) {
    31.            // ...
    32.        } catch (RpcException e) {
    33.            // ...
    34.        } catch (Throwable e) {
    35.            return new RpcResult(e);
    36.        }
    37.    }
    38.    protected abstract Result doInvoke(Invocation invocation) throws Throwable;
    39.    
    40.    // 省略其他方法
    41. }

    上面的代码来自AbstractInvoker 类，其中大部分代码用于添加信息到RpcInvocation#attachment 变量中，添加完毕后，调用doInvoke 执行后续的调用。doInvoke 是一个抽象方法，需要由子类实现，下面到DubboInvoker 中看一下。

    1. @Override
    2.    protected Result doInvoke(final Invocation invocation) throws Throwable {
    3.        RpcInvocation inv = (RpcInvocation) invocation;
    4.        final String methodName = RpcUtils.getMethodName(invocation);
    5.        //将目标方法以及版本号作为参数放入到Invocation中
    6.        inv.setAttachment(PATH_KEY, getUrl().getPath());
    7.        inv.setAttachment(VERSION_KEY, version);
    8.        //获得客户端连接
    9.        ExchangeClient currentClient; //初始化invoker的时候，构建的一个远程通信连接
    10.        if (clients.length == 1) { //默认
    11.            currentClient = clients[0];
    12.        } else {
    13.            //通过取模获得其中一个连接
    14.             currentClient = clients[index.getAndIncrement() % clients.length];
    15.        }
    16.        try {
    17.            //表示当前的方法是否存在返回值
    18.            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
    19.            int timeout = getUrl().getMethodParameter(methodName, TIMEOUT_KEY,
    20. DEFAULT_TIMEOUT);
    21.            //isOneway 为 true，表示“单向”通信
    22.            if (isOneway) {//异步无返回值
    23.                boolean isSent = getUrl().getMethodParameter(methodName,
    24. Constants.SENT_KEY, false);
    25.                currentClient.send(inv, isSent);
    26.                RpcContext.getContext().setFuture(null);
    27.                return AsyncRpcResult.newDefaultAsyncResult(invocation);
    28.            } else { //存在返回值
    29.                //是否采用异步
    30.                AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
    31.                CompletableFuture responseFuture =
    32. currentClient.request(inv, timeout);
    33.                responseFuture.whenComplete((obj, t) -> {
    34.                    if (t != null) {
    35.                        asyncRpcResult.completeExceptionally(t);
    36.                    } else {
    37.                        asyncRpcResult.complete((AppResponse) obj);
    38.                    }
    39.                });
    40.                RpcContext.getContext().setFuture(new
    41. FutureAdapter(asyncRpcResult));
    42.                return asyncRpcResult;
    43.            }
    44.        }
    45.        //省略无关代码
    46.    }
      47.  

      最终进入到HeaderExchangeChannel#request方法，拼装Request并将请求发送出去
       

      1. public CompletableFuture<Object> request(Object request, int timeout) throws
      2. RemotingException {
      3.        if (closed) {
      4.            throw new RemotingException(this.getLocalAddress(), null, "Failed
      5. tosend request " + request + ", cause: The channel " + this + " is closed!");
      6.        }
      7.        // 创建请求对象
      8.        Request req = new Request();
      9.        req.setVersion(Version.getProtocolVersion());
      10.        req.setTwoWay(true);
      11.        req.setData(request);
      12.        DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout);
      13.        try {
      14.            //NettyClient
      15.            channel.send(req);
      16.        } catch (RemotingException e) {
      17.            future.cancel();
      18.            throw e;
      19.        }
      20.        return future;
      21.    }
       

      请求编码如何做的?
       

      在netty启动时，我们设置了编解码器，其中通过ExchangeCodec完成编解码工作如下：
       

      1. public class ExchangeCodec extends TelnetCodec {
      2.    // 消息头长度
      3.    protected static final int HEADER_LENGTH = 16;
      4.    // 魔数内容
      5.    protected static final short MAGIC = (short) 0xdabb;
      6.    protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
      7.    protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
      8.    protected static final byte FLAG_REQUEST = (byte) 0x80;
      9.    protected static final byte FLAG_TWOWAY = (byte) 0x40;
      10.    protected static final byte FLAG_EVENT = (byte) 0x20;
      11.    protected static final int SERIALIZATION_MASK = 0x1f;
      12.    private static final Logger logger =
      13. LoggerFactory.getLogger(ExchangeCodec.class);
      14.    public Short getMagicCode() {
      15.        return MAGIC;
      16.    }
      17.    @Override
      18.    public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws
      19. IOException {
      20.        if (msg instanceof Request) {
      21.            // 对 Request 对象进行编码
      22.            encodeRequest(channel, buffer, (Request) msg);
      23.        } else if (msg instanceof Response) {
      24.            // 对 Response 对象进行编码，后面分析
      25.            encodeResponse(channel, buffer, (Response) msg);
      26.        } else {
      27.            super.encode(channel, buffer, msg);
      28.        }
      29.    }
      30.    protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request
      31. req) throws IOException {
      32.        Serialization serialization = getSerialization(channel);
      33.        // 创建消息头字节数组，长度为 16
      34.        byte[] header = new byte[HEADER_LENGTH];
      35.        // 设置魔数
      36.        Bytes.short2bytes(MAGIC, header);
      37.        // 设置数据包类型（Request/Response）和序列化器编号
      38.        header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
      39.        // 设置通信方式(单向/双向)
      40.        if (req.isTwoWay()) {
      41.            header[2] |= FLAG_TWOWAY;
      42.        }
      43.        
      44.        // 设置事件标识
      45.        if (req.isEvent()) { header[2] |= FLAG_EVENT;
      46.        }
      47.        // 设置请求编号，8个字节，从第4个字节开始设置
      48.        Bytes.long2bytes(req.getId(), header, 4);
      49.        // 获取 buffer 当前的写位置
      50.        int savedWriteIndex = buffer.writerIndex();
      51.        // 更新 writerIndex，为消息头预留 16 个字节的空间
      52.        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
      53.        ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
      54.        // 创建序列化器，比如 Hessian2ObjectOutput
      55.        ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
      56.        if (req.isEvent()) {
      57.            // 对事件数据进行序列化操作
      58.            encodeEventData(channel, out, req.getData());
      59.        } else {
      60.            // 对请求数据进行序列化操作
      61.            encodeRequestData(channel, out, req.getData(), req.getVersion());
      62.        }
      63.        out.flushBuffer();
      64.        if (out instanceof Cleanable) {
      65.            ((Cleanable) out).cleanup();
      66.        }
      67.        bos.flush();
      68.        bos.close();
      69.        
      70.        // 获取写入的字节数，也就是消息体长度
      71.        int len = bos.writtenBytes();
      72.        checkPayload(channel, len);
      73.        // 将消息体长度写入到消息头中
      74.        Bytes.int2bytes(len, header, 12);
      75.        // 将 buffer 指针移动到 savedWriteIndex，为写消息头做准备
      76.        buffer.writerIndex(savedWriteIndex);
      77.        // 从 savedWriteIndex 下标处写入消息头
      78.        buffer.writeBytes(header);
      79.        // 设置新的 writerIndex，writerIndex = 原写下标 + 消息头长度 + 消息体长度
      80.        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
      81.    }
      82.    
      83.    // 省略其他方法
      84. }
       

      折叠
       

      以上就是请求对象的编码过程，该过程首先会通过位运算将消息头写入到header 数组中。然后对Request 对象的data 字段执行序列化操作，序列化后的数据最终会存储到ChannelBuffer 中。序列化操作执行完后，可得到数据序列化后的长度len，紧接着将len 写入到header 指定位置处。最后再将消息头字节数组header 写入到ChannelBuffer 中，整个编码过程就结束了。本节的最后，我们再来看一下Request 对象的data 字段序列化过程，也就是encodeRequestData 方法的逻辑，如下：
       

      1. public class DubboCodec extends ExchangeCodec implements Codec2 {
      2.    
      3. protected void encodeRequestData(Channel channel, ObjectOutput out, Object
      4. data, String version) throws IOException {
      5.        RpcInvocation inv = (RpcInvocation) data;
      6.      // 依次序列化 dubbo version、path、version
      7.        out.writeUTF(version);
      8.        out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
      9.        out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));
      10.        // 序列化调用方法名
      11.        out.writeUTF(inv.getMethodName());
      12.        // 将参数类型转换为字符串，并进行序列化
      13.        out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
      14.        Object[] args = inv.getArguments();
      15.        if (args != null)
      16.            for (int i = 0; i < args.length; i++) {
      17.                // 对运行时参数进行序列化
      18.                out.writeObject(encodeInvocationArgument(channel, inv, i));
      19.            }
      20.        
      21.        // 序列化 attachments
      22.        out.writeObject(inv.getAttachments());
      23.    }
      24. }
       

      至此，关于服务消费方发送请求的过程就分析完了，接下来我们来看一下服务提供方是如何接收请求的。
       

      提供方接受请求
                  
                      
      dubbo源码解析
                      
                          pdf                                        5星                                        超过95%的资源                                        2.50MB                
                  
                  
                      
                      
      下载
                  
              
       

      请求如何解码?
       

      这里直接分析请求数据的解码逻辑，忽略中间过程，如下：
       

      1. public class ExchangeCodec extends TelnetCodec {
      2.    
      3.    @Override
      4.    public Object decode(Channel channel, ChannelBuffer buffer) throws
      5. IOException {
      6.        int readable = buffer.readableBytes();
      7.        // 创建消息头字节数组
      8.        byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
      9.        // 读取消息头数据
      10.        buffer.readBytes(header);
      11.        // 调用重载方法进行后续解码工作
      12.        return decode(channel, buffer, readable, header);
      13.    }
      14.    @Override
      15.    protected Object decode(Channel channel, ChannelBuffer buffer, int readable,
      16. byte[] header) throws IOException {
      17.        // 检查魔数是否相等
      18.        if (readable > 0 && header[0] != MAGIC_HIGH
      19.                || readable > 1 && header[1] != MAGIC_LOW) {
      20.            int length = header.length;
      21.            if (header.length < readable) {
      22.                header = Bytes.copyOf(header, readable);
      23.                buffer.readBytes(header, length, readable - length);
      24.            }
      25.            for (int i = 1; i < header.length - 1; i++) {
      26.                if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
      27.                     buffer.readerIndex(buffer.readerIndex() - header.length +
      28. i);
      29.                    header = Bytes.copyOf(header, i);
      30.                    break;
      31.                }
      32.            }
      33.            // 通过 telnet 命令行发送的数据包不包含消息头，所以这里
      34.            // 调用 TelnetCodec 的 decode 方法对数据包进行解码
      35.            return super.decode(channel, buffer, readable, header);
      36.        }
      37.        
      38.        // 检测可读数据量是否少于消息头长度，若小于则立即返回
      39. DecodeResult.NEED_MORE_INPUT
      40.        if (readable < HEADER_LENGTH) {
      41.            return DecodeResult.NEED_MORE_INPUT;
      42.        }
      43.        // 从消息头中获取消息体长度
      44.        int len = Bytes.bytes2int(header, 12);
      45.        // 检测消息体长度是否超出限制，超出则抛出异常
      46.        checkPayload(channel, len);
      47.        int tt = len + HEADER_LENGTH;
      48.        // 检测可读的字节数是否小于实际的字节数
      49.        if (readable < tt) {
      50.            return DecodeResult.NEED_MORE_INPUT;
      51.        }
      52.        
      53.        ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);
      54.        try {
      55.            // 继续进行解码工作
      56.            return decodeBody(channel, is, header);
      57.        } finally {
      58.            if (is.available() > 0) {
      59.                try {
      60.                    StreamUtils.skipUnusedStream(is);
      61.                } catch (IOException e) {
      62.                    logger.warn(e.getMessage(), e);
      63.                }
      64.            }
      65.        }
      66.    }
      67. }
       

      折叠
       

      上面方法通过检测消息头中的魔数是否与规定的魔数相等，提前拦截掉非常规数据包，比如通过telnet命令行发出的数据包。接着再对消息体长度，以及可读字节数进行检测。最后调用decodeBody 方法进行后续的解码工作，ExchangeCodec 中实现了decodeBody 方法，但因其子类DubboCodec 覆写了该方法，所以在运行时DubboCodec 中的decodeBody 方法会被调用。下面我们来看一下该方法的代码。
       

      1. public class DubboCodec extends ExchangeCodec implements Codec2 {
      2.    @Override
      3.    protected Object decodeBody(Channel channel, InputStream is, byte[] header)
      4. throws IOException {
      5.        // 获取消息头中的第三个字节，并通过逻辑与运算得到序列化器编号byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
      6.        Serialization s = CodecSupport.getSerialization(channel.getUrl(),
      7. proto);
      8.        // 获取调用编号
      9.        long id = Bytes.bytes2long(header, 4);
      10.        // 通过逻辑与运算得到调用类型，0 - Response，1 - Request
      11.        if ((flag & FLAG_REQUEST) == 0) {
      12.            // 对响应结果进行解码，得到 Response 对象。这个非本节内容，后面再分析
      13.            // ...
      14.        } else {
      15.            // 创建 Request 对象
      16.            Request req = new Request(id);
      17.            req.setVersion(Version.getProtocolVersion());
      18.            // 通过逻辑与运算得到通信方式，并设置到 Request 对象中
      19.            req.setTwoWay((flag & FLAG_TWOWAY) != 0);
      20.            
      21.            // 通过位运算检测数据包是否为事件类型
      22.            if ((flag & FLAG_EVENT) != 0) {
      23.                // 设置心跳事件到 Request 对象中
      24.                req.setEvent(Request.HEARTBEAT_EVENT);
      25.            }
      26.            try {
      27.                Object data;
      28.                if (req.isHeartbeat()) {
      29.                    // 对心跳包进行解码，该方法已被标注为废弃
      30.                    data = decodeHeartbeatData(channel, deserialize(s,
      31. channel.getUrl(), is));
      32.                } else if (req.isEvent()) {
      33.                    // 对事件数据进行解码
      34.                    data = decodeEventData(channel, deserialize(s,
      35. channel.getUrl(), is));
      36.                } else {
      37.                    DecodeableRpcInvocation inv;
      38.                    // 根据 url 参数判断是否在 IO 线程上对消息体进行解码
      39.                    if (channel.getUrl().getParameter(
      40.                            Constants.DECODE_IN_IO_THREAD_KEY,
      41.                            Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
      42.                        inv = new DecodeableRpcInvocation(channel, req, is,
      43. proto);
      44.                        // 在当前线程，也就是 IO 线程上进行后续的解码工作。此工作完成后，
      45. 可将
      46.                        // 调用方法名、attachment、以及调用参数解析出来
      47.                        inv.decode();
      48.                    } else {
      49.                        // 仅创建 DecodeableRpcInvocation 对象，但不在当前线程上执行解
      50. 码逻辑
      51.                        inv = new DecodeableRpcInvocation(channel, req,
      52.                                new
      53. UnsafeByteArrayInputStream(readMessageData(is)), proto);
      54.                    }
      55.                    data = inv;
      56.                }
      57.                
      58.                // 设置 data 到 Request 对象中
      59.                req.setData(data);
      60.            } catch (Throwable t) {
      61.                // 若解码过程中出现异常，则将 broken 字段设为 true，
      62.                // 并将异常对象设置到 Reqeust 对象中
      63.                 req.setBroken(true);
      64.                req.setData(t);
      65.            }
      66.            return req;
      67.        }
      68.    }
      69. }
       

      折叠
       

      如上，decodeBody 对部分字段进行了解码，并将解码得到的字段封装到Request 中。随后会调用DecodeableRpcInvocation 的decode 方法进行后续的解码工作。此工作完成后，可将调用方法名、attachment、以及调用参数解析出来。
       

      调用服务
       

      解码器将数据包解析成Request 对象后，NettyHandler 的messageReceived 方法紧接着会收到这个对象，并将这个对象继续向下传递。整个调用栈如下：
       

      1. NettyServerHandler#channelRead(ChannelHandlerContext, MessageEvent)
      2.  —> AbstractPeer#received(Channel, Object)
      3.    —> MultiMessageHandler#received(Channel, Object)
      4.      —> HeartbeatHandler#received(Channel, Object)
      5.        —> AllChannelHandler#received(Channel, Object)
      6.          —> ExecutorService#execute(Runnable)    // 由线程池执行后续的调用逻辑
       

      这里我们直接分析调用栈中的分析第一个和最后一个调用方法逻辑。如下：
       考虑到篇幅，以及很多中间调用的逻辑并非十分重要，所以这里就不对调用栈中的每个方法都进行分析了。这里我们直接分析最后一个调用方法逻辑。如下：
       

      1. public class ChannelEventRunnable implements Runnable {
      2.    
      3.    private final ChannelHandler handler;
      4.    private final Channel channel;
      5.    private final ChannelState state;
      6.    private final Throwable exception;
      7.    private final Object message;
      8.    
      9.    @Override
      10.    public void run() {
      11.        // 检测通道状态，对于请求或响应消息，此时 state = RECEIVED
      12.        if (state == ChannelState.RECEIVED) {
      13.            try {
      14.                // 将 channel 和 message 传给 ChannelHandler 对象，进行后续的调用
      15.                handler.received(channel, message);
      16.            } catch (Exception e) {
      17.                logger.warn("... operation error, channel is ... message is
      18. ...");
      19.            }
      20.        }
      21.        
      22.        // 其他消息类型通过 switch 进行处理
      23.        else {
      24.            switch (state) {
      25.            case CONNECTED:
      26.                try {
      27.                    handler.connected(channel);
      28.                } catch (Exception e) {
      29.                    logger.warn("... operation error, channel is ...");
      30.                }
      31.                break;
      32.            case DISCONNECTED:
      33.                // ...
      34.            case SENT:
      35.                // ...
      36.            case CAUGHT:
      37.                // ...
      38.            default:
      39.                logger.warn("unknown state: " + state + ", message is " +
      40. message);
      41.            }
      42.        }
      43.    }
      44. }
       

      如上，请求和响应消息出现频率明显比其他类型消息高，所以这里对该类型的消息进行了针对性判断ChannelEventRunnable 仅是一个中转站，它的run 方法中并不包含具体的调用逻辑，仅用于将参数传给其他ChannelHandler 对象进行处理，该对象类型为DecodeHandler
       

      1. public class DecodeHandler extends AbstractChannelHandlerDelegate {
      2.    public DecodeHandler(ChannelHandler handler) {
      3.        super(handler);
      4.    }
      5.    @Override
      6.    public void received(Channel channel, Object message) throws
      7. RemotingException {
      8.        if (message instanceof Decodeable) {
      9.            // 对 Decodeable 接口实现类对象进行解码
      10.            decode(message);
      11.        }
      12.        if (message instanceof Request) {
      13.            // 对 Request 的 data 字段进行解码
      14.            decode(((Request) message).getData());
      15.        }
      16.        if (message instanceof Response) {
      17.            // 对 Request 的 result 字段进行解码
      18.            decode(((Response) message).getResult());
      19.        }
      20.        // 执行后续逻辑
      21.        handler.received(channel, message);
      22.    }
      23.    private void decode(Object message) {
      24.        // Decodeable 接口目前有两个实现类，
      25.        // 分别为 DecodeableRpcInvocation 和 DecodeableRpcResult
      26.        if (message != null && message instanceof Decodeable) {
      27.            try {
      28.                // 执行解码逻辑
      29.                ((Decodeable) message).decode();
      30.                } catch (Throwable e) {
      31.                if (log.isWarnEnabled()) {
      32.                    log.warn("Call Decodeable.decode failed: " + e.getMessage(),
      33. e);
      34.                }
      35.            }
      36.        }
      37.    }
      38. }
       

      DecodeHandler 主要是包含了一些解码逻辑，完全解码后的Request 对象会继续向后传递
       

      1. public class DubboProtocol extends AbstractProtocol {
      2.    public static final String NAME = "dubbo";
      3.    
      4.    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
      5.        @Override
      6.        public Object reply(ExchangeChannel channel, Object message) throws
      7. RemotingException {
      8.            if (message instanceof Invocation) {
      9.                Invocation inv = (Invocation) message;
      10.                // 获取 Invoker 实例
      11.                Invoker> invoker = getInvoker(channel, inv);
      12.                if
      13. (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INV
      14. OKE))) {
      15.                    // 回调相关，忽略
      16.                }
      17.              
      18.  RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
      19.                // 通过 Invoker 调用具体的服务
      20.                return invoker.invoke(inv);
      21.            }
      22.            throw new RemotingException(channel, "Unsupported request: ...");
      23.        }
      24.        
      25.        // 忽略其他方法
      26.    }
      27.    
      28.    Invoker> getInvoker(Channel channel, Invocation inv) throws
      29. RemotingException {
      30.        // 忽略回调和本地存根相关逻辑
      31.        // ...
      32.        
      33.        int port = channel.getLocalAddress().getPort();
      34.        
      35.        // 计算 service key，格式为 groupName/serviceName:serviceVersion:port。比
      36. 如：
      37.        //   dubbo/com.alibaba.dubbo.demo.DemoService:1.0.0:20880
      38.        String serviceKey = serviceKey(port, path,
      39. inv.getAttachments().get(Constants.VERSION_KEY),
      40. inv.getAttachments().get(Constants.GROUP_KEY));
      41.        // 从 exporterMap 查找与 serviceKey 相对应的 DubboExporter 对象，
      42.        // 服务导出过程中会将  映射关系存储到 exporterMap
      43. 集合中
      44.        DubboExporter> exporter = (DubboExporter>)
      45. exporterMap.get(serviceKey);
      46.        if (exporter == null)
      47.            throw new RemotingException(channel, "Not found exported service
      48. ...");
      49.        // 获取 Invoker 对象，并返回
      50.        return exporter.getInvoker();
      51.    }
      52.    
      53.    // 忽略其他方法
      54. }
       

      折叠
       

      在之前课程中介绍过，服务全部暴露完成之后保存到exporterMap中。这里就是通过serviceKey获取exporter之后获取Invoker，并通过Invoker 的invoke 方法调用服务逻辑
       

      1. public abstract class AbstractProxyInvoker implements Invoker {
      2.    @Override
      3.    public Result invoke(Invocation invocation) throws RpcException {
      4.        try {
      5.            // 调用 doInvoke 执行后续的调用，并将调用结果封装到 RpcResult 中，并
      6.            return new RpcResult(doInvoke(proxy, invocation.getMethodName(),
      7. invocation.getParameterTypes(), invocation.getArguments()));
      8.        } catch (InvocationTargetException e) {
      9.            return new RpcResult(e.getTargetException());
      10.        } catch (Throwable e) {
      11.            throw new RpcException("Failed to invoke remote proxy method ...");
      12.        }
      13.    }
      14.    
      15.    protected abstract Object doInvoke(T proxy, String methodName, Class[]
      16. parameterTypes, Object[] arguments) throws Throwable;
      17. }
       

      如上，doInvoke 是一个抽象方法，这个需要由具体的Invoker 实例实现。Invoker 实例是在运行时通过JavassistProxyFactory 创建的，创建逻辑如下：
       

      1. public class JavassistProxyFactory extends AbstractProxyFactory {
      2.    
      3.    // 省略其他方法
      4.    @Override
      5.    public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
      6.        final Wrapper wrapper =
      7. Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ?
      8. proxy.getClass() : type);
      9.        // 创建匿名类对象
      10.        return new AbstractProxyInvoker<T>(proxy, type, url) {
      11.            @Override
      12.            protected Object doInvoke(T proxy, String methodName,
      13.                                      Class[] parameterTypes,
      14.                                      Object[] arguments) throws Throwable {
      15.                // 调用 invokeMethod 方法进行后续的调用
      16.                return wrapper.invokeMethod(proxy, methodName, parameterTypes,
      17. arguments);
      18.            }
      19.        };
      20.    }
      21. }
       

      Wrapper 是一个抽象类，其中invokeMethod 是一个抽象方法。Dubbo 会在运行时通过Javassist 框架为Wrapper 生成实现类，并实现invokeMethod 方法，该方法最终会根据调用信息调用具体的服务。以DemoServiceImpl 为例，Javassist 为其生成的代理类如下。
       

      1. /** Wrapper0 是在运行时生成的，大家可使用 Arthas 进行反编译 */
      2. public class Wrapper0 extends Wrapper implements ClassGenerator.DC {
      3.    public static String[] pns;
      4.    public static Map pts;
      5.    public static String[] mns;
      6.    public static String[] dmns;
      7.    public static Class[] mts0;
      8.    // 省略其他方法
      9.    public Object invokeMethod(Object object, String string, Class[] arrclass,
      10. Object[] arrobject) throws InvocationTargetException {
      11.        DemoService demoService;
      12.        try {
      13.            // 类型转换
      14.            demoService = (DemoService)object;
      15.        }
      16.        catch (Throwable throwable) {
      17.            throw new IllegalArgumentException(throwable);
      18.        }
      19.        try {
      20.            // 根据方法名调用指定的方法
      21.            if ("sayHello".equals(string) && arrclass.length == 1) {
      22.                return demoService.sayHello((String)arrobject[0]);
      23.            }
      24.        }
      25.        catch (Throwable throwable) {
      26.            throw new InvocationTargetException(throwable);
      27.        }
      28.        throw new NoSuchMethodException(new StringBuffer().append("Not found
      29. method \"").append(string).append("\" in class
      30. com.alibaba.dubbo.demo.DemoService.").toString());
      31.    }
      32. }
       

      到这里，整个服务调用过程就分析完了。最后把调用过程贴出来，如下：
       

      1. ChannelEventRunnable#run()
      2.  —> DecodeHandler#received(Channel, Object)
      3.    —> HeaderExchangeHandler#received(Channel, Object)
      4.      —> HeaderExchangeHandler#handleRequest(ExchangeChannel, Request)
      5.        —> DubboProtocol.requestHandler#reply(ExchangeChannel, Object)
      6.          —> Filter#invoke(Invoker, Invocation)
      7.            —> AbstractProxyInvoker#invoke(Invocation)
      8.              —> Wrapper0#invokeMethod(Object, String, Class[], Object[])
      9.                —> DemoServiceImpl#sayHello(String)
       

      提供方返回调用结果
       

      服务提供方调用指定服务后，会将调用结果封装到Response 对象中，并将该对象返回给服务消费方。服务提供方也是通过NettyChannel 的send 方法将Response 对象返回，这里就不在重复分析了。
       

      消费方接收调用结果
       

      服务消费方在收到响应数据后，首先要做的事情是对响应数据进行解码，得到Response 对象。然后再将该对象传递给下一个入站处理器，这个入站处理器就是NettyHandler。接下来NettyHandler 会将这个对象继续向下传递，最后AllChannelHandler 的received 方法会收到这个对象，并将这个对象派发到线程池中。这个过程和服务提供方接收请求的过程是一样的，因此这里就不重复分析了
       

      小结
       

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

                
    47. 

                    相关阅读:
      
                    
jasperreports6.12.2 could not load the following font解决方法                            
      
【k8s】(一) 使用kubeadm快速部署一个K8s集群                            
      
记一次 Sedona（GeoSpark） 空间计算优化                            
      
Flume从入门到精通一站式学习笔记                            
      
PostgreSQL数据库统计信息——examine_attribute单列预分析                            
      
354俄罗斯套娃信封问题                            
      
用OLED屏幕播放视频(3): 使用cuda编程加速视频处理                            
      
KMP超高效匹配算法                            
      
【数据库系统】单表数据查询                            
      
人工神经网络案例分析题,人工神经网络算法实例                            
      
                          
                
      48. 
                
    48. 
                    原文地址：https://blog.csdn.net/weixin_72753070/article/details/126035925
                
      49.