一、启动流程概述
    启动的入口在ReferenceConfig类的get()方法，消费者启动流程包含以下几步：
    1. 构建Registry实例、RegistryDirectory实例
    2. 向注册中心注册消费者信息
    3. 构建路由规则链
    4. 向注册中心订阅提供者信息
    5. 将订阅的提供者信息转换为DubboInvoker，并且开启通信服务（默认是netty，消费者在启动时会与多个提供者之间都建立连接，但缺省情况下同一个提供者下的所有rpc调用共用一个连接，但是也可以配置为不共用）
    6.将构建好的Invoker实例封装进集群容错策略类中
    7.将构建好的集群容错策略实例封装在InvokerInvocationHandler类中

    consumer://192.168.56.1/service.GreetingServiceAsync?application=first-dubbo-consumer&async=true&dubbo=2.0.2&group=dubbo&interface=service.GreetingServiceAsync&lazy=false&methods=sayHello,testGeneric&pid=13912&release=2.7.3&revision=1.0.0&side=consumer&sticky=false&timeout=3000&timestamp=1581562335685&version=1.0.0

    二、启动时序图
    image.png
    图1.0

    三、源码分析
    1、ReferenceConfig
    代码1.0

    1.  public synchronized T get() {
    2.         checkAndUpdateSubConfigs();
    3.         if (destroyed) {
    4.             throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
    5.         }
    6.         if (ref == null) {
    7.             //初始化
    8.             init();
    9.         }
    10.         return ref;
    11. }
    13. private void init() {
    14.     //省略的代码是用来构造参数map的，包括消费者的应用信息、模块信息、要调用的接口
    15.     ...
    16.     //创建的代理类才是调用具体方法时的入口。订阅的提供者信息将会被转化为DubboInvoker，
    17.     //DubboInvoker会被层层包装，最后封装在集群容错策略类中。dubbo最后对集群容错策略
    18.     //实例类进行代理封装，代理类才是实际的调用入口。细节随后再说    
    19.     ref = createProxy(map);
    20.     ...
    21. }
    24. private T createProxy(Map<String, String> map) {
    25.     //判断是否是本地引用，如果是的话，就不用发起远程调用了。但是本地调用dubbo服务，也不是直接调用
    26.     //实现类，消费者和提供者还是会经过各种包装类、过滤类，最终才调用到具体实现类
    27.     if (shouldJvmRefer(map)) {
    28.         URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
    29.         invoker = REF_PROTOCOL.refer(interfaceClass, url);
    30.         if (logger.isInfoEnabled()) {
    31.             logger.info("Using injvm service " + interfaceClass.getName());
    32.         }
    33.     } else {
    34.         urls.clear(); // reference retry init will add url to urls, lead to OOM
    35.         //直连方式
    36.         if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
    37.             String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
    38.             if (us != null && us.length > 0) {
    39.                 for (String u : us) {
    40.                     URL url = URL.valueOf(u);
    41.                     if (StringUtils.isEmpty(url.getPath())) {
    42.                         url = url.setPath(interfaceName);
    43.                     }
    44.                     if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    45.                         urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
    46.                     } else {
    47.                         urls.add(ClusterUtils.mergeUrl(url, map));
    48.                     }
    49.                 }
    50.             }
    51.         } else { // assemble URL from register center's configuration
    52.             // if protocols not injvm checkRegistry
    53.             if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
    54.                 checkRegistry();
    55.                 //加载注册地址
    56.                 List<URL> us = loadRegistries(false);
    57.                 if (CollectionUtils.isNotEmpty(us)) {
    58.                     for (URL u : us) {
    59.                         URL monitorUrl = loadMonitor(u);
    60.                         if (monitorUrl != null) {
    61.                             map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
    62.                         }
    63.                         //把map转为字符串添加进url中
    64.                         urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
    65.                     }
    66.                 }
    67.                 if (urls.isEmpty()) {
    68.                     throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
    69.                 }
    70.             }
    71.         }
    72.         //如果注册中心只有一个则走这段代码，否则走下面代码。但是流程和原理是相通的。
    73.         if (urls.size() == 1) {
    74.             //REF_PROTOCOL引用的是Protocol$Adaptive，这里获取的是RegistryProtocol，但是在获取
    75.             //RegistryProtocol的过程中，ExtensionLoader会对RegistryProtocol实例进行包装，包装
    76.             //类就是ProtocolListenerWrapper、QosProtocolWrapper、ProtocolFilterWrapper。在
    77.             //ProtocolFilterWrapper中还会经过filter过滤器的封装
    78.             invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
    79.         } else {
    80.             List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
    81.             URL registryURL = null;
    82.             for (URL url : urls) {
    83.                 invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
    84.                 if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    85.                     registryURL = url; // use last registry url
    86.                 }
    87.             }
    88.             if (registryURL != null) { // registry url is available
    89.                 // use RegistryAwareCluster only when register's CLUSTER is available
    90.                 URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
    91.                 // The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
    92.                 invoker = CLUSTER.join(new StaticDirectory(u, invokers));
    93.             } else { // not a registry url, must be direct invoke.
    94.                 invoker = CLUSTER.join(new StaticDirectory(invokers));
    95.             }
    96.         }
    97.     }
    99.     if (shouldCheck() && !invoker.isAvailable()) {
    100.         throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
    101.     }
    102.     if (logger.isInfoEnabled()) {
    103.         logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
    104.     }
    105.     //根据invoker创建代理类，PROXY_FACTORY引用的是ProxyFactory$Adaptive，默认是JavassistProxyFactory
    106.     //1）invoker实例指向的是MockClusterInvoker
    107.     //2）PROXY_FACTORY.getProxy(invoker)生成的实例指向的是InvokerInvocationHandler。为什么
    108.     //是InvokerInvocationHandler呢，自己去看一眼JavassistProxyFactory的代码便知
    109.     //参见图1.1
    110.     return (T) PROXY_FACTORY.getProxy(invoker);
    111. }

    image.png
    图1.1

    2、RegistryProtocol
    我们看一下这个实例：
    image.png
    图2.0

    包装Wrapper类代码：
    代码2.0

    1. public class ProtocolListenerWrapper implements Protocol {
    2.      public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    3.         //这时的url信息是注册信息，以registry开头 
    4.         if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    5.             //protocol指向QosProtocolWrapper
    6.             return protocol.refer(type, url);
    7.         }
    8.         return new ListenerInvokerWrapper<T>(protocol.refer(type, url),
    9.                 Collections.unmodifiableList(
    10.                         ExtensionLoader.getExtensionLoader(InvokerListener.class)
    11.                                 .getActivateExtension(url, INVOKER_LISTENER_KEY)));
    12.     }
    13. }    
    15. public class QosProtocolWrapper implements Protocol {
    16.      public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    17.         if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    18.             startQosServer(url);
    19.             //protocol指向ProtocolFilterWrapper
    20.             return protocol.refer(type, url);
    21.         }
    22.         return protocol.refer(type, url);
    23.     }
    24. }    
    26. public class ProtocolFilterWrapper implements Protocol {
    28.      public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    29.         if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    30.             return protocol.refer(type, url);
    31.         }
    32.         //protocol指向RegistryProtocol
    33.         return buildInvokerChain(protocol.refer(type, url), REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
    34.     }
    36.      //该方法用来构造过滤器，将invoker实例用各种filter封装。在filter中完成相应功能
    37.      private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
    38.         Invoker<T> last = invoker;
    39.         List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
    40.         if (!filters.isEmpty()) {
    41.             for (int i = filters.size() - 1; i >= 0; i--) {
    42.                 final Filter filter = filters.get(i);
    43.                 final Invoker<T> next = last;
    44.                 last = new Invoker<T>() {
    45.                     public Class<T> getInterface() {
    46.                         return invoker.getInterface();
    47.                     }
    48.                     public URL getUrl() {
    49.                         return invoker.getUrl();
    50.                     }
    51.                     public boolean isAvailable() {
    52.                         return invoker.isAvailable();
    53.                     }
    54.                     public Result invoke(Invocation invocation) throws RpcException {
    55.                         Result asyncResult;
    56.                         try {
    57.                             asyncResult = filter.invoke(next, invocation);
    58.                         } catch (Exception e) {
    59.                             // onError callback
    60.                             if (filter instanceof ListenableFilter) {
    61.                                 Filter.Listener listener = ((ListenableFilter) filter).listener();
    62.                                 if (listener != null) {
    63.                                     listener.onError(e, invoker, invocation);
    64.                                 }
    65.                             }
    66.                             throw e;
    67.                         }
    68.                         return asyncResult;
    69.                     }
    70.                     public void destroy() {
    71.                         invoker.destroy();
    72.                     }
    73.                     public String toString() {
    74.                         return invoker.toString();
    75.                     }
    76.                 };
    77.             }
    78.         }
    80.         return new CallbackRegistrationInvoker<>(last, filters);
    81.     }
    83.      static class CallbackRegistrationInvoker<T> implements Invoker<T> {
    84.         private final Invoker<T> filterInvoker;
    85.         private final List<Filter> filters;
    86.         public CallbackRegistrationInvoker(Invoker<T> filterInvoker, List<Filter> filters) {
    87.             this.filterInvoker = filterInvoker;
    88.             this.filters = filters;
    89.         }
    90.         public Result invoke(Invocation invocation) throws RpcException {
    91.             Result asyncResult = filterInvoker.invoke(invocation);
    92.             asyncResult = asyncResult.whenCompleteWithContext((r, t) -> {
    93.                 for (int i = filters.size() - 1; i >= 0; i--) {
    94.                     Filter filter = filters.get(i);
    95.                     // onResponse callback
    96.                     if (filter instanceof ListenableFilter) {
    97.                         Filter.Listener listener = ((ListenableFilter) filter).listener();
    98.                         if (listener != null) {
    99.                             if (t == null) {
    100.                                 listener.onResponse(r, filterInvoker, invocation);
    101.                             } else {
    102.                                 listener.onError(t, filterInvoker, invocation);
    103.                             }
    104.                         }
    105.                     } else {
    106.                         filter.onResponse(r, filterInvoker, invocation);
    107.                     }
    108.                 }
    109.             });
    110.             return asyncResult;
    111.         }
    112.         public Class<T> getInterface() {
    113.             return filterInvoker.getInterface();
    114.         }
    115.         public URL getUrl() {
    116.             return filterInvoker.getUrl();
    117.         }
    118.         public boolean isAvailable() {
    119.             return filterInvoker.isAvailable();
    120.         }
    121.         public void destroy() {
    122.             filterInvoker.destroy();
    123.         }
    124.     }
    125. }

    RegistryProtocol代码：
    代码2.1

    1. public class RegistryProtocol implements Protocol {
    3.     public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    4.         url = URLBuilder.from(url)
    5.                 .setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
    6.                 .removeParameter(REGISTRY_KEY)
    7.                 .build();
    8.         //获取注册实例，如果是zk的话，registry指向ZookeeperRegistry
    9.         Registry registry = registryFactory.getRegistry(url);
    10.         if (RegistryService.class.equals(type)) {
    11.             return proxyFactory.getInvoker((T) registry, type, url);
    12.         }
    13.         // group="a,b" or group="*"
    14.         Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
    15.         String group = qs.get(GROUP_KEY);
    16.         if (group != null && group.length() > 0) {
    17.             if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
    18.                 return doRefer(getMergeableCluster(), registry, type, url);
    19.             }
    20.         }
    21.         //cluster指向的是Cluster$Adaptive
    22.         return doRefer(cluster, registry, type, url);
    23.     }
    25.     private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    26.         //directory是目录，什么意思呢？通常微服务应用会部署多个实例，也就是说同一个api有多个服务者，
    27.         //那么怎么管理这些服务者呢？自然是通过相应Directory实例管理了
    28.         RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    29.         directory.setRegistry(registry);
    30.         directory.setProtocol(protocol);
    31.         // all attributes of REFER_KEY
    32.         Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
    33.         URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    34.         if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
    35.             directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
    36.             //向注册中心注册消费者信息
    37.             registry.register(directory.getRegisteredConsumerUrl());
    38.         }
    39.         //构建路由链，什么是路由呢？一个接口有十个提供者，业务上要求消费者只调用某几个提供者或者来自某
    40.         //个消费者的调用只调用某几个消费者，如何完成这些需求？通过路由规则
    41.         directory.buildRouterChain(subscribeUrl);
    42.         //订阅提供者信息。
    43.         //subscribeUrl.addParameter增加了类别参数，分别是providers、configurators、routers，将会
    44.         //从注册中心上订阅这三种类型的数据，并将订阅的信息转换成Invoker，还会开启NettyClient客户端，
    45.         //与服务者通信。这里的代码很绕，调用链参见图2.1，代码参加2.2
    46.         directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
    47.                 PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
    48.         //cluster指向Cluster$Adaptive，就是把directory封装在集群容错类中而已。看图就行，参见图2.0
    49.         //invoker指向的是MockClusterInvoker实例
    50.         Invoker invoker = cluster.join(directory);
    51.         ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
    52.         return invoker;
    53.     }
    54. }

    image.png
    图2.0

    image.png
    图2.1

    RegistryDirectory：
    代码2.2

    1. private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
    2.         Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<>();
    3.         if (urls == null || urls.isEmpty()) {
    4.             return newUrlInvokerMap;
    5.         }
    6.         Set<String> keys = new HashSet<>();
    7.         String queryProtocols = this.queryMap.get(PROTOCOL_KEY);
    8.         for (URL providerUrl : urls) {
    9.             // If protocol is configured at the reference side, only the matching protocol is selected
    10.             if (queryProtocols != null && queryProtocols.length() > 0) {
    11.                 boolean accept = false;
    12.                 String[] acceptProtocols = queryProtocols.split(",");
    13.                 for (String acceptProtocol : acceptProtocols) {
    14.                     if (providerUrl.getProtocol().equals(acceptProtocol)) {
    15.                         accept = true;
    16.                         break;
    17.                     }
    18.                 }
    19.                 if (!accept) {
    20.                     continue;
    21.                 }
    22.             }
    23.             if (EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
    24.                 continue;
    25.             }
    26.             if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
    27.                 logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() +
    28.                         " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() +
    29.                         " to consumer " + NetUtils.getLocalHost() + ", supported protocol: " +
    30.                         ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
    31.                 continue;
    32.             }
    33.             URL url = mergeUrl(providerUrl);
    35.             String key = url.toFullString(); // The parameter urls are sorted
    36.             if (keys.contains(key)) { // Repeated url
    37.                 continue;
    38.             }
    39.             keys.add(key);
    40.             // Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
    41.             Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
    42.             Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
    43.             if (invoker == null) { // Not in the cache, refer again
    44.                 try {
    45.                     boolean enabled = true;
    46.                     if (url.hasParameter(DISABLED_KEY)) {
    47.                         enabled = !url.getParameter(DISABLED_KEY, false);
    48.                     } else {
    49.                         enabled = url.getParameter(ENABLED_KEY, true);
    50.                     }
    51.                     if (enabled) {
    52.                         //这里是重点，protocol指向的是Protocol$Adaptive，到了这一步，url的协议变
    53.                         //成了dubbo协议，所以最终会调用DubboProtocol，生成DubboInvoker。
    54.                         //但是这个DubboInvoker肯定是要被ProtocolListenerWrapper、QosProtocolWrapper、
    55.                         //ProtocolFilterWrapper以及Filter过滤类进行封装的，这个过程参见RegistryProtocol
    56.                         //实例化的过程。我们来看一下DubboProtocol的代码
    57.                         invoker = new InvokerDelegate<>(protocol.refer(serviceType, url), url, providerUrl);
    58.                     }
    59.                 } catch (Throwable t) {
    60.                     logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
    61.                 }
    62.                 if (invoker != null) { // Put new invoker in cache
    63.                     newUrlInvokerMap.put(key, invoker);
    64.                 }
    65.             } else {
    66.                 newUrlInvokerMap.put(key, invoker);
    67.             }
    68.         }
    69.         keys.clear();
    70.         return newUrlInvokerMap;
    71.   }

    DubboProtocol：
    代码2.3

    1. public abstract class AbstractProtocol implements Protocol {
    3.     public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    4.         //protocolBindingRefer由具体子类实现，从这里可以看出，从dubbo2.7.3分支开始，
    5.         //消费者端的调用都是异步的了。消费者发起rpc之后，异步获取调用结果
    6.         return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
    7.     }
    8. }
    10. public class DubboProtocol extends AbstractProtocol {
    11.      //DubboProtocol的refer方法继承自父类
    12.      public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
    13.         optimizeSerialization(url);
    14.         //创建DubboInvoker，创建实例的同时还要创建netty客户端，开启通信
    15.         DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
    16.         invokers.add(invoker);
    17.         return invoker;
    18.     }
    20.     //1、微服务通常会部署多个实例，每个实例都是一个提供者
    21.     //2、消费者在启动时，会与每个提供者都建立连接，所以消费者会维护多个连接
    22.     //3、一个提供者尝尝会提供多个接口，消费者是针对每个接口都建立一个连接还是所有接口共用一个连接?
    23.     //缺省情况下，是同一个提供者下的所有接口共用一个连接。当然也可以设置不共享
    24.     private ExchangeClient[] getClients(URL url) {
    25.         //同一个提供者下的所有rpc调用是否共用一个连接
    26.         boolean useShareConnect = false;
    27.         int connections = url.getParameter(CONNECTIONS_KEY, 0);
    28.         List<ReferenceCountExchangeClient> shareClients = null;
    29.         // if not configured, connection is shared, otherwise, one connection for one service
    30.         if (connections == 0) {
    31.             useShareConnect = true;
    32.             //xml的配置方式优先级最高
    33.             String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
    34.             connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY,
    35.                     DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr);
    36.             //获取共享连接
    37.             shareClients = getSharedClient(url, connections);
    38.         }
    39.         ExchangeClient[] clients = new ExchangeClient[connections];
    40.         for (int i = 0; i < clients.length; i++) {
    41.             if (useShareConnect) {
    42.                 clients[i] = shareClients.get(i);
    44.             } else {
    45.                 clients[i] = initClient(url);
    46.             }
    47.         }
    48.         return clients;
    49.     }
    51.     //获取共享链接
    52.     private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
    53.         String key = url.getAddress();
    54.         List<ReferenceCountExchangeClient> clients = referenceClientMap.get(key);
    55.         if (checkClientCanUse(clients)) {
    56.             batchClientRefIncr(clients);
    57.             return clients;
    58.         }
    59.         locks.putIfAbsent(key, new Object());
    60.         synchronized (locks.get(key)) {
    61.             clients = referenceClientMap.get(key);
    62.             // dubbo check
    63.             if (checkClientCanUse(clients)) {
    64.                 batchClientRefIncr(clients);
    65.                 return clients;
    66.             }
    68.             // connectNum must be greater than or equal to 1
    69.             connectNum = Math.max(connectNum, 1);
    70.             // If the clients is empty, then the first initialization is
    71.             if (CollectionUtils.isEmpty(clients)) {
    72.                 clients = buildReferenceCountExchangeClientList(url, connectNum);
    73.                 referenceClientMap.put(key, clients);
    74.             } else {
    75.                 for (int i = 0; i < clients.size(); i++) {
    76.                     ReferenceCountExchangeClient referenceCountExchangeClient = clients.get(i);
    77.                     // If there is a client in the list that is no longer available, create a new one to replace him.
    78.                     if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
    79.                         clients.set(i, buildReferenceCountExchangeClient(url));
    80.                         continue;
    81.                     }
    82.                     referenceCountExchangeClient.incrementAndGetCount();
    83.                 }
    84.             }
    85.             locks.remove(key);
    86.             return clients;
    87.         }
    88.     }
    90.     //构建共享连接
    91.     private List<ReferenceCountExchangeClient> buildReferenceCountExchangeClientList(URL url, int connectNum) {
    92.         List<ReferenceCountExchangeClient> clients = new ArrayList<>();
    93.         for (int i = 0; i < connectNum; i++) {
    94.             clients.add(buildReferenceCountExchangeClient(url));
    95.         }
    96.         return clients;
    97.     }
    99.     private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) {
    100.         ExchangeClient exchangeClient = initClient(url);
    101.         return new ReferenceCountExchangeClient(exchangeClient);
    102.     }
    103.     //初始化一个链接
    104.     private ExchangeClient initClient(URL url) {
    105.         // client type setting.
    106.         String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));
    107.         url = url.addParameter(CODEC_KEY, DubboCodec.NAME);
    108.         // enable heartbeat by default
    109.         url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT));
    110.         // BIO is not allowed since it has severe performance issue.
    111.         if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
    112.             throw new RpcException("Unsupported client type: " + str + "," +
    113.                     " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
    114.         }
    115.         ExchangeClient client;
    116.         try {
    117.             // connection should be lazy
    118.             if (url.getParameter(LAZY_CONNECT_KEY, false)) {
    119.                 client = new LazyConnectExchangeClient(url, requestHandler);
    120.             } else {
    121.                 //创建netty客户端，这部分代码参见代码2.4
    122.                 client = Exchangers.connect(url, requestHandler);
    123.             }
    124.         } catch (RemotingException e) {
    125.             throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
    126.         }
    127.         return client;
    128.     }
    129. }

    Exchangers：
    代码2.4

    1. public class Exchangers {
    2.     public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
    3.         if (url == null) {
    4.             throw new IllegalArgumentException("url == null");
    5.         }
    6.         if (handler == null) {
    7.             throw new IllegalArgumentException("handler == null");
    8.         }
    9.         url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
    10.         return getExchanger(url).connect(url, handler);
    11.     }
    13.     public static Exchanger getExchanger(URL url) {
    14.         String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
    15.         return getExchanger(type);
    16.     }
    17.     //哈哈，都是dubbo spi。得到的实例是HeaderExchanger
    18.     public static Exchanger getExchanger(String type) {
    19.         return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
    20.     }
    21. }
    24. public class HeaderExchanger implements Exchanger {
    25.     public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
    26.         //封装链如下：requestHandler -> HeaderExchangeHandler -> DecodeHandler ->
    27.         // --> AllChannelHandler --> HeartbeatHandler --> MultiMessageHandler，
    28.         //MultiMessageHandler最终赋值给了NettyClient的handler属性，NettyClient也作为参数
    29.         //传给了NettyClientHandler。
    30.         //requestHandler是DubboProtocol中的属性，也是最终通道事件处理的地方
    31.         return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
    32.     }
    33. }    
    36. public class Transporters {
    37.      public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
    38.         if (url == null) {
    39.             throw new IllegalArgumentException("url == null");
    40.         }
    41.         ChannelHandler handler;
    42.         if (handlers == null || handlers.length == 0) {
    43.             handler = new ChannelHandlerAdapter();
    44.         } else if (handlers.length == 1) {
    45.             handler = handlers[0];
    46.         } else {
    47.             handler = new ChannelHandlerDispatcher(handlers);
    48.         }
    49.         //getTransporter()得到的是NettyTransporter实例 
    50.         return getTransporter().connect(url, handler);
    51.     }
    53.     public static Transporter getTransporter() {
    54.         //默认是NettyTransporter
    55.         return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
    56.     }
    57. }    
    60. public class NettyTransporter implements Transporter {
    61.     public Server bind(URL url, ChannelHandler listener) throws RemotingException {
    62.         return new NettyServer(url, listener);
    63.     }
    65.     public Client connect(URL url, ChannelHandler listener) throws RemotingException {
    66.         //实例化NettyClient的内容也很精彩，参加代码2.5
    67.         return new NettyClient(url, listener);
    68.     }
    69. }

    代码2.5

    1. public class NettyClient extends AbstractClient {
    3.      public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
    4.         // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
    5.         // the handler will be warped: MultiMessageHandler->HeartbeatHandler->handler
    6.         //wrapChannelHandler方法会创建dubbo线程池策略实例，用以处理netty通道上的连接、读、写等事件。
    7.         //线程池策略实例将会被MultiMessageHandler->HeartbeatHandler包装
    8.         super(url, wrapChannelHandler(url, handler));
    9.     }
    11.     protected static ChannelHandler wrapChannelHandler(URL url, ChannelHandler handler) {
    12.         url = ExecutorUtil.setThreadName(url, CLIENT_THREAD_POOL_NAME);
    13.         url = url.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
    14.         return ChannelHandlers.wrap(handler, url);
    15.     }
    17.     protected void doOpen() throws Throwable {
    18.         final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
    19.         bootstrap = new Bootstrap();
    20.         bootstrap.group(nioEventLoopGroup)
    21.                 .option(ChannelOption.SO_KEEPALIVE, true)
    22.                 .option(ChannelOption.TCP_NODELAY, true)
    23.                 .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
    24.                 //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
    25.                 .channel(NioSocketChannel.class);
    26.         if (getConnectTimeout() < 3000) {
    27.             bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
    28.         } else {
    29.             bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
    30.         }
    31.         bootstrap.handler(new ChannelInitializer() {
    32.             @Override
    33.             protected void initChannel(Channel ch) throws Exception {
    34.                 int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
    35.                 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
    36.                 ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
    37.                         .addLast("decoder", adapter.getDecoder())
    38.                         .addLast("encoder", adapter.getEncoder())
    39.                         .addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
    40.                         //nettyClientHandler这是通信的入口
    41.                         .addLast("handler", nettyClientHandler);
    42.                 String socksProxyHost = ConfigUtils.getProperty(SOCKS_PROXY_HOST);
    43.                 if(socksProxyHost != null) {
    44.                     int socksProxyPort = Integer.parseInt(ConfigUtils.getProperty(SOCKS_PROXY_PORT, DEFAULT_SOCKS_PROXY_PORT));
    45.                     Socks5ProxyHandler socks5ProxyHandler = new Socks5ProxyHandler(new InetSocketAddress(socksProxyHost, socksProxyPort));
    46.                     ch.pipeline().addFirst(socks5ProxyHandler);
    47.                 }
    48.             }
    49.         });
    50.     }
    52.     protected void doConnect() throws Throwable {
    53.         long start = System.currentTimeMillis();
    54.         ChannelFuture future = bootstrap.connect(getConnectAddress());
    55.         try {
    56.             boolean ret = future.awaitUninterruptibly(getConnectTimeout(), MILLISECONDS);
    57.             if (ret && future.isSuccess()) {
    58.                 Channel newChannel = future.channel();
    59.                 try {
    60.                     // Close old channel
    61.                     // copy reference
    62.                     Channel oldChannel = NettyClient.this.channel;
    63.                     if (oldChannel != null) {
    64.                         try {
    65.                             if (logger.isInfoEnabled()) {
    66.                                 logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
    67.                             }
    68.                             oldChannel.close();
    69.                         } finally {
    70.                             NettyChannel.removeChannelIfDisconnected(oldChannel);
    71.                         }
    72.                     }
    73.                 } finally {
    74.                     if (NettyClient.this.isClosed()) {
    75.                         try {
    76.                             if (logger.isInfoEnabled()) {
    77.                                 logger.info("Close new netty channel " + newChannel + ", because the client closed.");
    78.                             }
    79.                             newChannel.close();
    80.                         } finally {
    81.                             NettyClient.this.channel = null;
    82.                             NettyChannel.removeChannelIfDisconnected(newChannel);
    83.                         }
    84.                     } else {
    85.                         NettyClient.this.channel = newChannel;
    86.                     }
    87.                 }
    88.             } else if (future.cause() != null) {
    89.                 throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
    90.                         + getRemoteAddress() + ", error message is:" + future.cause().getMessage(), future.cause());
    91.             } else {
    92.                 throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
    93.                         + getRemoteAddress() + " client-side timeout "
    94.                         + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
    95.                         + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
    96.             }
    97.         } finally {
    98.             // just add new valid channel to NettyChannel's cache
    99.             if (!isConnected()) {
    100.                 //future.cancel(true);
    101.             }
    102.         }
    103.     }
    104. }
    107. public abstract class AbstractClient extends AbstractEndpoint implements Client {
    109.     public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
    110.         super(url, handler);
    111.         needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
    112.         try {
    113.             doOpen();
    114.         } catch (Throwable t) {
    115.             close();
    116.             throw new RemotingException(url.toInetSocketAddress(), null,
    117.                     "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
    118.                             + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
    119.         }
    120.         try {
    121.             // connect.
    122.             connect();
    123.             if (logger.isInfoEnabled()) {
    124.                 logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress());
    125.             }
    126.         } catch (RemotingException t) {
    127.             if (url.getParameter(Constants.CHECK_KEY, true)) {
    128.                 close();
    129.                 throw t;
    130.             } else {
    131.                 logger.warn("Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
    132.                         + " connect to the server " + getRemoteAddress() + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t);
    133.             }
    134.         } catch (Throwable t) {
    135.             close();
    136.             throw new RemotingException(url.toInetSocketAddress(), null,
    137.                     "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
    138.                             + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
    139.         }
    140.         executor = (ExecutorService) ExtensionLoader.getExtensionLoader(DataStore.class)
    141.                 .getDefaultExtension().get(CONSUMER_SIDE, Integer.toString(url.getPort()));
    142.         ExtensionLoader.getExtensionLoader(DataStore.class)
    143.                 .getDefaultExtension().remove(CONSUMER_SIDE, Integer.toString(url.getPort()));
    144.     }
    146. }
    150. public class ChannelHandlers {
    151.     public static ChannelHandler wrap(ChannelHandler handler, URL url) {
    152.         return ChannelHandlers.getInstance().wrapInternal(handler, url);
    153.     }
    155.      protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
    156.         //spi机制创建线程池策略实例，默认是all，即AllChannelHandler 
    157.         return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
    158.                 .getAdaptiveExtension().dispatch(handler, url)));
    159.     }
    160. }
    162. public class AllChannelHandler extends WrappedChannelHandler {
    163.     public AllChannelHandler(ChannelHandler handler, URL url) {
    164.         //super指WrappedChannelHandler
    165.         super(handler, url);
    166.     }
    167. }    
    169. public class WrappedChannelHandler implements ChannelHandlerDelegate {
    171.     public WrappedChannelHandler(ChannelHandler handler, URL url) {
    172.         this.handler = handler;
    173.         this.url = url;
    174.         //既然是线程池策略那当然要创建线程池了，又是spi，默认是fixed线程池，线程池的核心线程数和
    175.         //最大线程数相同，默认是200
    176.         executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
    177.         String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
    178.         if (CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(SIDE_KEY))) {
    179.             componentKey = CONSUMER_SIDE;
    180.         }
    181.         //线程池实例存储起来，优雅关机时要关闭线程池
    182.         DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
    183.         dataStore.put(componentKey, Integer.toString(url.getPort()), executor);
    184.     }
    185. }