Dubbo原理简述四：服务暴露、服务引用和调用流程

妖狐艹你老母 2022-03-26 05:22 349阅读 0赞

## 一、服务暴露 ##

先看一张Dubbo服务暴露的流程图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70]  
关于服务暴露，首先需要了解：`ServiceBean.java`  
此类实现了几个Spring中比较重要的接口：如`InitializingBean、ApplicationContextAware、ApplicationListener`等，查看该类结构图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 1]

接下来分析一下该类的一些主要方法：

在`ServiceBean.afterPropertiesSet()`方法中，就是将配置文件的各项属性信息都配置到该Bean中。最后通过一个`ServiceConfig.export();`方法暴露服务。

@SuppressWarnings({ "unchecked", "deprecation" })
 public void afterPropertiesSet() throws Exception { 
 if (getProvider() == null) { 
 Map<String, ProviderConfig> providerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProviderConfig.class, false, false);
 if (providerConfigMap != null && providerConfigMap.size() > 0) { 
 Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
 if ((protocolConfigMap == null || protocolConfigMap.size() == 0)
 && providerConfigMap.size() > 1) { // 兼容旧版本
 List<ProviderConfig> providerConfigs = new ArrayList<ProviderConfig>();
 for (ProviderConfig config : providerConfigMap.values()) { 
 if (config.isDefault() != null && config.isDefault().booleanValue()) { 
 providerConfigs.add(config);
 }
 }
 if (providerConfigs.size() > 0) { 
 //
 setProviders(providerConfigs);
 }
 } else { 
 ProviderConfig providerConfig = null;
 for (ProviderConfig config : providerConfigMap.values()) { 
 if (config.isDefault() == null || config.isDefault().booleanValue()) { 
 if (providerConfig != null) { 
 throw new IllegalStateException("Duplicate provider configs: " + providerConfig + " and " + config);
 }
 providerConfig = config;
 }
 }
 if (providerConfig != null) { 
 setProvider(providerConfig);
 }
 }
 }
 }
 if (getApplication() == null
 && (getProvider() == null || getProvider().getApplication() == null)) { 
 Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
 if (applicationConfigMap != null && applicationConfigMap.size() > 0) { 
 ApplicationConfig applicationConfig = null;
 for (ApplicationConfig config : applicationConfigMap.values()) { 
 if (config.isDefault() == null || config.isDefault().booleanValue()) { 
 if (applicationConfig != null) { 
 throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
 }
 applicationConfig = config;
 }
 }
 if (applicationConfig != null) { 
 setApplication(applicationConfig);
 }
 }
 }
 if (getModule() == null
 && (getProvider() == null || getProvider().getModule() == null)) { 
 Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
 if (moduleConfigMap != null && moduleConfigMap.size() > 0) { 
 ModuleConfig moduleConfig = null;
 for (ModuleConfig config : moduleConfigMap.values()) { 
 if (config.isDefault() == null || config.isDefault().booleanValue()) { 
 if (moduleConfig != null) { 
 throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
 }
 moduleConfig = config;
 }
 }
 if (moduleConfig != null) { 
 setModule(moduleConfig);
 }
 }
 }
 if ((getRegistries() == null || getRegistries().size() == 0)
 && (getProvider() == null || getProvider().getRegistries() == null || getProvider().getRegistries().size() == 0)
 && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().size() == 0)) { 
 Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
 if (registryConfigMap != null && registryConfigMap.size() > 0) { 
 List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
 for (RegistryConfig config : registryConfigMap.values()) { 
 if (config.isDefault() == null || config.isDefault().booleanValue()) { 
 registryConfigs.add(config);
 }
 }
 if (registryConfigs != null && registryConfigs.size() > 0) { 
 super.setRegistries(registryConfigs);
 }
 }
 }
 if (getMonitor() == null
 && (getProvider() == null || getProvider().getMonitor() == null)
 && (getApplication() == null || getApplication().getMonitor() == null)) { 
 Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
 if (monitorConfigMap != null && monitorConfigMap.size() > 0) { 
 MonitorConfig monitorConfig = null;
 for (MonitorConfig config : monitorConfigMap.values()) { 
 if (config.isDefault() == null || config.isDefault().booleanValue()) { 
 if (monitorConfig != null) { 
 throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
 }
 monitorConfig = config;
 }
 }
 if (monitorConfig != null) { 
 setMonitor(monitorConfig);
 }
 }
 }
 if ((getProtocols() == null || getProtocols().size() == 0)
 && (getProvider() == null || getProvider().getProtocols() == null || getProvider().getProtocols().size() == 0)) { 
 Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
 if (protocolConfigMap != null && protocolConfigMap.size() > 0) { 
 List<ProtocolConfig> protocolConfigs = new ArrayList<ProtocolConfig>();
 for (ProtocolConfig config : protocolConfigMap.values()) { 
 if (config.isDefault() == null || config.isDefault().booleanValue()) { 
 protocolConfigs.add(config);
 }
 }
 if (protocolConfigs != null && protocolConfigs.size() > 0) { 
 super.setProtocols(protocolConfigs);
 }
 }
 }
 if (getPath() == null || getPath().length() == 0) { 
 if (beanName != null && beanName.length() > 0 
 && getInterface() != null && getInterface().length() > 0
 && beanName.startsWith(getInterface())) { 
 setPath(beanName);
 }
 }
 if (! isDelay()) { 
 //此方法最为关键，即服务暴露的动作从此时触发
 export();
 }
 }

再来看看一下`ServiceConfig.export()`这个方法：

public class ServiceConfig<T> extends AbstractServiceConfig { 
 
 //...
 
 public synchronized void export() { 
 if (provider != null) { 
 if (export == null) { 
 export = provider.getExport();
 }
 if (delay == null) { 
 delay = provider.getDelay();
 }
 }
 if (export != null && ! export.booleanValue()) { 
 return;
 }
 if (delay != null && delay > 0) { 
 Thread thread = new Thread(new Runnable() { 
 public void run() { 
 try { 
 Thread.sleep(delay);
 } catch (Throwable e) { 
 }
 //主要核心操作在于doExport()
 doExport();
 }
 });
 thread.setDaemon(true);
 thread.setName("DelayExportServiceThread");
 thread.start();
 } else { 
 doExport();
 }
 }
 }

接下来看看`ServiceConfig.doExport()`方法：

public class ServiceConfig<T> extends AbstractServiceConfig { 
 
 //...
 
 protected synchronized void doExport() { 
 if (unexported) { 
 throw new IllegalStateException("Already unexported!");
 }
 if (exported) { 
 return;
 }
 exported = true;
 if (interfaceName == null || interfaceName.length() == 0) { 
 throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");
 }
 checkDefault();
 if (provider != null) { 
 if (application == null) { 
 application = provider.getApplication();
 }
 if (module == null) { 
 module = provider.getModule();
 }
 if (registries == null) { 
 registries = provider.getRegistries();
 }
 if (monitor == null) { 
 monitor = provider.getMonitor();
 }
 if (protocols == null) { 
 protocols = provider.getProtocols();
 }
 }
 if (module != null) { 
 if (registries == null) { 
 registries = module.getRegistries();
 }
 if (monitor == null) { 
 monitor = module.getMonitor();
 }
 }
 if (application != null) { 
 if (registries == null) { 
 registries = application.getRegistries();
 }
 if (monitor == null) { 
 monitor = application.getMonitor();
 }
 }
 if (ref instanceof GenericService) { 
 interfaceClass = GenericService.class;
 generic = true;
 } else { 
 try { 
 interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
 .getContextClassLoader());
 } catch (ClassNotFoundException e) { 
 throw new IllegalStateException(e.getMessage(), e);
 }
 checkInterfaceAndMethods(interfaceClass, methods);
 checkRef();
 generic = false;
 }
 if(local !=null){ 
 if(local=="true"){ 
 local=interfaceName+"Local";
 }
 Class<?> localClass;
 try { 
 localClass = ClassHelper.forNameWithThreadContextClassLoader(local);
 } catch (ClassNotFoundException e) { 
 throw new IllegalStateException(e.getMessage(), e);
 }
 if(!interfaceClass.isAssignableFrom(localClass)){ 
 throw new IllegalStateException("The local implemention class " + localClass.getName() + " not implement interface " + interfaceName);
 }
 }
 if(stub !=null){ 
 if(stub=="true"){ 
 stub=interfaceName+"Stub";
 }
 Class<?> stubClass;
 try { 
 stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub);
 } catch (ClassNotFoundException e) { 
 throw new IllegalStateException(e.getMessage(), e);
 }
 if(!interfaceClass.isAssignableFrom(stubClass)){ 
 throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + interfaceName);
 }
 }
 checkApplication();
 checkRegistry();
 checkProtocol();
 appendProperties(this);
 checkStubAndMock(interfaceClass);
 if (path == null || path.length() == 0) { 
 path = interfaceName;
 }
 //暴露url操作
 doExportUrls();
 }

再来看看上述代码中暴露url的代码`doExportUrls()`（比较烦躁，坚持啊）。

@SuppressWarnings({ "unchecked", "rawtypes" })
 private void doExportUrls() { 
 List<URL> registryURLs = loadRegistries(true);
 for (ProtocolConfig protocolConfig : protocols) { 
 //获取注册中心的信息，根据协议暴露对应的url
 doExportUrlsFor1Protocol(protocolConfig, registryURLs);
 }
 }

`ServiceConfig.doExportUrlsFor1Protocol(protocolConfig, registryURLs)`源码如下，进行Url暴露的细节：

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) { 
 String name = protocolConfig.getName();
 if (name == null || name.length() == 0) { 
 name = "dubbo";
 }
 
 String host = protocolConfig.getHost();
 if (provider != null && (host == null || host.length() == 0)) { 
 host = provider.getHost();
 }
 boolean anyhost = false;
 if (NetUtils.isInvalidLocalHost(host)) { 
 anyhost = true;
 try { 
 host = InetAddress.getLocalHost().getHostAddress();
 } catch (UnknownHostException e) { 
 logger.warn(e.getMessage(), e);
 }
 if (NetUtils.isInvalidLocalHost(host)) { 
 if (registryURLs != null && registryURLs.size() > 0) { 
 for (URL registryURL : registryURLs) { 
 try { 
 Socket socket = new Socket();
 try { 
 SocketAddress addr = new InetSocketAddress(registryURL.getHost(), registryURL.getPort());
 socket.connect(addr, 1000);
 host = socket.getLocalAddress().getHostAddress();
 break;
 } finally { 
 try { 
 socket.close();
 } catch (Throwable e) { }
 }
 } catch (Exception e) { 
 logger.warn(e.getMessage(), e);
 }
 }
 }
 if (NetUtils.isInvalidLocalHost(host)) { 
 host = NetUtils.getLocalHost();
 }
 }
 }
 
 Integer port = protocolConfig.getPort();
 if (provider != null && (port == null || port == 0)) { 
 port = provider.getPort();
 }
 final int defaultPort = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(name).getDefaultPort();
 if (port == null || port == 0) { 
 port = defaultPort;
 }
 if (port == null || port <= 0) { 
 port = getRandomPort(name);
 if (port == null || port < 0) { 
 port = NetUtils.getAvailablePort(defaultPort);
 putRandomPort(name, port);
 }
 logger.warn("Use random available port(" + port + ") for protocol " + name);
 }
 
 Map<String, String> map = new HashMap<String, String>();
 if (anyhost) { 
 map.put(Constants.ANYHOST_KEY, "true");
 }
 map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE);
 map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
 map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
 if (ConfigUtils.getPid() > 0) { 
 map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
 }
 appendParameters(map, application);
 appendParameters(map, module);
 appendParameters(map, provider, Constants.DEFAULT_KEY);
 appendParameters(map, protocolConfig);
 appendParameters(map, this);
 if (methods != null && methods.size() > 0) { 
 for (MethodConfig method : methods) { 
 appendParameters(map, method, method.getName());
 String retryKey = method.getName() + ".retry";
 if (map.containsKey(retryKey)) { 
 String retryValue = map.remove(retryKey);
 if ("false".equals(retryValue)) { 
 map.put(method.getName() + ".retries", "0");
 }
 }
 List<ArgumentConfig> arguments = method.getArguments();
 if (arguments != null && arguments.size() > 0) { 
 for (ArgumentConfig argument : arguments) { 
 //类型自动转换.
 if(argument.getType() != null && argument.getType().length() >0){ 
 Method[] methods = interfaceClass.getMethods();
 //遍历所有方法
 if(methods != null && methods.length > 0){ 
 for (int i = 0; i < methods.length; i++) { 
 String methodName = methods[i].getName();
 //匹配方法名称，获取方法签名.
 if(methodName.equals(method.getName())){ 
 Class<?>[] argtypes = methods[i].getParameterTypes();
 //一个方法中单个callback
 if (argument.getIndex() != -1 ){ 
 if (argtypes[argument.getIndex()].getName().equals(argument.getType())){ 
 appendParameters(map, argument, method.getName() + "." + argument.getIndex());
 }else { 
 throw new IllegalArgumentException("argument config error : the index attribute and type attirbute not match :index :"+argument.getIndex() + ", type:" + argument.getType());
 }
 } else { 
 //一个方法中多个callback
 for (int j = 0 ;j<argtypes.length ;j++) { 
 Class<?> argclazz = argtypes[j];
 if (argclazz.getName().equals(argument.getType())){ 
 appendParameters(map, argument, method.getName() + "." + j);
 if (argument.getIndex() != -1 && argument.getIndex() != j){ 
 throw new IllegalArgumentException("argument config error : the index attribute and type attirbute not match :index :"+argument.getIndex() + ", type:" + argument.getType());
 }
 }
 }
 }
 }
 }
 }
 }else if(argument.getIndex() != -1){ 
 appendParameters(map, argument, method.getName() + "." + argument.getIndex());
 }else { 
 throw new IllegalArgumentException("argument config must set index or type attribute.eg: <dubbo:argument index='0' .../> or <dubbo:argument type=xxx .../>");
 }
 
 }
 }
 } // end of methods for
 }
 
 if (generic) { 
 map.put("generic", String.valueOf(true));
 map.put("methods", Constants.ANY_VALUE);
 } else { 
 String revision = Version.getVersion(interfaceClass, version);
 if (revision != null && revision.length() > 0) { 
 map.put("revision", revision);
 }
 
 String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
 if(methods.length == 0) { 
 logger.warn("NO method found in service interface " + interfaceClass.getName());
 map.put("methods", Constants.ANY_VALUE);
 }
 else { 
 map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
 }
 }
 if (! ConfigUtils.isEmpty(token)) { 
 if (ConfigUtils.isDefault(token)) { 
 map.put("token", UUID.randomUUID().toString());
 } else { 
 map.put("token", token);
 }
 }
 if ("injvm".equals(protocolConfig.getName())) { 
 protocolConfig.setRegister(false);
 map.put("notify", "false");
 }
 // 导出服务
 String contextPath = protocolConfig.getContextpath();
 if ((contextPath == null || contextPath.length() == 0) && provider != null) { 
 contextPath = provider.getContextpath();
 }
 URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
 
 if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
 .hasExtension(url.getProtocol())) { 
 url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
 .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
 }
 
 String scope = url.getParameter(Constants.SCOPE_KEY);
 //配置为none不暴露
 if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { 
 
 //配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务)
 if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) { 
 exportLocal(url);
 }
 //如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露远程服务)
 if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){ 
 if (logger.isInfoEnabled()) { 
 logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
 }
 if (registryURLs != null && registryURLs.size() > 0
 && url.getParameter("register", true)) { 
 for (URL registryURL : registryURLs) { 
 url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
 URL monitorUrl = loadMonitor(registryURL);
 if (monitorUrl != null) { 
 url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
 }
 if (logger.isInfoEnabled()) { 
 logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
 }
 /** * 暴露远程服务： * 1. 协议在接收请求时，应记录请求来源方地址信息：RpcContext.getContext().setRemoteAddress(); * 2. export()必须是幂等的，也就是暴露同一个URL的Invoker两次，和暴露一次没有区别。 * 3. export()传入的Invoker由框架实现并传入，协议不需要关心。 * * @param <T> 服务的类型 * @param invoker 服务的执行体 * @return exporter 暴露服务的引用，用于取消暴露 * @throws RpcException 当暴露服务出错时抛出，比如端口已占用 */
 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
 
 Exporter<?> exporter = protocol.export(invoker);
 exporters.add(exporter);
 }
 } else { 
 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
 /** * 暴露远程服务： * 1. 协议在接收请求时，应记录请求来源方地址信息：RpcContext.getContext().setRemoteAddress(); * 2. export()必须是幂等的，也就是暴露同一个URL的Invoker两次，和暴露一次没有区别。 * 3. export()传入的Invoker由框架实现并传入，协议不需要关心。 * * @param <T> 服务的类型 * @param invoker 服务的执行体 * @return exporter 暴露服务的引用，用于取消暴露 * @throws RpcException 当暴露服务出错时抛出，比如端口已占用 */
 Exporter<?> exporter = protocol.export(invoker);
 exporters.add(exporter);
 }
 }
 }
 this.urls.add(url);
 }

然后通过`protocol.export(invoker)`方法把服务暴露出去（会根据当前的协议做相应的暴露操作），最后将Url和对应的执行器存起来，存放在注册表中，以供消费者调用。  
![在这里插入图片描述][20190115194402444.png]

这就是一个简洁的Dubbo服务暴露的过程了。

## 二、服务引用 ##

如下为服务引用的流程图，可以参考`服务暴露`的过程进行源码调试。  
其过程简而言之通过一系列的信息，获取到接口对应的代理对象，存放于注册表中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 2]

## 三、调用流程 ##

Dubbo的调用流程，可以参考官网的一张服务调用流程图来对源码进行一步步的调试，便于理解（防止文章过长此处就不贴源码了）：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 3]

更多原理了解，可以到官网查看[Dubbo源码导读][Dubbo]。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70]: /images/20220326/27e3b8a439b6466abac58d286b2a09a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 1]: /images/20220326/6a93557a7ee941daac56329f62fe6802.png
[20190115194402444.png]: /images/20220326/8648b7b6e5344644ab2de296327db18e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 2]: /images/20220326/3d6a4a6a7cba49f2add866485abc47b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 3]: /images/20220326/ab727b1015564e178f5725418c675f12.png
[Dubbo]: http://dubbo.apache.org/zh-cn/docs/source_code_guide/export-service.html