Dubbo源码解析之客户端初始化及服务调用

一时失言乱红尘 2022-04-22 11:02 285阅读 0赞

准备

dubbo 版本：2.5.4

客户端初始化过程

初始化过程

先上时序图，帮助理解客户端初始化过程。
Dubbo客户端初始化

ReferenceBean 是客户端初始化入口，其实现 InitializingBean 接口，在 bean 初始化过程中会调用其 afterPropertiesSet 方法，进而调用 getObject() -> get() -> init() ，之后再调用 ReferenceConfig 的 createProxy() 方法。

ReferenceConfig

  1. private T createProxy(Map<String, String> map) { 
  2.     URL tmpUrl = new URL("temp", "localhost", 0, map);
  3.     final boolean isJvmRefer;
  4.     if (isInjvm() == null) { 
  5.         if (url != null && url.length() > 0) {  // 指定URL的情况下，不做本地引用
  6.             isJvmRefer = false;
  7.         } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) { 
  8.             // 默认情况下如果本地有服务暴露，则引用本地服务
  9.             isJvmRefer = true;
  10.         } else { 
  11.             isJvmRefer = false;
  12.         }
  13.     } else { 
  14.         isJvmRefer = isInjvm().booleanValue();
  15.     }
  17.     if (isJvmRefer) { 
  18.         URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
  19.         invoker = refprotocol.refer(interfaceClass, url);
  20.         if (logger.isInfoEnabled()) { 
  21.             logger.info("Using injvm service " + interfaceClass.getName());
  22.         }
  23.     } else { 
  24.         // 用户自己指定URL，可能是点对点直连地址，也可能是注册中心URL
  25.         if (url != null && url.length() > 0) {  
  26.             String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
  27.             if (us != null && us.length > 0) { 
  28.                 for (String u : us) { 
  29.                     URL url = URL.valueOf(u);
  30.                     if (url.getPath() == null || url.getPath().length() == 0) { 
  31.                         url = url.setPath(interfaceName);
  32.                     }
  33.                     if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { 
  34.                         urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
  35.                     } else { 
  36.                         urls.add(ClusterUtils.mergeUrl(url, map));
  37.                     }
  38.                 }
  39.             }
  40.         } else {  
  41.             // 通过注册中心配置拼装URL
  42.             List<URL> us = loadRegistries(false); // 从注册中心上获得相应的协议url地址
  43.             if (us != null && us.size() > 0) { 
  44.                 for (URL u : us) { 
  45.                     URL monitorUrl = loadMonitor(u);
  46.                     if (monitorUrl != null) { 
  47.                         map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
  48.                     }
  49.                     urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
  50.                 }
  51.             }
  52.             if (urls == null || urls.size() == 0) { 
  53.                 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.");
  54.             }
  55.         }
  57.         if (urls.size() == 1) { 
  58.             // refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
  59.             // refprotocol -> Protocol$Adaptive
  60.             invoker = refprotocol.refer(interfaceClass, urls.get(0));
  61.         } else { 
  62.             List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
  63.             URL registryURL = null;
  64.             for (URL url : urls) { 
  65.                 invokers.add(refprotocol.refer(interfaceClass, url));
  66.                 if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { 
  67.                     registryURL = url; // 用了最后一个registry url
  68.                 }
  69.             }
  70.             // 注册中心地址URL不为空
  71.             if (registryURL != null) {  
  72.                 // 对有注册中心的Cluster只用AvailableCluster
  73.                 URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
  74.                 // 返回的是MockClusterInvoker(FailoverClusterInvoker)
  75.                 invoker = cluster.join(new StaticDirectory(u, invokers));
  76.             }  else { 
  77.                 invoker = cluster.join(new StaticDirectory(invokers));
  78.             }
  79.         }
  80.     }
  82.     Boolean c = check;
  83.     if (c == null && consumer != null) { 
  84.         c = consumer.isCheck();
  85.     }
  86.     if (c == null) { 
  87.         c = true; // default true
  88.     }
  89.     if (c && ! invoker.isAvailable()) { 
  90.         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());
  91.     }
  92.     if (logger.isInfoEnabled()) { 
  93.         logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
  94.     }
  95.     // 创建服务代理
  96.     return (T) proxyFactory.getProxy(invoker);
  97. }

Protocol$Adpative

  1. public class Protocol$Adpative implements com.alibaba.dubbo.rpc.Protocol { 
  2.     public void destroy() { 
  3.         throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
  4.     }
  6.     public int getDefaultPort() { 
  7.         throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
  8.     }
  10.     public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { 
  11.         if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
  12.         if (arg0.getUrl() == null)
  13.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");
  14.         com.alibaba.dubbo.common.URL url = arg0.getUrl();
  15.         String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
  16.         if (extName == null)
  17.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
  18.         com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
  19.         return extension.export(arg0);
  20.     }
  22.     public com.alibaba.dubbo.rpc.Invoker refer(Class arg0, com.alibaba.dubbo.common.URL arg1) throws com.alibaba.dubbo.rpc.RpcException { 
  23.         if (arg1 == null) throw new IllegalArgumentException("url == null");
  24.         com.alibaba.dubbo.common.URL url = arg1;
  25.         String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
  26.         if (extName == null)
  27.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
  28.         // extName -> registry
  29.         // extension -> RegistryProtocol
  30.         com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
  31.         return extension.refer(arg0, arg1);
  32.     }
  33. }

RegistryProtocol

  1. public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { 
  2.     url = url.setProtocol(url.getParameter("registry", "dubbo")).removeParameter("registry");
  3.     Registry registry = this.registryFactory.getRegistry(url);
  4.     if (RegistryService.class.equals(type)) { 
  5.         return this.proxyFactory.getInvoker(registry, type, url);
  6.     } else { 
  7.         Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded("refer"));
  8.         String group = (String)qs.get("group");
  9.         return group == null || group.length() <= 0 || Constants.COMMA_SPLIT_PATTERN.split(group).length <= 1 && !"*".equals(group) ? this.doRefer(this.cluster, registry, type, url) : this.doRefer(this.getMergeableCluster(), registry, type, url);
  10.     }
  11. }

RegistryProtocol.doRefer方法存在一个参数 Cluster ，而 Cluster 是一个扩展点，存在加在方法级别上的 @Adaptive 注解，说明会动态生成自适应适配器( Cluster$Adaptive )。在 RegistryProtocol 中存在 Cluster 扩展点成员变量及 setter 方法，说明是一个自动注入的扩展点。

  1. @SPI("failover")
  2. public interface Cluster { 
  3.     @Adaptive
  4.     <T> Invoker<T> join(Directory<T> var1) throws RpcException;
  5. }
  7. // RegistryProtocol
  8. private Cluster cluster;
  10. public void setCluster(Cluster cluster) { 
  11.     this.cluster = cluster;
  12. }

Cluster$Adpative

  1. public class Cluster$Adpative implements com.alibaba.dubbo.rpc.cluster.Cluster { 
  3.     public com.alibaba.dubbo.rpc.Invoker join(com.alibaba.dubbo.rpc.cluster.Directory arg0) throws com.alibaba.dubbo.rpc.RpcException { 
  4.         if (arg0 == null)
  5.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument == null");
  6.         if (arg0.getUrl() == null)
  7.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument getUrl() == null");
  8.         com.alibaba.dubbo.common.URL url = arg0.getUrl();
  9.         String extName = url.getParameter("cluster", "failover");
  10.         if (extName == null)
  11.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.cluster.Cluster) name from url(" + url.toString() + ") use keys([cluster])");
  12.         com.alibaba.dubbo.rpc.cluster.Cluster extension = (com.alibaba.dubbo.rpc.cluster.Cluster) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.cluster.Cluster.class).getExtension(extName);
  13.         return extension.join(arg0);
  14.     }
  15. }

又因为 Cluster 扩展点实现中存在以扩展点作为参数的构造方法，所以会被 Wrapper 装饰，而该装饰器就是 MockClusterWrapper 。

  1. public class MockClusterWrapper implements Cluster { 
  2.     private Cluster cluster;
  4.     public MockClusterWrapper(Cluster cluster) { 
  5.         this.cluster = cluster;
  6.     }
  8.     public <T> Invoker<T> join(Directory<T> directory) throws RpcException { 
  9.         return new MockClusterInvoker(directory, this.cluster.join(directory));
  10.     }
  11. }
  13. // RegistryProtocol
  14. private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { 
  15.     RegistryDirectory<T> directory = new RegistryDirectory(type, url);
  16.     directory.setRegistry(registry); // registry -> ZookeeperRegistry
  17.     directory.setProtocol(this.protocol); // protocol -> Protocol$Adaptive
  18.     URL subscribeUrl = new URL("consumer", NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
  19.     if (!"*".equals(url.getServiceInterface()) && url.getParameter("register", true)) { 
  20.         // 注册consumer://协议地址到注册中心
  21.         registry.register(subscribeUrl.addParameters(new String[]{ "category", "consumers", "check", String.valueOf(false)}));
  22.     }
  23.     // 注册zookeeper地址变更
  24.     directory.subscribe(subscribeUrl.addParameter("category", "providers,configurators,routers"));
  25.     // 返回一个MockClusterInvoker(FailoverClusterInvoker)
  26.     return cluster.join(directory);
  27. }

MockClusterWrapper

  1. public <T> Invoker<T> join(Directory<T> directory) throws RpcException { 
  2.     return new MockClusterInvoker(directory, this.cluster.join(directory));
  3. }

FailoverCluster

  1. public <T> Invoker<T> join(Directory<T> directory) throws RpcException { 
  2.     return new FailoverClusterInvoker(directory);
  3. }

回到ReferenceConfig.createProxy中的proxyFactory.getProxy()

  1. private T createProxy(Map<String, String> map) { 
  2.     ......
  3.     // 创建服务代理
  4.     // proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
  5.     // proxyFactory -> ProxyFactory$Adpative
  6.     return (T) proxyFactory.getProxy(invoker);
  7. }

ProxyFactory$Adpative

  1. public class ProxyFactory$Adpative implements com.alibaba.dubbo.rpc.ProxyFactory { 
  2.     public java.lang.Object getProxy(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { 
  3.         if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
  4.         if (arg0.getUrl() == null)
  5.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");
  6.         com.alibaba.dubbo.common.URL url = arg0.getUrl();
  7.         String extName = url.getParameter("proxy", "javassist");
  8.         if (extName == null)
  9.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
  10.         // extName -> javassist
  11.         // extension -> StubProxyFactoryWrapper(JavassistProxyFactory)
  12.         com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
  13.         return extension.getProxy(arg0);
  14.     }
  15.     ......
  16. }

StubProxyFactoryWrapper

  1. public <T> T getProxy(Invoker<T> invoker) throws RpcException { 
  2.     T proxy = this.proxyFactory.getProxy(invoker);
  3.     if (GenericService.class != invoker.getInterface()) { 
  4.         String stub = invoker.getUrl().getParameter("stub", invoker.getUrl().getParameter("local"));
  5.         if (ConfigUtils.isNotEmpty(stub)) { 
  6.             Class<?> serviceType = invoker.getInterface();
  7.             if (ConfigUtils.isDefault(stub)) { 
  8.                 if (invoker.getUrl().hasParameter("stub")) { 
  9.                     stub = serviceType.getName() + "Stub";
  10.                 } else { 
  11.                     stub = serviceType.getName() + "Local";
  12.                 }
  13.             }
  15.             try { 
  16.                 Class<?> stubClass = ReflectUtils.forName(stub);
  17.                 if (!serviceType.isAssignableFrom(stubClass)) { 
  18.                     throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + serviceType.getName());
  19.                 }
  21.                 try { 
  22.                     Constructor<?> constructor = ReflectUtils.findConstructor(stubClass, serviceType);
  23.                     proxy = constructor.newInstance(proxy);
  24.                     URL url = invoker.getUrl();
  25.                     if (url.getParameter("dubbo.stub.event", false)) { 
  26.                         url = url.addParameter("dubbo.stub.event.methods", StringUtils.join(Wrapper.getWrapper(proxy.getClass()).getDeclaredMethodNames(), ","));
  27.                         url = url.addParameter("isserver", Boolean.FALSE.toString());
  29.                         try { 
  30.                             this.export(proxy, invoker.getInterface(), url);
  31.                         } catch (Exception var9) { 
  32.                             LOGGER.error("export a stub service error.", var9);
  33.                         }
  34.                     }
  35.                 } catch (NoSuchMethodException var10) { 
  36.                     throw new IllegalStateException("No such constructor \"public " + stubClass.getSimpleName() + "(" + serviceType.getName() + ")\" in stub implemention class " + stubClass.getName(), var10);
  37.                 }
  38.             } catch (Throwable var11) { 
  39.                 LOGGER.error("Failed to create stub implemention class " + stub + " in consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", cause: " + var11.getMessage(), var11);
  40.             }
  41.         }
  42.     }
  44.     return proxy;
  45. }

JavassistProxyFactory

  1. public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) { 
  2.     return Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
  3. }

客户端与服务端创建连接过程

客户端与服务端创建连接时序图。
Dubbo客户端与服务端创建连接过程

RegistryProtocol

  1. private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { 
  2.     RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
  3.     directory.setRegistry(registry);
  4.     directory.setProtocol(protocol);
  5.     URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
  6.     if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
  7.             && url.getParameter(Constants.REGISTER_KEY, true)) { 
  8.         registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
  9.                 Constants.CHECK_KEY, String.valueOf(false)));
  10.     }
  11.     // 客户端与服务端创建连接过程
  12.     directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
  13.             Constants.PROVIDERS_CATEGORY 
  14.             + "," + Constants.CONFIGURATORS_CATEGORY 
  15.             + "," + Constants.ROUTERS_CATEGORY));
  16.     return cluster.join(directory);
  17. }

RegistryDirectory

  1. public void subscribe(URL url) { 
  2.     setConsumerUrl(url);
  3.     registry.subscribe(url, this);
  4. }

FailbackRegistry

  1. public void subscribe(URL url, NotifyListener listener) { 
  2.     super.subscribe(url, listener);
  3.     removeFailedSubscribed(url, listener);
  4.     try { 
  5.         // 向服务器端发送订阅请求
  6.         doSubscribe(url, listener);
  7.     } catch (Exception e) { 
  8.         Throwable t = e;
  10.         List<URL> urls = getCacheUrls(url);
  11.         if (urls != null && urls.size() > 0) { 
  12.             notify(url, listener, urls);
  13.             logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
  14.         } else { 
  15.             // 如果开启了启动时检测，则直接抛出异常
  16.             boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
  17.                     && url.getParameter(Constants.CHECK_KEY, true);
  18.             boolean skipFailback = t instanceof SkipFailbackWrapperException;
  19.             if (check || skipFailback) { 
  20.                 if(skipFailback) { 
  21.                     t = t.getCause();
  22.                 }
  23.                 throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t);
  24.             } else { 
  25.                 logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  26.             }
  27.         }
  29.         // 将失败的订阅请求记录到失败列表，定时重试
  30.         addFailedSubscribed(url, listener);
  31.     }
  32. }

ZookeeperRegistry

  1. protected void doSubscribe(final URL url, final NotifyListener listener) { 
  2.     try { 
  3.         if (Constants.ANY_VALUE.equals(url.getServiceInterface())) { 
  4.             String root = toRootPath();
  5.             ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
  6.             if (listeners == null) { 
  7.                 zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
  8.                 listeners = zkListeners.get(url);
  9.             }
  10.             ChildListener zkListener = listeners.get(listener);
  11.             if (zkListener == null) { 
  12.                 listeners.putIfAbsent(listener, new ChildListener() { 
  13.                     public void childChanged(String parentPath, List<String> currentChilds) { 
  14.                         for (String child : currentChilds) { 
  15.             child = URL.decode(child);
  16.                             if (! anyServices.contains(child)) { 
  17.                                 anyServices.add(child);
  18.                                 subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child, 
  19.                                         Constants.CHECK_KEY, String.valueOf(false)), listener);
  20.                             }
  21.                         }
  22.                     }
  23.                 });
  24.                 zkListener = listeners.get(listener);
  25.             }
  26.             zkClient.create(root, false);
  27.             List<String> services = zkClient.addChildListener(root, zkListener);
  28.             if (services != null && services.size() > 0) { 
  29.                 for (String service : services) { 
  30.       service = URL.decode(service);
  31.       anyServices.add(service);
  32.                     subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service, 
  33.                             Constants.CHECK_KEY, String.valueOf(false)), listener);
  34.                 }
  35.             }
  36.         } else { 
  37.             List<URL> urls = new ArrayList<URL>();
  38.             for (String path : toCategoriesPath(url)) { 
  39.                 ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
  40.                 if (listeners == null) { 
  41.                     zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
  42.                     listeners = zkListeners.get(url);
  43.                 }
  44.                 ChildListener zkListener = listeners.get(listener);
  45.                 if (zkListener == null) { 
  46.                     listeners.putIfAbsent(listener, new ChildListener() { 
  47.                         public void childChanged(String parentPath, List<String> currentChilds) { 
  48.                            ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
  49.                         }
  50.                     });
  51.                     zkListener = listeners.get(listener);
  52.                 }
  53.                 zkClient.create(path, false);
  54.                 List<String> children = zkClient.addChildListener(path, zkListener);
  55.                 if (children != null) { 
  56.                    urls.addAll(toUrlsWithEmpty(url, path, children));
  57.                 }
  58.             }
  59.             notify(url, listener, urls);
  60.         }
  61.     } catch (Throwable e) { 
  62.         throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  63.     }
  64. }

FailbackRegistry

  1. protected void notify(URL url, NotifyListener listener, List<URL> urls) { 
  2.     if (url == null) { 
  3.         throw new IllegalArgumentException("notify url == null");
  4.     }
  5.     if (listener == null) { 
  6.         throw new IllegalArgumentException("notify listener == null");
  7.     }
  8.     try { 
  9.        doNotify(url, listener, urls);
  10.     } catch (Exception t) { 
  11.         // 将失败的通知请求记录到失败列表，定时重试
  12.         Map<NotifyListener, List<URL>> listeners = failedNotified.get(url);
  13.         if (listeners == null) { 
  14.             failedNotified.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, List<URL>>());
  15.             listeners = failedNotified.get(url);
  16.         }
  17.         listeners.put(listener, urls);
  18.         logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  19.     }
  20. }
  22. protected void doNotify(URL url, NotifyListener listener, List<URL> urls) { 
  23.     super.notify(url, listener, urls);
  24. }

AbstractRegistry

  1. protected void notify(URL url, NotifyListener listener, List<URL> urls) { 
  2.     if (url == null) { 
  3.         throw new IllegalArgumentException("notify url == null");
  4.     }
  5.     if (listener == null) { 
  6.         throw new IllegalArgumentException("notify listener == null");
  7.     }
  8.     if ((urls == null || urls.size() == 0) 
  9.             && ! Constants.ANY_VALUE.equals(url.getServiceInterface())) { 
  10.         logger.warn("Ignore empty notify urls for subscribe url " + url);
  11.         return;
  12.     }
  13.     if (logger.isInfoEnabled()) { 
  14.         logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
  15.     }
  16.     Map<String, List<URL>> result = new HashMap<String, List<URL>>();
  17.     for (URL u : urls) { 
  18.         if (UrlUtils.isMatch(url, u)) { 
  19.            String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
  20.            List<URL> categoryList = result.get(category);
  21.            if (categoryList == null) { 
  22.               categoryList = new ArrayList<URL>();
  23.               result.put(category, categoryList);
  24.            }
  25.            categoryList.add(u);
  26.         }
  27.     }
  28.     if (result.size() == 0) { 
  29.         return;
  30.     }
  31.     Map<String, List<URL>> categoryNotified = notified.get(url);
  32.     if (categoryNotified == null) { 
  33.         notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
  34.         categoryNotified = notified.get(url);
  35.     }
  36.     // 对providers、configurators、routers路径下的进行更新
  37.     for (Map.Entry<String, List<URL>> entry : result.entrySet()) { 
  38.         String category = entry.getKey();
  39.         List<URL> categoryList = entry.getValue();
  40.         categoryNotified.put(category, categoryList);
  41.         saveProperties(url);
  42.         listener.notify(categoryList);
  43.     }
  44. }

RegistryDirectory

  1. public synchronized void notify(List<URL> urls) { 
  2.     List<URL> invokerUrls = new ArrayList<URL>();
  3.     List<URL> routerUrls = new ArrayList<URL>();
  4.     List<URL> configuratorUrls = new ArrayList<URL>();
  5.     for (URL url : urls) { 
  6.         String protocol = url.getProtocol();
  7.         String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
  8.         if (Constants.ROUTERS_CATEGORY.equals(category) 
  9.                 || Constants.ROUTE_PROTOCOL.equals(protocol)) { 
  10.             routerUrls.add(url);
  11.         } else if (Constants.CONFIGURATORS_CATEGORY.equals(category) 
  12.                 || Constants.OVERRIDE_PROTOCOL.equals(protocol)) { 
  13.             configuratorUrls.add(url);
  14.         } else if (Constants.PROVIDERS_CATEGORY.equals(category)) { 
  15.             invokerUrls.add(url);
  16.         } else { 
  17.             logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
  18.         }
  19.     }
  20.     // configurators 
  21.     if (configuratorUrls != null && configuratorUrls.size() >0 ){ 
  22.         this.configurators = toConfigurators(configuratorUrls);
  23.     }
  24.     // routers
  25.     if (routerUrls != null && routerUrls.size() >0 ){ 
  26.         List<Router> routers = toRouters(routerUrls);
  27.         if(routers != null){  // null - do nothing
  28.             setRouters(routers);
  29.         }
  30.     }
  31.     List<Configurator> localConfigurators = this.configurators; // local reference
  32.     // 合并override参数
  33.     this.overrideDirectoryUrl = directoryUrl;
  34.     if (localConfigurators != null && localConfigurators.size() > 0) { 
  35.         for (Configurator configurator : localConfigurators) { 
  36.             this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
  37.         }
  38.     }
  39.     // providers
  40.     // 刷新Invoker>>
  41.     refreshInvoker(invokerUrls);
  42. }
  44. /** * 根据invokerURL列表转换为invoker列表。转换规则如下： * 1.如果url已经被转换为invoker，则不在重新引用，直接从缓存中获取，注意如果url中任何一个参数变更也会重新引用 * 2.如果传入的invoker列表不为空，则表示最新的invoker列表 * 3.如果传入的invokerUrl列表是空，则表示只是下发的override规则或route规则，需要重新交叉对比，决定是否需要重新引用。 * @param invokerUrls 传入的参数不能为null */
  45. // RegistryDirectory
  46. private void refreshInvoker(List<URL> invokerUrls){ 
  47.     if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
  48.             && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) { 
  49.         this.forbidden = true; // 禁止访问
  50.         this.methodInvokerMap = null; // 置空列表
  51.         destroyAllInvokers(); // 关闭所有Invoker
  52.     } else { 
  53.         this.forbidden = false; // 允许访问
  54.         Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
  55.         if (invokerUrls.size() == 0 && this.cachedInvokerUrls != null){ 
  56.             invokerUrls.addAll(this.cachedInvokerUrls);
  57.         } else { 
  58.             this.cachedInvokerUrls = new HashSet<URL>();
  59.             this.cachedInvokerUrls.addAll(invokerUrls);//缓存invokerUrls列表，便于交叉对比
  60.         }
  61.         if (invokerUrls.size() ==0 ){ 
  62.            return;
  63.         }
  64.         Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
  65.         Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 换方法名映射Invoker列表
  66.         // state change
  67.         //如果计算错误，则不进行处理.
  68.         if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0 ){ 
  69.             logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :"+invokerUrls.size() + ", invoker.size :0. urls :"+invokerUrls.toString()));
  70.             return ;
  71.         }
  72.         this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
  73.         this.urlInvokerMap = newUrlInvokerMap;
  74.         try{ 
  75.             destroyUnusedInvokers(oldUrlInvokerMap,newUrlInvokerMap); // 关闭未使用的Invoker
  76.         }catch (Exception e) { 
  77.             logger.warn("destroyUnusedInvokers error. ", e);
  78.         }
  79.     }
  80. }
  82. // RegistryDirectory
  83. private Map<String, Invoker<T>> toInvokers(List<URL> urls) { 
  84.     Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
  85.     if(urls == null || urls.size() == 0){ 
  86.         return newUrlInvokerMap;
  87.     }
  88.     Set<String> keys = new HashSet<String>();
  89.     String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
  90.     for (URL providerUrl : urls) { 
  91.        //如果reference端配置了protocol，则只选择匹配的protocol
  92.        if (queryProtocols != null && queryProtocols.length() >0) { 
  93.           boolean accept = false;
  94.           String[] acceptProtocols = queryProtocols.split(",");
  95.           for (String acceptProtocol : acceptProtocols) { 
  96.              if (providerUrl.getProtocol().equals(acceptProtocol)) { 
  97.                 accept = true;
  98.                 break;
  99.              }
  100.           }
  101.           if (!accept) { 
  102.              continue;
  103.           }
  104.        }
  105.         if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) { 
  106.             continue;
  107.         }
  108.         if (! ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) { 
  109.             logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() + " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost() 
  110.                     + ", supported protocol: "+ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
  111.             continue;
  112.         }
  113.         URL url = mergeUrl(providerUrl);
  115.         String key = url.toFullString(); // URL参数是排序的
  116.         if (keys.contains(key)) {  // 重复URL
  117.             continue;
  118.         }
  119.         keys.add(key);
  120.         // 缓存key为没有合并消费端参数的URL，不管消费端如何合并参数，如果服务端URL发生变化，则重新refer
  121.         Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
  122.         Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
  123.         if (invoker == null) {  // 缓存中没有，重新refer
  124.             try { 
  125.                boolean enabled = true;
  126.                if (url.hasParameter(Constants.DISABLED_KEY)) { 
  127.                   enabled = ! url.getParameter(Constants.DISABLED_KEY, false);
  128.                } else { 
  129.                   enabled = url.getParameter(Constants.ENABLED_KEY, true);
  130.                }
  131.                if (enabled) { 
  132.                   /** * url -> dubbo://...... * providerUrl -> dubbo://...... * protocol -> Protocol$Adaptive * extension -> getExtension("dubbo") -> ProtocolFilterWrapper(ProtocolListenerWrapper(DubboProtocol)) * 最后调用DubboProtocol的refer方法 */
  133.                   invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
  134.                }
  135.             } catch (Throwable t) { 
  136.                 logger.error("Failed to refer invoker for interface:"+serviceType+",url:("+url+")" + t.getMessage(), t);
  137.             }
  138.             if (invoker != null) {  // 将新的引用放入缓存
  139.                 newUrlInvokerMap.put(key, invoker);
  140.             }
  141.         }else { 
  142.             newUrlInvokerMap.put(key, invoker);
  143.         }
  144.     }
  145.     keys.clear();
  146.     return newUrlInvokerMap;
  147. }

DubboProtocol

  1. public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException { 
  2.     // create rpc invoker.
  3.     DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
  4.     invokers.add(invoker);
  5.     return invoker;
  6. }
  8. // DubboProtocol
  9. private ExchangeClient[] getClients(URL url){ 
  10.     //是否共享连接
  11.     boolean service_share_connect = false;
  12.     int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
  13.     //如果connections不配置，则共享连接，否则每服务每连接
  14.     if (connections == 0){ 
  15.         service_share_connect = true;
  16.         connections = 1;
  17.     }
  19.     ExchangeClient[] clients = new ExchangeClient[connections];
  20.     for (int i = 0; i < clients.length; i++) { 
  21.         if (service_share_connect){ 
  22.             clients[i] = getSharedClient(url);
  23.         } else { 
  24.             // 初始化客户端
  25.             clients[i] = initClient(url);
  26.         }
  27.     }
  28.     return clients;
  29. }
  31. // DubboProtocol
  32. private ExchangeClient initClient(URL url) { 
  34.     // client type setting.
  35.     String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
  37.     String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
  38.     boolean compatible = (version != null && version.startsWith("1.0."));
  39.     url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
  40.     //默认开启heartbeat
  41.     url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
  43.     // BIO存在严重性能问题，暂时不允许使用
  44.     if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { 
  45.         throw new RpcException("Unsupported client type: " + str + "," +
  46.                 " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
  47.     }
  49.     ExchangeClient client ;
  50.     try { 
  51.         //设置连接应该是lazy的 
  52.         if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)){ 
  53.             client = new LazyConnectExchangeClient(url ,requestHandler);
  54.         } else { 
  55.             // HeaderExchanger
  56.             client = Exchangers.connect(url ,requestHandler);
  57.         }
  58.     } catch (RemotingException e) { 
  59.         throw new RpcException("Fail to create remoting client for service(" + url
  60.                 + "): " + e.getMessage(), e);
  61.     }
  62.     return client;
  63. }

HeaderExchanger

  1. public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException { 
  2.     return new HeaderExchangeClient(Transporters.connect(url, new ChannelHandler[]{ new DecodeHandler(new HeaderExchangeHandler(handler))}));
  3. }

NettyTransporter

  1. public Client connect(URL url, ChannelHandler listener) throws RemotingException { 
  2.     // 创建netty客户端
  3.     return new NettyClient(url, listener);
  4. }

客户端初始化过程完成。

服务调用过程

服务调用过程时序图。
Dubbo服务调用过程

客户端生成 Proxy0 代理

  • client：Proxy0
  • server：JavassistProxyFactory

Proxy0

  1. public java.lang.String sayHello(new InvokerInvocationHandler(invoker)) { 
  2.     Object[] args = new Object[1];
  3.     args[0] = ($w) $1;
  4.     Object ret = handler.invoke(this, methods[0], args);
  5.     return (java.lang.String) ret;
  6. }

InvokerInvocationHandler

  1. public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
  2.     String methodName = method.getName();
  3.     Class<?>[] parameterTypes = method.getParameterTypes();
  4.     if (method.getDeclaringClass() == Object.class) { 
  5.         return method.invoke(this.invoker, args);
  6.     } else if ("toString".equals(methodName) && parameterTypes.length == 0) { 
  7.         return this.invoker.toString();
  8.     } else if ("hashCode".equals(methodName) && parameterTypes.length == 0) { 
  9.         return this.invoker.hashCode();
  10.     } else { 
  11.         // invoker -> MockClusterInvoker
  12.         return "equals".equals(methodName) && parameterTypes.length == 1 ? this.invoker.equals(args[0]) : this.invoker.invoke(new RpcInvocation(method, args)).recreate();
  13.     }
  14. }

MockClusterInvoker

  1. public Result invoke(Invocation invocation) throws RpcException { 
  2.     Result result = null;
  3.     String value = this.directory.getUrl().getMethodParameter(invocation.getMethodName(), "mock", Boolean.FALSE.toString()).trim();
  4.     if (value.length() != 0 && !value.equalsIgnoreCase("false")) { 
  5.         if (value.startsWith("force")) { 
  6.             if (logger.isWarnEnabled()) { 
  7.                 logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + this.directory.getUrl());
  8.             }
  9.             // 直接调用mock
  10.             result = this.doMockInvoke(invocation, (RpcException)null);
  11.         } else { 
  12.             try { 
  13.                 result = this.invoker.invoke(invocation);
  14.             } catch (RpcException var5) { 
  15.                 if (var5.isBiz()) { 
  16.                     throw var5;
  17.                 }
  19.                 if (logger.isWarnEnabled()) { 
  20.                     logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + this.directory.getUrl(), var5);
  21.                 }
  22.                 // 调用异常执行mock方法
  23.                 result = this.doMockInvoke(invocation, var5);
  24.             }
  25.         }
  26.     } else { 
  27.         // 调用执行
  28.         // invoker -> FailoverClusterInvoker invoke方法在其父类AbstractClusterInvoker中
  29.         result = this.invoker.invoke(invocation);
  30.     }
  32.     return result;
  33. }

AbstractClusterInvoker

  1. public Result invoke(Invocation invocation) throws RpcException { 
  2.     this.checkWhetherDestroyed();
  3.     // 获取所有invoker列表>>
  4.     List<Invoker<T>> invokers = this.list(invocation);
  5.     LoadBalance loadbalance;
  6.     if (invokers != null && invokers.size() > 0) { 
  7.         loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(((Invoker)invokers.get(0)).getUrl().getMethodParameter(invocation.getMethodName(), "loadbalance", "random"));
  8.     } else { 
  9.         loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension("random");
  10.     }
  12.     RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
  13.     // 执行调用
  14.     return this.doInvoke(invocation, invokers, loadbalance);
  15. }
  17. // 获取所有invoker列表
  18. protected List<Invoker<T>> list(Invocation invocation) throws RpcException { 
  19.     List<Invoker<T>> invokers = this.directory.list(invocation);
  20.     return invokers;
  21. }

AbstractDirectory

  1. public List<Invoker<T>> list(Invocation invocation) throws RpcException { 
  2.     if (this.destroyed) { 
  3.         throw new RpcException("Directory already destroyed .url: " + this.getUrl());
  4.     } else { 
  5.         // 执行获取所有invoker列表
  6.         // invoker -> ProtocolFilterWrapper
  7.         List<Invoker<T>> invokers = this.doList(invocation);
  8.         List<Router> localRouters = this.routers;
  9.         if (localRouters != null && localRouters.size() > 0) { 
  10.             Iterator var4 = localRouters.iterator();
  12.             while(var4.hasNext()) { 
  13.                 Router router = (Router)var4.next();
  15.                 try { 
  16.                     if (router.getUrl() == null || router.getUrl().getParameter("runtime", true)) { 
  17.                         invokers = router.route(invokers, this.getConsumerUrl(), invocation);
  18.                     }
  19.                 } catch (Throwable var7) { 
  20.                     logger.error("Failed to execute router: " + this.getUrl() + ", cause: " + var7.getMessage(), var7);
  21.                 }
  22.             }
  23.         }
  25.         return invokers;
  26.     }
  27. }

RegistryDirectory

  1. public List<Invoker<T>> doList(Invocation invocation) { 
  2.     if (this.forbidden) { 
  3.         throw new RpcException(4, "No provider available from registry " + this.getUrl().getAddress() + " for service " + this.getConsumerUrl().getServiceKey() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", may be providers disabled or not registered ?");
  4.     } else { 
  5.         List<Invoker<T>> invokers = null;
  6.         Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;
  7.         if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) { 
  8.             String methodName = RpcUtils.getMethodName(invocation);
  9.             Object[] args = RpcUtils.getArguments(invocation);
  10.             if (args != null && args.length > 0 && args[0] != null && (args[0] instanceof String || args[0].getClass().isEnum())) { 
  11.                 invokers = (List)localMethodInvokerMap.get(methodName + "." + args[0]);
  12.             }
  14.             if (invokers == null) { 
  15.                 invokers = (List)localMethodInvokerMap.get(methodName);
  16.             }
  18.             if (invokers == null) { 
  19.                 invokers = (List)localMethodInvokerMap.get("*");
  20.             }
  22.             if (invokers == null) { 
  23.                 Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
  24.                 if (iterator.hasNext()) { 
  25.                     invokers = (List)iterator.next();
  26.                 }
  27.             }
  28.         }
  30.         return (List)(invokers == null ? new ArrayList(0) : invokers);
  31.     }
  32. }

回到 AbstractClusterInvoker 的 invoke 方法，最后会调用 FailoverClusterInvoker 的 doInvoke 方法

FailoverClusterInvoker

  1. public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
  2.     List<Invoker<T>> copyinvokers = invokers;
  3.     this.checkInvokers(invokers, invocation);
  4.     int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries", 2) + 1;
  5.     if (len <= 0) { 
  6.         len = 1;
  7.     }
  9.     RpcException le = null;
  10.     List<Invoker<T>> invoked = new ArrayList(invokers.size());
  11.     Set<String> providers = new HashSet(len);
  13.     for(int i = 0; i < len; ++i) { 
  14.         if (i > 0) { 
  15.             this.checkWhetherDestroyed();
  16.             copyinvokers = this.list(invocation);
  17.             this.checkInvokers(copyinvokers, invocation);
  18.         }
  19.         // 选择调用invoker>>
  20.         Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
  21.         invoked.add(invoker);
  22.         RpcContext.getContext().setInvokers(invoked);
  24.         try { 
  25.             // 执行调用
  26.             Result result = invoker.invoke(invocation);
  27.             if (le != null && logger.isWarnEnabled()) { 
  28.                 logger.warn("Although retry the method " + invocation.getMethodName() + " in the service " + this.getInterface().getName() + " was successful by the provider " + invoker.getUrl().getAddress() + ", but there have been failed providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + le.getMessage(), le);
  29.             }
  31.             Result var12 = result;
  32.             return var12;
  33.         } catch (RpcException var17) { 
  34.             if (var17.isBiz()) { 
  35.                 throw var17;
  36.             }
  38.             le = var17;
  39.         } catch (Throwable var18) { 
  40.             le = new RpcException(var18.getMessage(), var18);
  41.         } finally { 
  42.             providers.add(invoker.getUrl().getAddress());
  43.         }
  44.     }
  46.     throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method " + invocation.getMethodName() + " in the service " + this.getInterface().getName() + ". Tried " + len + " times of the providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + (le != null ? le.getMessage() : ""), (Throwable)(le != null && le.getCause() != null ? le.getCause() : le));
  47. }
  49. // AbstractClusterInvoker
  50. protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException { 
  51.     if (invokers != null && invokers.size() != 0) { 
  52.         String methodName = invocation == null ? "" : invocation.getMethodName();
  53.         boolean sticky = ((Invoker)invokers.get(0)).getUrl().getMethodParameter(methodName, "sticky", false);
  54.         if (this.stickyInvoker != null && !invokers.contains(this.stickyInvoker)) { 
  55.             this.stickyInvoker = null;
  56.         }
  58.         if (sticky && this.stickyInvoker != null && (selected == null || !selected.contains(this.stickyInvoker)) && this.availablecheck && this.stickyInvoker.isAvailable()) { 
  59.             return this.stickyInvoker;
  60.         } else { 
  61.             Invoker<T> invoker = this.doselect(loadbalance, invocation, invokers, selected);
  62.             if (sticky) { 
  63.                 this.stickyInvoker = invoker;
  64.             }
  66.             return invoker;
  67.         }
  68.     } else { 
  69.         return null;
  70.     }
  71. }
  73. // AbstractClusterInvoker
  74. private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException { 
  75.     if (invokers != null && invokers.size() != 0) { 
  76.         if (invokers.size() == 1) { 
  77.             return (Invoker)invokers.get(0);
  78.         } else if (invokers.size() == 2 && selected != null && selected.size() > 0) { 
  79.             return selected.get(0) == invokers.get(0) ? (Invoker)invokers.get(1) : (Invoker)invokers.get(0);
  80.         } else { 
  81.             // 负载均衡策略执行选择
  82.             Invoker<T> invoker = loadbalance.select(invokers, this.getUrl(), invocation);
  83.             if (selected != null && selected.contains(invoker) || !invoker.isAvailable() && this.getUrl() != null && this.availablecheck) { 
  84.                 try { 
  85.                     Invoker<T> rinvoker = this.reselect(loadbalance, invocation, invokers, selected, this.availablecheck);
  86.                     if (rinvoker != null) { 
  87.                         invoker = rinvoker;
  88.                     } else { 
  89.                         int index = invokers.indexOf(invoker);
  91.                         try { 
  92.                             invoker = index < invokers.size() - 1 ? (Invoker)invokers.get(index + 1) : invoker;
  93.                         } catch (Exception var9) { 
  94.                             logger.warn(var9.getMessage() + " may because invokers list dynamic change, ignore.", var9);
  95.                         }
  96.                     }
  97.                 } catch (Throwable var10) { 
  98.                     logger.error("clustor relselect fail reason is :" + var10.getMessage() + " if can not slove ,you can set cluster.availablecheck=false in url", var10);
  99.                 }
  100.             }
  102.             return invoker;
  103.         }
  104.     } else { 
  105.         return null;
  106.     }
  107. }

AbstractLoadBalance

  1. public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) { 
  2.     if (invokers != null && invokers.size() != 0) { 
  3.         return invokers.size() == 1 ? (Invoker)invokers.get(0) : this.doSelect(invokers, url, invocation);
  4.     } else { 
  5.         return null;
  6.     }
  7. }
  8. // 模板方法
  9. protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> var1, URL var2, Invocation var3);

RandomLoadBalance

  1. protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { 
  2.     int length = invokers.size();
  3.     int totalWeight = 0;
  4.     boolean sameWeight = true;
  6.     int offset;
  7.     int i;
  8.     for(offset = 0; offset < length; ++offset) { 
  9.         i = this.getWeight((Invoker)invokers.get(offset), invocation);
  10.         totalWeight += i;
  11.         if (sameWeight && offset > 0 && i != this.getWeight((Invoker)invokers.get(offset - 1), invocation)) { 
  12.             sameWeight = false;
  13.         }
  14.     }
  16.     if (totalWeight > 0 && !sameWeight) { 
  17.         offset = this.random.nextInt(totalWeight);
  19.         for(i = 0; i < length; ++i) { 
  20.             offset -= this.getWeight((Invoker)invokers.get(i), invocation);
  21.             if (offset < 0) { 
  22.                 return (Invoker)invokers.get(i);
  23.             }
  24.         }
  25.     }
  27.     return (Invoker)invokers.get(this.random.nextInt(length));
  28. }

回到 FailoverClusterInvoker 的 doInvoke 方法，会继续调用 DubboProtocol ，而 invoke 方法在其父类 AbstractInvoker 中

AbstractInvoker

  1. public Result invoke(Invocation inv) throws RpcException { 
  2.     if (this.destroyed.get()) { 
  3.         throw new RpcException("Rpc invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + " is DESTROYED, can not be invoked any more!");
  4.     } else { 
  5.         RpcInvocation invocation = (RpcInvocation)inv;
  6.         invocation.setInvoker(this);
  7.         if (this.attachment != null && this.attachment.size() > 0) { 
  8.             invocation.addAttachmentsIfAbsent(this.attachment);
  9.         }
  11.         Map<String, String> context = RpcContext.getContext().getAttachments();
  12.         if (context != null) { 
  13.             invocation.addAttachmentsIfAbsent(context);
  14.         }
  16.         if (this.getUrl().getMethodParameter(invocation.getMethodName(), "async", false)) { 
  17.             invocation.setAttachment("async", Boolean.TRUE.toString());
  18.         }
  20.         RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
  22.         try { 
  23.             return this.doInvoke(invocation);
  24.         } catch (InvocationTargetException var6) { 
  25.             Throwable te = var6.getTargetException();
  26.             if (te == null) { 
  27.                 return new RpcResult(var6);
  28.             } else { 
  29.                 if (te instanceof RpcException) { 
  30.                     ((RpcException)te).setCode(3);
  31.                 }
  33.                 return new RpcResult(te);
  34.             }
  35.         } catch (RpcException var7) { 
  36.             if (var7.isBiz()) { 
  37.                 return new RpcResult(var7);
  38.             } else { 
  39.                 throw var7;
  40.             }
  41.         } catch (Throwable var8) { 
  42.             return new RpcResult(var8);
  43.         }
  44.     }
  45. }
  47. protected abstract Result doInvoke(Invocation var1) throws Throwable;

DubboInvoker

  1. protected Result doInvoke(Invocation invocation) throws Throwable { 
  2.     RpcInvocation inv = (RpcInvocation)invocation;
  3.     String methodName = RpcUtils.getMethodName(invocation);
  4.     inv.setAttachment("path", this.getUrl().getPath());
  5.     inv.setAttachment("version", this.version);
  6.     ExchangeClient currentClient;
  7.     if (this.clients.length == 1) { 
  8.         currentClient = this.clients[0];
  9.     } else { 
  10.         currentClient = this.clients[this.index.getAndIncrement() % this.clients.length];
  11.     }
  13.     try { 
  14.         boolean isAsync = RpcUtils.isAsync(this.getUrl(), invocation);
  15.         boolean isOneway = RpcUtils.isOneway(this.getUrl(), invocation);
  16.         int timeout = this.getUrl().getMethodParameter(methodName, "timeout", 1000);
  17.         if (isOneway) { 
  18.             boolean isSent = this.getUrl().getMethodParameter(methodName, "sent", false);
  19.             currentClient.send(inv, isSent);
  20.             RpcContext.getContext().setFuture((Future)null);
  21.             return new RpcResult();
  22.         } else if (isAsync) { 
  23.             ResponseFuture future = currentClient.request(inv, timeout);
  24.             RpcContext.getContext().setFuture(new FutureAdapter(future));
  25.             return new RpcResult();
  26.         } else { 
  27.             RpcContext.getContext().setFuture((Future)null);
  28.             // 发送请求>>
  29.             // currentClient -> 当前创建的连接
  30.             return (Result)currentClient.request(inv, timeout).get();
  31.         }
  32.     } catch (TimeoutException var9) { 
  33.         throw new RpcException(2, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + this.getUrl() + ", cause: " + var9.getMessage(), var9);
  34.     } catch (RemotingException var10) { 
  35.         throw new RpcException(1, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + this.getUrl() + ", cause: " + var10.getMessage(), var10);
  36.     }
  37. }

HeaderExchangeClient

  1. public ResponseFuture request(Object request, int timeout) throws RemotingException { 
  2.     return this.channel.request(request, timeout);
  3. }

HeaderExchangeChannel

  1. public ResponseFuture request(Object request, int timeout) throws RemotingException { 
  2.     if (this.closed) { 
  3.         throw new RemotingException(this.getLocalAddress(), (InetSocketAddress)null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
  4.     } else { 
  5.         Request req = new Request();
  6.         req.setVersion("2.0.0");
  7.         req.setTwoWay(true);
  8.         req.setData(request);
  9.         DefaultFuture future = new DefaultFuture(this.channel, req, timeout);
  11.         try { 
  12.             this.channel.send(req);
  13.             return future;
  14.         } catch (RemotingException var6) { 
  15.             future.cancel();
  16.             throw var6;
  17.         }
  18.     }
  19. }
  21. public void send(Object message) throws RemotingException { 
  22.     this.send(message, this.getUrl().getParameter("sent", false));
  23. }
  25. public void send(Object message, boolean sent) throws RemotingException { 
  26.     if (this.closed) { 
  27.         throw new RemotingException(this.getLocalAddress(), (InetSocketAddress)null, "Failed to send message " + message + ", cause: The channel " + this + " is closed!");
  28.     } else { 
  29.         if (!(message instanceof Request) && !(message instanceof Response) && !(message instanceof String)) { 
  30.             Request request = new Request();
  31.             request.setVersion("2.0.0");
  32.             request.setTwoWay(false);
  33.             request.setData(message);
  34.             this.channel.send(request, sent);
  35.         } else { 
  36.             // channel -> NettyChannel
  37.             this.channel.send(message, sent);
  38.         }
  39.     }
  40. }

NettyChannel

  1. public void send(Object message, boolean sent) throws RemotingException { 
  2.     super.send(message, sent);
  3.     boolean success = true;
  4.     int timeout = 0;
  6.     try { 
  7.         ChannelFuture future = this.channel.write(message);
  8.         if (sent) { 
  9.             timeout = this.getUrl().getPositiveParameter("timeout", 1000);
  10.             success = future.await((long)timeout);
  11.         }
  13.         Throwable cause = future.getCause();
  14.         if (cause != null) { 
  15.             throw cause;
  16.         }
  17.     } catch (Throwable var7) { 
  18.         throw new RemotingException(this, "Failed to send message " + message + " to " + this.getRemoteAddress() + ", cause: " + var7.getMessage(), var7);
  19.     }
  21.     if (!success) { 
  22.         throw new RemotingException(this, "Failed to send message " + message + " to " + this.getRemoteAddress() + "in timeout(" + timeout + "ms) limit");
  23.     }
  24. }

服务调用完成。

发表评论

表情：
评论列表 （有 0 条评论，285人围观）

还没有评论，来说两句吧...

相关阅读