8.1 构建客户端总体流程

Posted 2020-10-14 赵计刚

一 示例

1 配置文件：

 1 <?xml version="1.0" encoding="UTF-8"?>
 2 <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 3        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
 4        xmlns="http://www.springframework.org/schema/beans"
 5        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
 6     http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd">
 7     <!-- 消费方应用名，用于计算依赖关系，不是匹配条件，不要与提供方一样 -->
 8     <dubbo:application name="demo-consumer"/>
 9     <!-- 使用zookeeper注册中心 -->
10     <dubbo:registry protocol="zookeeper" address="10.211.55.5:2181"/>
11     <!-- 生成远程服务代理，可以和本地bean一样使用demoService -->
12     <dubbo:reference id="demoService" check="false" interface="com.alibaba.dubbo.demo.DemoService"/>
13 </beans>

2 Consumer

 1 package com.alibaba.dubbo.demo.consumer;
 2 
 3 import com.alibaba.dubbo.demo.DemoService;
 4 import org.springframework.context.support.ClassPathXmlApplicationContext;
 5 
 6 public class Consumer {
 7     public static void main(String[] args) {
 8         ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"});
 9         context.start();
10 
11         DemoService demoService = (DemoService) context.getBean("demoService"); // 获取远程服务代理
12         String hello = demoService.sayHello("world"); // 执行远程方法
13 
14         System.out.println(hello); // 显示调用结果
15     }
16 }

先来看DemoService demoService = (DemoService) context.getBean("demoService"); // 获取远程服务代理。

 

二 调用简图

 

三 总体代码调用链

ReferenceConfig.init()
-->createProxy(Map<String, String> map)
  //一 获取Invoker
  -->RegistryProtocol.refer(Class<T> type, URL url)
    //1 获取注册中心：创建ZkClient实例，连接zk
    -->Registry registry = registryFactory.getRegistry(url)
      -->AbstractRegistryFactory.getRegistry(URL url)
        -->ZookeeperRegistryFactory.createRegistry(URL url)
          -->new ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter)
            -->ZkclientZookeeperTransporter.connect(URL url)
              -->new ZkclientZookeeperClient(URL url)
                -->new ZkClient(url.getBackupAddress())
            -->AbstractRegistryFactory.Map<String, Registry> REGISTRIES.put("zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService", 上边的ZookeeperRegistry实例)
    -->doRefer(Cluster cluster, Registry registry, Class<T> type, URL url)
      -->new RegistryDirectory<T>(type, url)
      //2 向注册中心注册服务
      -->registry.register(url)
        -->ZookeeperRegistry.doRegister(URL url)
          -->AbstractZookeeperClient.create(String path, boolean ephemeral)
　　　 //3 订阅providers、configurators、routers
      -->RegistryDirectory.subscribe(URL url)
        -->ZookeeperRegistry.doSubscribe(final URL url, final NotifyListener listener)
          //3.1 会获取当前节点下已经存在的字节点（第一次服务发现发生在这里），添加子节点变化监听器
          -->List<String> children = zkClient.addChildListener(path, zkListener)
          -->AbstractRegistry.notify(URL url, NotifyListener listener, List<URL> urls)
            -->saveProperties(url)
            -->RegistryDirectory.notify(List<URL> urls)
              //仅仅针对的是providers
              -->refreshInvoker(List<URL> invokerUrls)
                -->toInvokers(List<URL> urls
                  -->ProtocolFilterWrapper.refer(Class<T> type, URL url)
                    -->DubboProtocol.refer(Class<T> serviceType, URL url)
                      //3.1.1 创建ExchangeClient，对第一次服务发现providers路径下的相关url建立长连接
                      -->getClients(URL url)
                        -->getSharedClient(URL url)
                          -->ExchangeClient exchangeClient = initClient(url)
                            -->Exchangers.connect(url, requestHandler)
                              -->HeaderExchanger.connect(URL url, ExchangeHandler handler)
                                -->new DecodeHandler(new HeaderExchangeHandler(handler)))
                                  -->Transporters.connect(URL url, ChannelHandler... handlers)
                                    -->NettyTransporter.connect(URL url, ChannelHandler listener)
                                      -->new NettyClient(url, listener)
                                        -->new MultiMessageHandler(HeartbeatHandler(AllChannelHandler(handler)))
                                        -->getChannelCodec(url)//获取Codec2，这里是DubboCountCodec实例
                                        -->doOpen()//开启netty客户端
                                        -->doConnect()//连接服务端，建立长连接
                                -->new HeaderExchangeClient(Client client, boolean needHeartbeat)//上述的NettyClient实例，needHeartbeat：true
                                  -->startHeatbeatTimer()//启动心跳计数器
                          -->ReferenceCountExchangeClient(ExchangeClient client, ConcurrentMap<String, LazyConnectExchangeClient> ghostClientMap)/
                          -->Map<String, ReferenceCountExchangeClient> referenceClientMap.put("10.10.10.10:20880", 上边的ReferenceCountExchangeClient实例)
                      //3.2 创建DubboInvoker
                      -->new DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers)
                      -->DubboProtocol.Set<Invoker<?>> invokers.add(上边的DubboInvoker实例)
                    -->ProtocolFilterWrapper.buildInvokerChain(final Invoker<T> invoker, String key, String group)
                  -->new InvokerDelegete(Invoker<T> invoker, URL url, URL providerUrl)
                  //3.3 将创建出来的Invoker缓存起来
                  -->newUrlInvokerMap.put("dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-consumer&check=false&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=16001&register.ip=10.10.10.10&remote.timestamp=1510127991625&side=consumer&timestamp=1510128022123", 上边的InvokerDelegate实例)
                -->toMethodInvokers(newUrlInvokerMap)
                -->Map<String, List<Invoker<T>>> newMethodInvokerMap:{sayHello=[InvokerDelegete实例], *=[InvokerDelegete实例]}
      //4 将directory封装成一个ClusterInvoker（MockClusterInvoker）
      -->cluster.join(directory)
        -->Cluster$Adaptive.join(directory)
          -->ExtensionLoader.getExtensionLoader(Cluster.class).getExtension("failover")//MockClusterWrapper包装FailoverCluster
            -->MockClusterWrapper.join(Directory<T> directory)
              -->FailoverCluster.join(Directory<T> directory)
                -->new FailoverClusterInvoker<T>(directory)
              -->MockClusterInvoker(Directory<T> directory, Invoker<T> invoker)//invoker:上边的FailoverClusterInvoker实例
  //二 获取代理
  -->JavassistProxyFactory.getProxy(Invoker<T> invoker, Class<?>[] interfaces)//invoker：上边的MockClusterInvoker实例， interfaces：[interface com.alibaba.dubbo.demo.DemoService, interface com.alibaba.dubbo.rpc.service.EchoService]
    -->Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker))
      -->Proxy.getProxy(ClassLoader cl, Class<?>... ics)//使用javassist获取一个动态类
      -->new InvokerInvocationHandler(invoker)//invoker：上边的MockClusterInvoker实例

极简版流程图：

proxy0.xxxMethod()
-->InvokerInvocationHandler.invoke
  // RpcInvocation [methodName=sayHello, parameterTypes=[class java.lang.String], arguments=[world], attachments={}]
  -->MockClusterInvoker.invoke(Invocation invocation)
    -->FailoverClusterInvoker.invoke(final Invocation invocation)
      -->RegistryDirectory.list(Invocation invocation) //根据RpcInvocation中的methodName获取Invoker
        -->router过滤
        -->loadBalancer选取一个Invoker
      -->执行filter链
        // RpcInvocation [methodName=sayHello, parameterTypes=[class java.lang.String], arguments=[world], attachments={path=com.alibaba.dubbo.demo.DemoService, interface=com.alibaba.dubbo.demo.DemoService, version=2.0.0, timeout=60000, group=dev}]
        -->DubboInvoker.invoke(Invocation inv)

在服务端会根据RpcInvocation中的attachments中的path、group、version以及从channel中获取的port拼接一个serviceKey：group/path:version:port，例如：dev/com.alibaba.dubbo.demo.DemoService:2.0.0:20880，之后根据这个serviceKey从服务端获取DubboExporter。

注意：

1、每一个代理都会有自己的一个MockClusterInvoker，也就有自己的一个RegistryDirectory，所以，假设A引用了服务B和C，B和C中都有sayHello(String name)方法时，B的RegistryDirectory存储自己的sayHello=[Invoker对象]，C的RegistryDirectory存储自己的sayHello=[Invoker对象]。

2、假设A调用了服务B，B中有方法String sayHello(String name)/String sayHello(String name, Integer age)时，在B的RegistryDirectory只存储一份的sayHello=[Invoker对象]，注意这里一个Invoker对象其实对应一个Provider实例，根据从serviceKey：group/path:version:port获取出DubboExporter，DubboExporter获取到AbstractProxyInvoker，AbstractProxyInvoker中的wrapper类中就有String sayHello(String name)/String sayHello(String name, Integer age)两个方法，根据Request中的方法名、参数类型和参数值就可以找出执行哪一个方法。