qidao123.com ToB IT社区-企服评测·应用市场»论坛 › 软件与程序人生 › 后端开发 › Java › dubbo服务袒露源码

dubbo服务袒露源码

[复制链接]
发表于 2026-5-18 09:10:52 | 显示全部楼层 |阅读模式

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

×
Dubbo 调用图解

dubbo的调用图（来自官网），如下图，共包罗了5个模块


  • Provider 服务提供方
  • Registry 服务注册中央（这里可以以为是zookeeper
  • Consumer 服务利用方
  • Container 服务提供方的容器
  • Monitor 服务监控监控中央
服务调用流程

  • 提供服务的容器启动之后，把该服务提交给服务提供方
  • 服务提供方把该服务细节以约定的协议（此处可以为是dubbo协议）把IP、端口、服务名等等上报给服务注册中央，由服务注册中央同一管理（别的存在心跳检测，便于及时相识服务康健情况；服务平衡负责等也由其管理）
  • 服务调用方从服务注册中央获取到一个可用的服务提供方的信息
  • 服务注册中央把符合的服务提供方的细节信息下发到调用方
  • 服务调用方持有服务提供方的调用信息，可直连服务提供方举行invoke调用操纵
  • 服务提供方以及服务调用方的调用情况，在必备的情况下都可以定时上报到监控监控中央，从而相识服务调用的数据统计情况
接着把上述过程模块化，细化出来，如下图

引入

从spring分析xml开始，我们可以或许很显着的查察到DubboNamespaceHandler文件
  1. public class DubboNamespaceHandler extends NamespaceHandlerSupport implements ConfigurableSourceBeanMetadataElement {
  2.     public DubboNamespaceHandler() {
  3.     }
  5.     public void init() {
  6.         this.registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class));
  7.         this.registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class));
  8.         this.registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class));
  9.         this.registerBeanDefinitionParser("config-center", new DubboBeanDefinitionParser(ConfigCenterBean.class));
  10.         this.registerBeanDefinitionParser("metadata-report", new DubboBeanDefinitionParser(MetadataReportConfig.class));
  11.         this.registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class));
  12.         this.registerBeanDefinitionParser("metrics", new DubboBeanDefinitionParser(MetricsConfig.class));
  13.         this.registerBeanDefinitionParser("tracing", new DubboBeanDefinitionParser(TracingConfig.class));
  14.         this.registerBeanDefinitionParser("ssl", new DubboBeanDefinitionParser(SslConfig.class));
  15.         this.registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class));
  16.         this.registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class));
  17.         this.registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class));
  18.         this.registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class));
  19.         this.registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class));
  20.         this.registerBeanDefinitionParser("annotation", new AnnotationBeanDefinitionParser());
  21.     }
  23.     public BeanDefinition parse(Element element, ParserContext parserContext) {
  24.         BeanDefinitionRegistry registry = parserContext.getRegistry();
  25.         this.registerAnnotationConfigProcessors(registry);
  26.         DubboSpringInitializer.initialize(parserContext.getRegistry());
  27.         BeanDefinition beanDefinition = super.parse(element, parserContext);
  28.         this.setSource(beanDefinition);
  29.         return beanDefinition;
  30.     }
  32.     private void registerAnnotationConfigProcessors(BeanDefinitionRegistry registry) {
  33.         AnnotationConfigUtils.registerAnnotationConfigProcessors(registry);
  34.     }
  35. }
复制代码
在为外界袒露服务的时间，常利用，再团结DubboBeanDefinitionParser类的parse细节，可知服务袒露，我们应该关注的类是ServiceBean.class
ServiceBean 先容

如下图，是ServiceBean类的继承关系图，右边圈出来的是关于dubbo的，下面底部的是和spring有关的。

很清晰的看到serviceBean也是一个可由Spring管理的很平常的bean

  • 通过BeanNameAware修改bean的名称
  • ApplicationContextAware去获取Spring IOC的容器
  • IntializingBean的afterPropertiesSet去自界说实现bean的实例对象
  • ApplicationListener的onApplicationEvent罗致各种变乱
  • DisposableBean的destroy去烧毁bean
右侧则和dubbo有关，一层一层的config扩展实例化，包罗了在xml设置中的各种参数设置。
属性

属性是整个的服务袒露的这个ServiceBean包罗的各种属性信息，xml设置信息都会归并到这个属性中
ServiceConfig 类
  1. private String              interfaceName;    // 接口类型  
  2. private Class<?>            interfaceClass;
  3. private T                   ref;         // 接口实现类引用
  4. private String              path;     // 服务名称
  5. private List<MethodConfig>  methods;    // 方法配置
  6. private ProviderConfig provider;        // 提供方配置
复制代码

  • 方法设置参数methods,一样平常情况下是没有设置的，也就意味着该接口下的全部的方法都会被袒暴露去，如果设置了就意味着设置的方法才会被袒暴露去。
  • 提供方设置provider也是负责服务袒露方的一些认识信息，比方负载平衡等信息。
AbstractServiceConfig 类
  1. protected String               version;    // 服务版本
  2. protected String               group;      // 服务分组
  3. protected Boolean              deprecated;      // 服务是否已经deprecated
  4. protected Integer              delay;     // 延迟暴露
  5. protected Boolean              export;    // 是否暴露
  6. protected Integer              weight;     // 权重
  7. protected String               document;    // 应用文档
  8. protected Boolean              dynamic;    // 在注册中心上注册成动态的还是静态的服务
  9. protected String               token;     // 是否使用令牌
  10. protected String               accesslog;   // 访问日志日志
  11. private Integer                executes;   // 允许执行请求数
  12. protected List<ProtocolConfig> protocols;   // 暴露的协议
  13. private Boolean                register;   // 是否注册
复制代码
此中protocols就是常说的dubbo协议了，这里指明list也就是意味着支持可以同时多种协议对外袒露
AbstractInterfaceConfig 类
  1. protected String               local;   // 服务接口的本地实现类名
  2. protected String               stub;   // 服务接口的本地实现类名
  3. protected MonitorConfig        monitor;   // 服务监控监控
  4. protected String               proxy;    // 代理类型
  5. protected String               cluster;   // 集群方式
  6. protected String               filter;    // 过滤器
  7. protected String               listener;   // 监听器
  8. protected String               owner;   // 负责人
  9. // 连接数限制,0表示共享连接，否则为该服务独享连接数
  10. protected Integer              connections;
  11. protected String               layer;    // 连接数限制
  12. protected ApplicationConfig    application;    // 应用信息
  13. protected ModuleConfig         module;    // 模块信息
  14. protected List<RegistryConfig> registries;    // 注册中心
  15. private Integer                callbacks;    // callback实例个数限制
  16. protected String              onconnect;   // 连接事件
  17. protected String              ondisconnect;   // 断开事件
  18. // 服务暴露或引用的scope,如果为local，则表示只在当前JVM内查找.
  19. private String scope;
复制代码
注册中央registries应该是比力紧张的属性信息了，包罗了注册中央的数据，好比设置zk的干系属性信息，后期袒露也紧张是把服务按照约定的协议推送给注册中央。
其他继承的类的属性更多的是涉及到体系管理、监控等层级的属性，在此不做过多先容了
设置注入

bean继承了IntializingBean，那肯定就利用了afterPropertiesSet方法
注意在运行到这个时间，servicebean实例化是已经完成了的。
  1. public void afterPropertiesSet() throws Exception {
  2.     if (getProvider() == null) {
  3.         // 提供方为null
  4.         Map<String, ProviderConfig> providerConfigMap = applicationContext == null ? null  : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProviderConfig.class, false, false);
  5.         // 此处的applicationContext就是通过继承ApplicationContextAware注入的Spring IOC容器
  6.         // 获取所有类型是ProviderConfig的bean信息
  7.         if (providerConfigMap != null && providerConfigMap.size() > 0) {
  8.             Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null  : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
  9.             // 获取ProtocolConfig
  10.             if ((protocolConfigMap == null || protocolConfigMap.size() == 0)
  11.                     && providerConfigMap.size() > 1) { // 兼容旧版本
  12.                     // 如果没有protocolconfig同时有providerconfig
  13.                 List<ProviderConfig> providerConfigs = new ArrayList<ProviderConfig>();
  14.                 for (ProviderConfig config : providerConfigMap.values()) {
  15.                     if (config.isDefault() != null && config.isDefault().booleanValue()) {
  16.                         // config不为空，而且是默认值
  17.                         providerConfigs.add(config);
  18.                     }
  19.                 }
  20.                 if (providerConfigs.size() > 0) {
  21.                     setProviders(providerConfigs);
  22.                 }
  23.             } else {
  24.                 ProviderConfig providerConfig = null;
  25.                 for (ProviderConfig config : providerConfigMap.values()) {
  26.                     if (config.isDefault() == null || config.isDefault().booleanValue()) {
  27.                         if (providerConfig != null) {
  28.                             throw new IllegalStateException("Duplicate provider configs: " + providerConfig + " and " + config);
  29.                         }
  30.                         providerConfig = config;
  31.                     }
  32.                 }
  33.                 if (providerConfig != null) {
  34.                     // 默认的只应该存在一个providerconfig信息
  35.                     setProvider(providerConfig);
  36.                 }
  37.             }
  38.         }
  39.     }
  40.     if (getApplication() == null
  41.             && (getProvider() == null || getProvider().getApplication() == null)) {
  42.         Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
  43.         if (applicationConfigMap != null && applicationConfigMap.size() > 0) {
  44.             ApplicationConfig applicationConfig = null;
  45.             for (ApplicationConfig config : applicationConfigMap.values()) {
  46.                 if (config.isDefault() == null || config.isDefault().booleanValue()) {
  47.                     if (applicationConfig != null) {
  48.                         throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
  49.                     }
  50.                     applicationConfig = config;
  51.                 }
  52.             }
  53.             if (applicationConfig != null) {
  54.                 // 填充ApplicationConfig信息
  55.                 setApplication(applicationConfig);
  56.             }
  57.         }
  58.     }
  59.     if (getModule() == null
  60.             && (getProvider() == null || getProvider().getModule() == null)) {
  61.         Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
  62.         if (moduleConfigMap != null && moduleConfigMap.size() > 0) {
  63.             ModuleConfig moduleConfig = null;
  64.             for (ModuleConfig config : moduleConfigMap.values()) {
  65.                 if (config.isDefault() == null || config.isDefault().booleanValue()) {
  66.                     if (moduleConfig != null) {
  67.                         throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
  68.                     }
  69.                     moduleConfig = config;
  70.                 }
  71.             }
  72.             if (moduleConfig != null) {
  73.                 // 填充ModuleConfig信息
  74.                 setModule(moduleConfig);
  75.             }
  76.         }
  77.     }
  78.     if ((getRegistries() == null || getRegistries().size() == 0)
  79.             && (getProvider() == null || getProvider().getRegistries() == null || getProvider().getRegistries().size() == 0)
  80.             && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().size() == 0)) {
  81.         Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
  82.         if (registryConfigMap != null && registryConfigMap.size() > 0) {
  83.             List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
  84.             for (RegistryConfig config : registryConfigMap.values()) {
  85.                 if (config.isDefault() == null || config.isDefault().booleanValue()) {
  86.                     registryConfigs.add(config);
  87.                 }
  88.             }
  89.             if (registryConfigs != null && registryConfigs.size() > 0) {
  90.                 // 填充注册信息
  91.                 super.setRegistries(registryConfigs);
  92.             }
  93.         }
  94.     }
  95.     if (getMonitor() == null
  96.             && (getProvider() == null || getProvider().getMonitor() == null)
  97.             && (getApplication() == null || getApplication().getMonitor() == null)) {
  98.         Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
  99.         if (monitorConfigMap != null && monitorConfigMap.size() > 0) {
  100.             MonitorConfig monitorConfig = null;
  101.             for (MonitorConfig config : monitorConfigMap.values()) {
  102.                 if (config.isDefault() == null || config.isDefault().booleanValue()) {
  103.                     if (monitorConfig != null) {
  104.                         throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
  105.                     }
  106.                     monitorConfig = config;
  107.                 }
  108.             }
  109.             if (monitorConfig != null) {
  110.                 // 填充监控信息
  111.                 setMonitor(monitorConfig);
  112.             }
  113.         }
  114.     }
  115.     if ((getProtocols() == null || getProtocols().size() == 0)
  116.             && (getProvider() == null || getProvider().getProtocols() == null || getProvider().getProtocols().size() == 0)) {
  117.         Map<String, ProtocolConfig> protocolConfigMap = applicationContext == null ? null  : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ProtocolConfig.class, false, false);
  118.         if (protocolConfigMap != null && protocolConfigMap.size() > 0) {
  119.             List<ProtocolConfig> protocolConfigs = new ArrayList<ProtocolConfig>();
  120.             for (ProtocolConfig config : protocolConfigMap.values()) {
  121.                 if (config.isDefault() == null || config.isDefault().booleanValue()) {
  122.                     protocolConfigs.add(config);
  123.                 }
  124.             }
  125.             if (protocolConfigs != null && protocolConfigs.size() > 0) {
  126.                 // 设置使用的协议
  127.                 super.setProtocols(protocolConfigs);
  128.             }
  129.         }
  130.     }
  131.     if (getPath() == null || getPath().length() == 0) {
  132.         if (beanName != null && beanName.length() > 0
  133.                 && getInterface() != null && getInterface().length() > 0
  134.                 && beanName.startsWith(getInterface())) {
  135.             setPath(beanName);
  136.         }
  137.     }
  138.     if (! isDelay()) {
  139.         // 如果没有设置延迟，则立即暴露出去
  140.         export();
  141.     }
  142. }
复制代码
在这一段代码中就是从Spring IOC容器中获取符合的bean注入到ServiceBean中，比方利用的服务信息、服务注册中央、利用的协议、平衡负责的方式（provider的loanBanance）、监控等。
袒露服务启动

服务袒露着实就是export函数，如果设置了耽误，则会在ApplicationListener的变乱中去袒露服务。
  1. public void onApplicationEvent(ApplicationEvent event) {
  2.     if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
  3.      // 当前事件是bean刷新结束
  4.         if (isDelay() && ! isExported() && ! isUnexported()) {
  5.               // 是延期而且还没有暴富同时 没有 不希望 暴露服务
  6.             if (logger.isInfoEnabled()) {
  7.                 logger.info("The service ready on spring started. service: " + getInterface());
  8.             }
  9.             // 跳转到ServiceConfig类中
  10.             export();
  11.         }
  12.     }
  13. }
复制代码
ServiceConfig 类
  1. public synchronized void export() {
  2.     // 这里面的provider等信息都是之前注入的信息属性
  3.     if (provider != null) {
  4.         if (export == null) {
  5.             export = provider.getExport();
  6.         }
  7.         if (delay == null) {
  8.             delay = provider.getDelay();
  9.         }
  10.     }
  11.     if (export != null && ! export.booleanValue()) {
  12.         return;
  13.     }
  14.     if (delay != null && delay > 0) {
  15.          // 如果设置了延迟，则设置成为守护线程，睡眠延迟的时间数，再执行暴露服务的任务
  16.         Thread thread = new Thread(new Runnable() {
  17.             public void run() {
  18.                 try {
  19.                     Thread.sleep(delay);
  20.                 } catch (Throwable e) {
  21.                 }
  22.                 doExport();
  23.             }
  24.         });
  25.         thread.setDaemon(true);
  26.         thread.setName("DelayExportServiceThread");
  27.         thread.start();
  28.     } else {
  29.         doExport();
  30.     }
  31. }
复制代码
在doExport方法中更多的是对一些数据的check操纵，随厥后到了doExportUrls方法
  1. private void doExportUrls() {
  2.     List<URL> registryURLs = loadRegistries(true);
  3.     // 先获取注册中心的属性信息
  4.     for (ProtocolConfig protocolConfig : protocols) {
  5.         // 存在多个暴露协议，例如DUBBO、HTTP等，依次对外暴露
  6.         doExportUrlsFor1Protocol(protocolConfig, registryURLs);
  7.     }
  8. }
复制代码
服务注册中央属性获取

AbstractInterfaceConfig 类
  1. protected List<URL> loadRegistries(boolean provider) {
  2.     checkRegistry();
  3.     // 兼容老版本的dubbo服务
  4.     List<URL> registryList = new ArrayList<URL>();
  5.     if (registries != null && registries.size() > 0) {
  6.         // 循环遍历所有的注册中心配置
  7.         for (RegistryConfig config : registries) {
  8.             String address = config.getAddress();
  9.             // 注册中心的地址
  10.             if (address == null || address.length() == 0) {
  11.                 address = Constants.ANYHOST_VALUE;
  12.                 // ANYHOST_VALUE = "0.0.0.0"
  13.             }
  14.             String sysaddress = System.getProperty("dubbo.registry.address");
  15.             // 从系统属性中获取dubbo.registry.address的值
  16.             if (sysaddress != null && sysaddress.length() > 0) {
  17.                 address = sysaddress;
  18.                 // 真存在这个数据，就替换掉之前的地址数据
  19.             }
  20.             if (address != null && address.length() > 0
  21.                     && ! RegistryConfig.NO_AVAILABLE.equalsIgnoreCase(address)) {
  22.                 // RegistryConfig.NO_AVAILABLE = 'N\A',如果地址有效   
  23.                 Map<String, String> map = new HashMap<String, String>();
  24.                 appendParameters(map, application);
  25.                 appendParameters(map, config);
  26.                 map.put("path", RegistryService.class.getName());
  27.                 map.put("dubbo", Version.getVersion());
  28.                 map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
  29.                 // 把应用程序、当前的注册配置信息、以及dubbo、时间戳等信息注入到map中
  30.                 if (ConfigUtils.getPid() > 0) {
  31.                     // 获取当前服务的进程PID
  32.                     map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
  33.                 }
  34.                 if (! map.containsKey("protocol")) {
  35.                     if (ExtensionLoader.getExtensionLoader(RegistryFactory.class).hasExtension("remote")) {
  36.                         // 如果在SPI中包含了支持remote的类，则设置当前的协议为remote，否则是dubbo
  37.                         map.put("protocol", "remote");
  38.                     } else {
  39.                         map.put("protocol", "dubbo");
  40.                     }
  41.                 }
  42.                 List<URL> urls = UrlUtils.parseURLs(address, map);
  43.                 // 这里为啥会生成一个list数据的URL信息呢？
  44.                 // 主要的情况是address可能包含了多个，现实中zk也很多是zk集群了，使用逗号区分开来即可
  45.                 // 其实是通过Pattern.compile("\\s*[|;]+\\s*")的正则方式区分开来的
  46.                 for (URL url : urls) {
  47.                     url = url.addParameter(Constants.REGISTRY_KEY, url.getProtocol());
  48.                     // 添加注册协议，此处为dubbo
  49.                     url = url.setProtocol(Constants.REGISTRY_PROTOCOL);
  50.                     if ((provider && url.getParameter(Constants.REGISTER_KEY, true))
  51.                             || (! provider && url.getParameter(Constants.SUBSCRIBE_KEY, true))) {
  52.                             // 一定是注册方的信息，则添加到这个集合中
  53.                         registryList.add(url);
  54.                     }
  55.                 }
  56.             }
  57.         }
  58.     }
  59.     return registryList;
  60. }
复制代码
末了天生的 registryList 数据大概为 registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?application=dubbo-demo&client=zkclient&dubbo=2.5.3&group=dubbo-demo&owner=jwfy&pid=2772&registry=zookeeper&timestamp=1525276569763
仅仅是包罗了一些注册的根本数据
服务袒露

如今来到了doExportUrlsFor1Protocol方法，当前protocolConfig为
  1. private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
  2.     String name = protocolConfig.getName();
  3.     // 获取对外暴露的协议，如果无效，则默认为dubbo
  4.     if (name == null || name.length() == 0) {
  5.         name = "dubbo";
  6.     }
  8.     String host = protocolConfig.getHost();
  9.     if (provider != null && (host == null || host.length() == 0)) {
  10.         // 如果提供的provider不为空而且暴露的协议配置host无效，设置为提供方的host数据
  11.         host = provider.getHost();
  12.     }
  13.     boolean anyhost = false;
  14.     if (NetUtils.isInvalidLocalHost(host)) {
  15.        // 如果host无效，例如为null，空字符串，localhost，0.0.0.0，127开头的IP
  16.         anyhost = true;
  17.         try {
  18.             host = InetAddress.getLocalHost().getHostAddress();
  19.             // 获取本机IP
  20.         } catch (UnknownHostException e) {
  21.             logger.warn(e.getMessage(), e);
  22.         }
  23.         if (NetUtils.isInvalidLocalHost(host)) {
  24.             // 还是无效
  25.             if (registryURLs != null && registryURLs.size() > 0) {
  26.                 for (URL registryURL : registryURLs) {
  27.                     try {
  28.                         Socket socket = new Socket();
  29.                         try {
  30.                             SocketAddress addr = new InetSocketAddress(registryURL.getHost(), registryURL.getPort());
  31.                             socket.connect(addr, 1000);
  32.                             host = socket.getLocalAddress().getHostAddress();
  33.                             // 通过socket套接字的方式获取host信息
  34.                             break;
  35.                         } finally {
  36.                             try {
  37.                                 socket.close();
  38.                             } catch (Throwable e) {}
  39.                         }
  40.                     } catch (Exception e) {
  41.                         logger.warn(e.getMessage(), e);
  42.                     }
  43.                 }
  44.             }
  45.             if (NetUtils.isInvalidLocalHost(host)) {
  46.                 host = NetUtils.getLocalHost();
  47.             }
  48.         }
  49.     }
  51.     Integer port = protocolConfig.getPort();
  52.     // 获取端口
  53.     if (provider != null && (port == null || port == 0)) {
  54.         port = provider.getPort();
  55.     }
  56.     final int defaultPort = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(name).getDefaultPort();
  57.     // 通过SPI获取名称为name的实体类的默认端口信息，dubbo默认为20880
  58.     if (port == null || port == 0) {
  59.         port = defaultPort;
  60.     }
  61.     if (port == null || port <= 0) {
  62.         port = getRandomPort(name);
  63.         if (port == null || port < 0) {
  64.             port = NetUtils.getAvailablePort(defaultPort);
  65.             putRandomPort(name, port);
  66.         }
  67.         logger.warn("Use random available port(" + port + ") for protocol " + name);
  68.     }
  70.     Map<String, String> map = new HashMap<String, String>();
  71.     if (anyhost) {
  72.        // 如果起初的host为无效信息，设置anyhost为true
  73.         map.put(Constants.ANYHOST_KEY, "true");
  74.     }
  75.     map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE);
  76.     map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
  77.     map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
  78.     if (ConfigUtils.getPid() > 0) {
  79.         map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
  80.     }
  81.     appendParameters(map, application);
  82.     appendParameters(map, module);
  83.     appendParameters(map, provider, Constants.DEFAULT_KEY);
  84.     appendParameters(map, protocolConfig);
  85.     appendParameters(map, this);
  86.     if (methods != null && methods.size() > 0) {
  87.         // 如果当前暴露的接口设置了暴露的方法列表
  88.         for (MethodConfig method : methods) {
  89.             appendParameters(map, method, method.getName());
  90.             String retryKey = method.getName() + ".retry";
  91.             if (map.containsKey(retryKey)) {
  92.                 String retryValue = map.remove(retryKey);
  93.                 if ("false".equals(retryValue)) {
  94.                     map.put(method.getName() + ".retries", "0");
  95.                 }
  96.             }
  97.             List arguments = method.getArguments();
  98.             if (arguments != null && arguments.size() > 0) {
  99.                 for (ArgumentConfig argument : arguments) {
  100.                     //类型自动转换.
  101.                     if(argument.getType() != null && argument.getType().length() >0){
  102.                         Method[] methods = interfaceClass.getMethods();
  103.                         //遍历所有方法
  104.                         if(methods != null && methods.length > 0){
  105.                             for (int i = 0; i < methods.length; i++) {
  106.                                 String methodName = methods[i].getName();
  107.                                 //匹配方法名称，获取方法签名.
  108.                                 if(methodName.equals(method.getName())){
  109.                                     Class<?>[] argtypes = methods[i].getParameterTypes();
  110.                                     //一个方法中单个callback
  111.                                     if (argument.getIndex() != -1 ){
  112.                                         if (argtypes[argument.getIndex()].getName().equals(argument.getType())){
  113.                                             appendParameters(map, argument, method.getName() + "." + argument.getIndex());
  114.                                         }else {
  115.                                             throw new IllegalArgumentException("argument config error : the index attribute and type attirbute not match :index :"+argument.getIndex() + ", type:" + argument.getType());
  116.                                         }
  117.                                     } else {
  118.                                         //一个方法中多个callback
  119.                                         for (int j = 0 ;j argclazz = argtypes[j];
  120.                                             if (argclazz.getName().equals(argument.getType())){
  121.                                                 appendParameters(map, argument, method.getName() + "." + j);
  122.                                                 if (argument.getIndex() != -1 && argument.getIndex() != j){
  123.                                                     throw new IllegalArgumentException("argument config error : the index attribute and type attirbute not match :index :"+argument.getIndex() + ", type:" + argument.getType());
  124.                                                 }
  125.                                             }
  126.                                         }
  127.                                     }
  128.                                 }
  129.                             }
  130.                         }
  131.                     }else if(argument.getIndex() != -1){
  132.                         appendParameters(map, argument, method.getName() + "." + argument.getIndex());
  133.                     }else {
  134.                         throw new IllegalArgumentException("argument config must set index or type attribute.eg: <dubbo:argument index='0' .../> or <dubbo:argument type=xxx .../>");
  135.                     }
  137.                 }
  138.             }
  139.         } // end of methods for
  140.     }
  141.     // 到这里参数已经处理完成了
  143.     if (generic) {
  144.         // 暴露的服务是否为GenericService类的子类
  145.         map.put("generic", String.valueOf(true));
  146.         map.put("methods", Constants.ANY_VALUE);
  147.     } else {
  148.         String revision = Version.getVersion(interfaceClass, version);
  149.         // 设置接口的版本号（便于新旧版本的更迭）
  150.         if (revision != null && revision.length() > 0) {
  151.             map.put("revision", revision);
  152.         }
  154.         String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
  155.         // 获取该接口的所有的可用的方法名称集合
  156.         if(methods.length == 0) {
  157.             logger.warn("NO method found in service interface " + interfaceClass.getName());
  158.             map.put("methods", Constants.ANY_VALUE);
  159.         }
  160.         else {
  161.             map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
  162.             // 拼接的数据为 "getStr,print" 这样类似的数据
  163.         }
  164.     }
  165.     if (! ConfigUtils.isEmpty(token)) {
  166.         // 如果设置了校验令牌token
  167.         if (ConfigUtils.isDefault(token)) {
  168.            // 如果token的值为true或者default，随机设置一个UUID
  169.             map.put("token", UUID.randomUUID().toString());
  170.         } else {
  171.             map.put("token", token);
  172.         }
  173.     }
  174.     if ("injvm".equals(protocolConfig.getName())) {
  175.         protocolConfig.setRegister(false);
  176.         map.put("notify", "false");
  177.     }
  178.     // 导出服务
  179.     String contextPath = protocolConfig.getContextpath();
  180.     if ((contextPath == null || contextPath.length() == 0) && provider != null) {
  181.         contextPath = provider.getContextpath();
  182.     }
  183.     URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
  184.    
  185.     /*
  186.     到这里生成的URL属于具体暴富服务的信息数据了
  187.     例如dubbo://172.16.109.110:20880/com.jwfy.dubbo.product.ProductService?anyhost=true&application=dubbo-demo&default.loadbalance=random&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=2772&side=provider&timestamp=1525313423899
  188.     URL的协议是dubbo，也是暴露出去的协议
  189.     */
  191.     if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
  192.             .hasExtension(url.getProtocol())) {
  193.             // 如果配置工厂类有相应协议，则重新设置其值
  194.         url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
  195.                 .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
  196.     }
  198.     String scope = url.getParameter(Constants.SCOPE_KEY);
  199.     //配置为none不暴露，此处为null,也就是既可以暴露为本地服务也可以暴露为远程服务
  200.     if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
  202.         //配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务)
  203.         if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
  204.             // 真正的暴露服务，后续详解
  205.             exportLocal(url);
  206.         }
  207.         //如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露远程服务)
  208.         if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
  209.             if (logger.isInfoEnabled()) {
  210.                 logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
  211.             }
  212.             if (registryURLs != null && registryURLs.size() > 0
  213.                     && url.getParameter("register", true)) {
  214.                 // 对外暴露服务，必须要注册中心存在
  215.                 for (URL registryURL : registryURLs) {
  216.                     url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
  217.                     URL monitorUrl = loadMonitor(registryURL);
  218.                     // 加载监控中心，后续详解
  219.                     if (monitorUrl != null) {
  220.                         url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
  221.                     }
  222.                     if (logger.isInfoEnabled()) {
  223.                         logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
  224.                     }
  225.                     Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
  226.                     // 远程服务暴露
  227.                     Exporter<?> exporter = protocol.export(invoker);
  228.                     exporters.add(exporter);
  229.                 }
  230.             } else {
  231.                 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
  233.                 Exporter<?> exporter = protocol.export(invoker);
  234.                 exporters.add(exporter);
  235.             }
  236.         }
  237.     }
  238.     this.urls.add(url);
  239. }
复制代码
本地服务袒露

本地服务袒露的入口是exportLocal(url);
  1. private void exportLocal(URL url) {
  2.     if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
  3.         // 这时候的url.getProtocol 为dubbo，而本地服务是injvm
  4.         // 而且如果服务已经是injvm 则不需要走这个操作暴露了
  5.         URL local = URL.valueOf(url.toFullString())
  6.                 .setProtocol(Constants.LOCAL_PROTOCOL)
  7.                 .setHost(NetUtils.LOCALHOST)
  8.                 .setPort(0);
  9.         // 替换为injvm协议和端口号，这个local数据就变成了
  10.         /*
  11.         injvm://127.0.0.1/com.jwfy.dubbo.product.ProductService?anyhost=true&application=dubbo-demo&default.loadbalance=random&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=2772&side=provider&timestamp=1525313423899
  12.         */
  13.         Exporter<?> exporter = protocol.export(
  14.                 proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
  15.         // 通过getInvoke获取invoke对象，通过export进行暴露操作                       
  16.         exporters.add(exporter);
  17.         logger.info("Export dubbo service " + interfaceClass.getName() +" to local registry");
  18.     }
  19. }
复制代码
如今应该是来到了真正服务袒露的入口了，Exporter exporter = protocol.export(proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
protocol & proxyFactory

根据之前对dubbo spi的学习和相识，到这里已经知道了protocol和proxyFactory指的具体是什么了。
protocol 类
  1. package com.alibaba.dubbo.rpc;
  2. import com.alibaba.dubbo.common.extension.ExtensionLoader;
  4. public class Protocol$Adpative implements com.alibaba.dubbo.rpc.Protocol {
  5.     public com.alibaba.dubbo.rpc.Invoker refer(java.lang.Class arg0, com.alibaba.dubbo.common.URL arg1) throws java.lang.Class {
  6.         if (arg1 == null)
  7.             throw new IllegalArgumentException("url == null");
  9.         com.alibaba.dubbo.common.URL url = arg1;
  10.         String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
  11.         if(extName == null)
  12.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
  14.         com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
  16.         return extension.refer(arg0, arg1);
  17.     }
  19.     public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.Invoker {
  20.         if (arg0 == null)
  21.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
  23.         if (arg0.getUrl() == null)
  24.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
  25.         //根据URL配置信息获取Protocol协议，默认是dubbo
  26.         String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
  27.         if(extName == null)
  28.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
  29.             //根据协议名，获取Protocol实体类
  30.         com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
  32.         return extension.export(arg0);
  33.     }
  35.     public void destroy() {
  36.         throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
  37.     }
  39.     public int getDefaultPort() {
  40.         throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
  41.     }
  42. }
复制代码
proxyFactory 类
  1. package com.alibaba.dubbo.rpc;
  2. import com.alibaba.dubbo.common.extension.ExtensionLoader;
  3. public class ProxyFactory$Adpative implements com.alibaba.dubbo.rpc.ProxyFactory {
  4.     public com.alibaba.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, com.alibaba.dubbo.common.URL arg2) throws java.lang.Object {
  5.         if (arg2 == null)
  6.             throw new IllegalArgumentException("url == null");
  8.         com.alibaba.dubbo.common.URL url = arg2;
  9.         String extName = url.getParameter("proxy", "javassist");
  10.         if(extName == null)
  11.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
  13.         com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
  14.         // 获取ProxyFactory类的名称为extName的实现类
  16.         return extension.getInvoker(arg0, arg1, arg2);
  17.         // arg0 是具体被调用的实现类
  18.         // arg1是实现类的接口
  19.         // arg2 是对外暴露的URL信息
  20.         // 只是调用链路已经转交给具体的extension了
  21.     }
  23.     public java.lang.Object getProxy(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.Invoker {
  24.         if (arg0 == null)
  25.             throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
  27.        if (arg0.getUrl() == null)
  28.         throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
  30.         String extName = url.getParameter("proxy", "javassist");
  31.         // 获取URL中的proxy参数信息，没有则设置为javassist
  32.         if(extName == null)
  33.             throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
  35.         com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
  37.         return extension.getProxy(arg0);
  38.     }
  39. }
复制代码
可以很显着的看出来默认的protocol是利用的dubbo协议，对应的实例是包装DubboProtocol后的实例，ProxyFactory利用的是javassist，对应的实例是JavassistProxyFactory
袒露操纵包罗了两个部分，一个Invoke,另一个是export
获取Invoke

JavassistProxyFactory获取Invoke

proxyFactory.getInvoker(ref, (Class) interfaceClass, local) 方法得到具体的Invoke。
此时的ref是袒露的具体实现类，interfaceClass是对应的接口信息，local就是URL信息，具体内容是
registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?application=dubbo-demo&client=zkclient&dubbo=2.5.3&export=dubbo%3A%2F%2F172.16.109.110%3A20880%2Fcom.jwfy.dubbo.product.ProductService%3Fanyhost%3Dtrue%26application%3Ddubbo-demo%26default.loadbalance%3Drandom%26dubbo%3D2.5.3%26interface%3Dcom.jwfy.dubbo.product.ProductService%26methods%3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D13859%26side%3Dprovider%26timestamp%3D1525772505371%26token%3Dfdfdf&group=dubbo-demo&owner=jwfy&pid=13859&registry=zookeeper&timestamp=1525772500246
须要对外袒露的服务就是包罗在URL信息中的ProductService信息
在本demo中，getInvoke操纵获取到JavassistProxyFactory对象后实验他的getInvoke操纵
JavassistProxyFactory 类
  1. public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
  2.     // TODO Wrapper类不能正确处理带$的类名
  3.     final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
  4.     // 这里生产的wrapper也是动态生成的
  5.     return new AbstractProxyInvoker<T>(proxy, type, url) {
  6.         @Override
  7.         protected Object doInvoke(T proxy, String methodName,
  8.                                   Class<?>[] parameterTypes,
  9.                                   Object[] arguments) throws Throwable {
  10.                                   // 这个proxy也就是具体的实现类，对应上面的ref
  11.                                   // methodName是方法名
  12.                                   // 剩下的是参数类型以及名称
  13.             return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
  14.         }
  15.     };
  16. }
复制代码
生产的invoke对象着实是个AbstractProxyInvoker，只不外在调用他的doInvoke操纵时，末了会实验拼接天生的wrapper对象的invokeMethod方法上。
getWrapper
在获取wrapper操纵，也同样是动态拼接字符串天生的，重点看此中的invokeMethod方法
  1. public Object invokeMethod(Object o, String n, Class[] p, Object[] v)
  2.                      throws java.lang.reflect.InvocationTargetException{
  3. com.jwfy.dubbo.product.ProductServiceImpl w;
  4. // 看这个就已经非常明显的说明实现类的具体对象
  5. try{
  6.    w = ((com.jwfy.dubbo.product.ProductServiceImpl)$1);
  7.    // 格式化，强转类型
  8. } catch (Throwable e) {
  9.   throw new IllegalArgumentException(e);
  10. }
  12. try{
  13.   // 对函数名称和参数进行匹配校验操作
  14.   if( "print".equals( $2 )  &&  $3.length == 0 ) {  
  15.      w.print();
  16.      // 函数本身是返回void类型，则直接返回
  17.      return null;
  18.    }
  19.    if( "getStr".equals( $2 )  &&  $3.length == 0 ) {  
  20.       // 调用执行结果后强转格式返回
  21.       return ($w)w.getStr();
  22.     }
  23. } catch(Throwable e) {      
  24.     throw new java.lang.reflect.InvocationTargetException(e);  
  25. }
  27. // 每个接口的方法都会遍历一遍，如果啥都没匹配到，就提示没有方法异常
  28. throw new com.alibaba.dubbo.common.bytecode.NoSuchMethodException("Not found method ""+$2+"" in class com.jwfy.dubbo.product.ProductServiceImpl.");
  29. }
复制代码
如今应该就很清晰了，在实验invokeMethod的时间，背后着实就是调用了实现类的对应方法，只是这个wrapper自己是动态天生的
JdkProxyFactory获取Invoke

上面说了在动态天生的署理工厂中默认实现的是JavassistProxyFactory，但是也可以利用java自己的协议，也就是JdkProxyFactory
  1. public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
  2.     return new AbstractProxyInvoker<T>(proxy, type, url) {
  3.         @Override
  4.         protected Object doInvoke(T proxy, String methodName,
  5.                                   Class<?>[] parameterTypes,
  6.                                   Object[] arguments) throws Throwable {
  7.             Method method = proxy.getClass().getMethod(methodName, parameterTypes);
  8.             return method.invoke(proxy, arguments);
  9.         }
  10.     };
  11. }
复制代码
完全就是通过java的反射去调用实验
Invoke是什么

着实刚刚开始看源码的时间并不黑白常的明确Invoke到底是个什么，如今可以说Invoke是实现类的包装类，并包罗了URL等信息，后续可以通过invoke方法去调用具体服务方。

Invoke袒露为export

获取真实的Protocol类

protocol.export(invoke),此中的invoke就是上面天生的抽象invoke类,但是在单步调试的时间却发现并没有直接进入到我们假想的RegistryProtocol类中
这个须要追踪到Dubbo SPI中的cachedWrapperClasses数据处置处罚中

上述代码已经很清晰了，获取wrapper，起首不应该被Adaptive注解（未贴出），其次肯定得存在包罗了参数为type的构造函数，而如下文件则是protocol的spi文件，可以知道只有filter和listener符合操纵


如许就非常清晰了，在获取getExtension中，应该是获取到了RegistryProtocol对象，但是后续的cachedWrapperClasses操纵又加上了包装操纵，先后参加了ProtocolFilterWrapper、ProtocolListenerWrapper对象，使得在后续protocol.export操纵不是进入到RegistryProtocol中，而是起首辈入到ProtocolFilterWrapper
ProtocolFilterWrapper 类
  1. public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
  2.     if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
  3.         // 如果invoke的协议是registry类型
  4.        return protocol.export(invoker);
  5.     }
  6.     return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
  7. }
复制代码
然厥后到了ProtocolListenerWrapper类
  1. public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
  2.     if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
  3.         // 如果invoke的协议是registry类型
  4.         return protocol.export(invoker);
  5.     }
  6.     return new ListenerExporterWrapper<T>(protocol.export(invoker),
  7.             Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
  8.                     .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));
  9. }
复制代码
注册协议袒露

不颠末任那边置处罚，通过两个wrapper的转发，直接来到RegistryProtocol的export操纵
  1. public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
  2.     final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
  3.     // 本地暴露服务
  4.     final Registry registry = getRegistry(originInvoker);
  5.     // 获取远程注册中心
  6.     final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
  7.     // 注册到注册中心的URL信息，其中包含了一个接口的信息以及协议等信息
  8.     registry.register(registedProviderUrl);
  9.     // 注册操作，如果查看zk记录，可以发现注册成功的操作日志日志
  10.    
  11.     // 订阅override数据
  12.     // FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。
  13.     final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
  14.     final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
  15.     overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
  16.     registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
  17.     //保证每次export都返回一个新的exporter实例
  18.     return new Exporter<T>() {
  19.         public Invoker<T> getInvoker() {
  20.             return exporter.getInvoker();
  21.         }
  22.         public void unexport() {
  23.             try {
  24.                 exporter.unexport();
  25.             } catch (Throwable t) {
  26.                 logger.warn(t.getMessage(), t);
  27.             }
  28.             try {
  29.                 registry.unregister(registedProviderUrl);
  30.             } catch (Throwable t) {
  31.                 logger.warn(t.getMessage(), t);
  32.             }
  33.             try {
  34.                 overrideListeners.remove(overrideSubscribeUrl);
  35.                 registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
  36.             } catch (Throwable t) {
  37.                 logger.warn(t.getMessage(), t);
  38.             }
  39.         }
  40.     };
  41. }
复制代码
获取长途控制中央地点 getRegistry
  1. private Registry getRegistry(final Invoker<?> originInvoker){
  2.     URL registryUrl = originInvoker.getUrl();
  3.     // 获取的invoke的URL信息
  4.     // registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?
  5.     //application=dubbo-demo&client=zkclient&dubbo=2.5.3&export=dubbo
  6.     //%3A%2F%2F192.168.10.123%3A20880%2Fcom.jwfy.dubbo.product.ProductService
  7.     //%3Fanyhost%3Dtrue%26application%3Ddubbo-demo%26default.loadbalance%3Drandom
  8.     //%26dubbo%3D2.5.3%26interface%3Dcom.jwfy.dubbo.product.ProductService%26methods
  9.     // %3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D12663%26side%3Dprovider
  10.     // %26timestamp%3D1525733684167%26token%3Dfdfdf&
  11.     // group=dubbo-demo&owner=jwfy&pid=12663&registry=zookeeper&timestamp=1525733671009
  12.     if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
  13.        // 如果协议是register
  14.         String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
  15.         // 默认协议是dubbo
  16.         registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
  17.         // 替换协议为dubbo，并移除参数中的注册数据
  18.         // 现在的URL信息是
  19.         // zookeeper://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?
  20.         // application=dubbo-demo&client=zkclient&dubbo=2.5.3&export=dubbo%3A%2F%2F192.168.10.123%3A20880
  21.         //%2Fcom.jwfy.dubbo.product.ProductService%3Fanyhost%3Dtrue%26application%3Ddubbo-demo%26default.loadbalance%3Drandom%26dubbo%3D2.5.3
  22.         //%26interface%3Dcom.jwfy.dubbo.product.ProductService%26
  23.         //methods%3Dprint%2CgetStr%26owner%3Djwfy%26pid%3D12663%26
  24.         // side%3Dprovider%26timestamp%3D1525733684167%26
  25.         //token%3Dfdfdf&group=dubbo-demo&owner=jwfy&pid=12663&timestamp=1525733671009
  26.     }
  27.     return registryFactory.getRegistry(registryUrl);
  28. }
复制代码
registryFactory 同样是在RegistryProtocol实例完后注入的动态对象
  1. import com.alibaba.dubbo.common.extension.ExtensionLoader;
  2. public class RegistryFactory$Adpative implements com.alibaba.dubbo.registry.RegistryFactory {
  3.     public com.alibaba.dubbo.registry.Registry getRegistry(com.alibaba.dubbo.common.URL arg0) {
  4.    
  5.         if (arg0 == null) throw new IllegalArgumentException("url == null");
  7.         com.alibaba.dubbo.common.URL url = arg0;
  8.         String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
  10.         if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.registry.RegistryFactory) name from url(" + url.toString() + ") use keys([protocol])");
  12.         com.alibaba.dubbo.registry.RegistryFactory extension = (com.alibaba.dubbo.registry.RegistryFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.registry.RegistryFactory.class).getExtension(extName);
  14.         return extension.getRegistry(arg0);
  15.     }
  16. }
复制代码
对应实现的对象是ZookeeperRegistryFactory，调用其getRegistry方法，来到了AbstractRegistryFactory类
  1. public Registry getRegistry(URL url) {
  2.     url = url.setPath(RegistryService.class.getName())
  3.             .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
  4.             .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
  5.     String key = url.toServiceString();
  6.     // 获取的key是zookeeper://127.0.0.1:2182/dubbo-demo/com.alibaba.dubbo.registry.RegistryService
  7.     // 锁定注册中心获取过程，保证注册中心单一实例
  8.     LOCK.lock();
  9.     try {
  10.         Registry registry = REGISTRIES.get(key);
  11.         // 从缓存获取注册中心，REGISTRIES是一个线程安全的map
  12.         if (registry != null) {
  13.             return registry;
  14.         }
  15.         registry = createRegistry(url);
  16.         // 创建一个注册中心ZookeeperRegistry
  17.         if (registry == null) {
  18.             throw new IllegalStateException("Can not create registry " + url);
  19.         }
  20.         REGISTRIES.put(key, registry);
  21.         return registry;
  22.     } finally {
  23.         // 释放锁
  24.         LOCK.unlock();
  25.     }
  26. }
复制代码
AbstractRegistry 类
  1. public AbstractRegistry(URL url) {
  2.     setUrl(url);
  3.     // 启动文件保存定时器
  4.     syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
  5.     String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getHost() + ".cache");
  6.     // 文件名为/Users/XXX/.dubbo/dubbo-registry-127.0.0.1.cache
  7.     File file = null;
  8.     if (ConfigUtils.isNotEmpty(filename)) {
  9.         file = new File(filename);
  10.         if(! file.exists() && file.getParentFile() != null && ! file.getParentFile().exists()){
  11.             if(! file.getParentFile().mkdirs()){
  12.                 throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
  13.             }
  14.         }
  15.     }
  16.     this.file = file;
  17.     loadProperties();
  18.     // 从.cache 文件中获取已经存储的zk信息
  19.     notify(url.getBackupUrls());
  20.     // 通知订阅
  21. }
复制代码


来到订阅notify方法
  1. protected void notify(List<URL> urls) {
  2.     if(urls == null || urls.isEmpty()) return;
  3.    
  4.     for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
  5.         // 获取订阅者
  6.         URL url = entry.getKey();
  7.         
  8.         if(! UrlUtils.isMatch(url, urls.get(0))) {
  9.             continue;
  10.         }
  11.         
  12.         Set<NotifyListener> listeners = entry.getValue();
  13.         if (listeners != null) {
  14.             for (NotifyListener listener : listeners) {
  15.                 try {
  16.                     notify(url, listener, filterEmpty(url, urls));
  17.                     // 真正的通知触发
  18.                 } catch (Throwable t) {
  19.                     logger.error("Failed to notify registry event, urls: " +  urls + ", cause: " + t.getMessage(), t);
  20.                 }
  21.             }
  22.         }
  23.     }
  24. }
复制代码
  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.             // 处理的URL信息u和url是匹配的
  20.             String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
  21.             List<URL> categoryList = result.get(category);
  22.             if (categoryList == null) {
  23.                 categoryList = new ArrayList<URL>();
  24.                 result.put(category, categoryList);
  25.             }
  26.             categoryList.add(u);
  27.         }
  28.     }
  29.     // 一个类目下存储的多个URL信息
  30.     if (result.size() == 0) {
  31.         return;
  32.     }
  33.     Map<String, List<URL>> categoryNotified = notified.get(url);
  34.     if (categoryNotified == null) {
  35.         notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
  36.         categoryNotified = notified.get(url);
  37.     }
  38.     for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
  39.         String category = entry.getKey();
  40.         List<URL> categoryList = entry.getValue();
  41.         categoryNotified.put(category, categoryList);
  42.         saveProperties(url);
  43.         // 存储到文件中，对于上面的读取文件
  44.         listener.notify(categoryList);
  45.         // 由监听器掌握处理
  46.     }
  47. }
复制代码
FailbackRegistry 类
  1. public FailbackRegistry(URL url) {
  2.     super(url);
  3.     int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
  4.     // 设置重试的时间，默认为5s，当链接失败就会触发这个重连操作
  5.     this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
  6.         public void run() {
  7.             // 检测并连接注册中心
  8.             try {
  9.                 retry();
  10.             } catch (Throwable t) { // 防御性容错
  11.                 logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
  12.             }
  13.         }
  14.     }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
  15. }
复制代码
链接失败时，举行重试的操纵就是在这里举行的，retry就是获取当前registry中的failedRegistered等信息，如果failedRegistered中有URL信息存在，意味着之前存在链接失败的情况，如今实验retry举行重连操纵
ZookeeperRegistry 类
  1. public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
  2.     super(url);
  3.     if (url.isAnyHost()) {
  4.         throw new IllegalStateException("registry address == null");
  5.     }
  6.     String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
  7.     // dubbo 分组概念 group
  8.     if (! group.startsWith(Constants.PATH_SEPARATOR)) {
  9.         // 如果组名称不是/开头，则添加该数据
  10.         group = Constants.PATH_SEPARATOR + group;
  11.     }
  12.     this.root = group;
  13.     zkClient = zookeeperTransporter.connect(url);
  14.     // 调用zk的操作方式连接注册中心，并持有该client
  15.     zkClient.addStateListener(new StateListener() {
  16.         public void stateChanged(int state) {
  17.             if (state == RECONNECTED) {
  18.                 try {
  19.                     recover();
  20.                 } catch (Exception e) {
  21.                     logger.error(e.getMessage(), e);
  22.                 }
  23.             }
  24.         }
  25.     });
  26. }
复制代码
通过上述操纵得到的注册中央对象实例，而且其URL为zookeeper://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?application=dubbo-demo&client=zkclient&dubbo=2.5.3&group=dubbo-demo&interface=com.alibaba.dubbo.registry.RegistryService&owner=jwfy&pid=12663&timestamp=1525733671009
注册到注册中央
  1. final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
  2. // 先获取invoke对象的注册URL
  3. registry.register(registedProviderUrl);
  4. // 注册到注册中心
  5. // 此时观察zk的日志日志会发现注册操作
复制代码
从invoke对错获取的注册URL是dubbo://192.168.10.123:20880/com.jwfy.dubbo.product.ProductService?anyhost=true&application=dubbo-demo&default.loadbalance=random&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy&pid=12663&side=provider&timestamp=1525733684167&token=fdfdf
其包罗了当前bean的根本信息，把这些信息提交给注册中央，服务利用方就可以获取到这些数据，然后反转天生invoke去调用实验
FailbackRegistry 类
  1. public void register(URL url) {
  2.     super.register(url);
  3.     failedRegistered.remove(url);
  4.     failedUnregistered.remove(url);
  5.     try {
  6.         // 向服务器端发送注册请求
  7.         // 真正调用ZK包的接口注册到zk注册中心
  8.         doRegister(url);
  9.     } catch (Exception e) {
  10.         Throwable t = e;
  12.         // 如果开启了启动时检测，则直接抛出异常
  13.         boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
  14.                 && url.getParameter(Constants.CHECK_KEY, true)
  15.                 && ! Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
  16.         boolean skipFailback = t instanceof SkipFailbackWrapperException;
  17.         if (check || skipFailback) {
  18.             if(skipFailback) {
  19.                 t = t.getCause();
  20.             }
  21.             throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
  22.         } else {
  23.             logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
  24.         }
  26.         // 将失败的注册请求记录到失败列表，定时重试
  27.         // 上面的rety方法就有使用这个failedRegistered容器内的数据
  28.         failedRegistered.add(url);
  29.     }
  30. }
复制代码
以下就是zk的输出日志，可以很清晰的看到确实创建了节点信息
  1. 2018-05-08 22:52:52,808 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2182:NIOServerCnxn$Factory@251] - Accepted socket connection from /127.0.0.1:55708
  2. 2018-05-08 22:52:52,926 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2182:NIOServerCnxn@770] - Client attempting to renew session 0x163267acb7f000b at /127.0.0.1:55708
  3. 2018-05-08 22:52:52,941 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2182:NIOServerCnxn@1573] - Invalid session 0x163267acb7f000b for client /127.0.0.1:55708, probably expired
  4. 2018-05-08 22:52:52,949 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2182:NIOServerCnxn@1435] - Closed socket connection for client /127.0.0.1:55708 which had sessionid 0x163267acb7f000b
  5. 2018-05-08 22:52:53,437 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2182:NIOServerCnxn$Factory@251] - Accepted socket connection from /127.0.0.1:55709
  6. 2018-05-08 22:52:53,444 - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2182:NIOServerCnxn@777] - Client attempting to establish new session at /127.0.0.1:55709
  7. 2018-05-08 22:52:53,454 - INFO  [SyncThread:0:NIOServerCnxn@1580] - Established session 0x163267acb7f000c with negotiated timeout 30000 for client /127.0.0.1:55709
  8. 2018-05-08 22:52:56,957 - INFO  [ProcessThread:-1:PrepRequestProcessor@419] - Got user-level KeeperException when processing sessionid:0x163267acb7f000c type:create cxid:0x1 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/dubbo-demo Error:KeeperErrorCode = NodeExists for /dubbo-demo
  9. 2018-05-08 22:52:57,018 - INFO  [ProcessThread:-1:PrepRequestProcessor@419] - Got user-level KeeperException when processing sessionid:0x163267acb7f000c type:create cxid:0x2 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/dubbo-demo/com.jwfy.dubbo.product.ProductService Error:KeeperErrorCode = NodeExists for /dubbo-demo/com.jwfy.dubbo.product.ProductService
  10. 2018-05-08 22:52:57,029 - INFO  [ProcessThread:-1:PrepRequestProcessor@419] - Got user-level KeeperException when processing sessionid:0x163267acb7f000c type:create cxid:0x3 zxid:0xfffffffffffffffe txntype:unknown reqpath:n/a Error Path:/dubbo-demo/com.jwfy.dubbo.product.ProductService/providers Error:KeeperErrorCode = NodeExists for /dubbo-demo/com.jwfy.dubbo.product.ProductService/providers
复制代码
到这里服务注册到注册中央就已经完成了，同时还陪同着从文件加载注册信息和生存注册信息，可自行通过zKcli下令去
袒露服务之注册
  1. private <T> ExporterChangeableWrapper<T>  doLocalExport(final Invoker<T> originInvoker){
  2.     // invoke的URL信息是registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?
  3.     // application=dubbo-demo&client=zkclient&dubbo=2.5.3&export=dubbo%3A%2F%2F172.16.109.110%3A20880%2F
  4.     // com.jwfy.dubbo.product.ProductService%3Fanyhost%3Dtrue%26application%3D
  5.     // dubbo-demo%26default.loadbalance%3Drandom%26dubbo%3D2.5.3%26
  6.     // interface%3Dcom.jwfy.dubbo.product.ProductService%26methods%3Dprint%2CgetStr%26
  7.     // owner%3Djwfy%26pid%3D13375%26side%3Dprovider%26timestamp%3D1525749656129%26
  8.     // token%3Dfdfdf&group=dubbo-demo&owner=jwfy&pid=13375&registry=zookeeper&
  9.     // timestamp=1525749652060
  10.     String key = getCacheKey(originInvoker);
  11.     // key就是dubbo://172.16.109.110:20880/com.jwfy.dubbo.product.ProductService?  
  12.     // anyhost=true&application=dubbo-demo&default.loadbalance=random&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService
  13.     // &methods=print,getStr&owner=jwfy&pid=13375&side=provider&timestamp=1525749656129&token=fdfdf
  14.     ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
  15.     if (exporter == null) {
  16.         synchronized (bounds) {
  17.             exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
  18.             if (exporter == null) {
  19.                 // 又出现了double-check操作
  20.                 final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
  21.                 // 包装类
  22.                 exporter = new ExporterChangeableWrapper<T>((Exporter<T>)protocol.export(invokerDelegete), originInvoker);
  23.                
  24.                 bounds.put(key, exporter);
  25.             }
  26.         }
  27.     }
  28.     return (ExporterChangeableWrapper<T>) exporter;
  29. }
复制代码
在上面的protocol.export操纵中，protocol也不是DubboProtocol自己，而是包装了ProtocolFilterWrapper、ProtocolListenerWrapper，协议不是register，各种处置处罚之后进入到DubboProtocol的export举行袒露操纵。
网络端口开启

DubboProtocol 类
  1. public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
  2.     URL url = invoker.getUrl();
  3.     // URL是dubbo://172.16.109.110:20880/com.jwfy.dubbo.product.ProductService?anyhost=true&
  4.     //application=dubbo-demo&default.loadbalance=random&dubbo=2.5.3&
  5.     //interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy
  6.    // &pid=13375&side=provider&timestamp=1525749656129&token=fdfdf
  7.    
  8.     // export service.
  9.     String key = serviceKey(url);
  10.     // key是com.jwfy.dubbo.product.ProductService:20880
  11.     DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
  12.     // 组成成为一个新的包装类DubboExporter
  13.     exporterMap.put(key, exporter);
  14.    
  15.     //export an stub service for dispaching event
  16.     Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
  17.     Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
  18.     if (isStubSupportEvent && !isCallbackservice){
  19.         String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
  20.         if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
  21.             if (logger.isWarnEnabled()){
  22.                 logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
  23.                         "], has set stubproxy support event ,but no stub methods founded."));
  24.             }
  25.         } else {
  26.             stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
  27.         }
  28.     }
  30.     openServer(url);
  31.     // 来了，最关键的时候
  32.    
  33.     return exporter;
  34. }
复制代码
  1. private ExchangeServer createServer(URL url) {
  2.     //默认开启server关闭时发送readonly事件
  3.     url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
  4.     //默认开启heartbeat
  5.     url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
  6.     String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
  7.     // RPC服务的名称，默认是netty
  9.     if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
  10.         throw new RpcException("Unsupported server type: " + str + ", url: " + url);
  12.     url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
  13.     ExchangeServer server;
  14.     try {
  15.         server = Exchangers.bind(url, requestHandler);
  16.         // 绑定操作
  17.     } catch (RemotingException e) {
  18.         throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
  19.     }
  20.     str = url.getParameter(Constants.CLIENT_KEY);
  21.     if (str != null && str.length() > 0) {
  22.         Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
  23.         if (!supportedTypes.contains(str)) {
  24.             throw new RpcException("Unsupported client type: " + str);
  25.         }
  26.     }
  27.     return server;
  28. }
复制代码
Exchangers 类
  1. public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
  2.     if (url == null) {
  3.         throw new IllegalArgumentException("url == null");
  4.     }
  5.     if (handler == null) {
  6.         throw new IllegalArgumentException("handler == null");
  7.     }
  8.     url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
  9.     return getExchanger(url).bind(url, handler);
  10.     // getExchanger 是通过SPI返回一个HeaderExchanger对象
  11. }
复制代码
HeaderExchanger 类
  1. public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
  2.     // 包装一个HeaderExchangeHandler对象
  3.     // 再包装一个DecodeHandler对象
  4.     // Transporters.bind绑定操做
  5.     // 最后包装成HeaderExchangeServer返回
  6.     return new HeaderExchangeServer(
  7.         Transporters.bind(url,
  8.             new DecodeHandler(
  9.                 new HeaderExchangeHandler(handler)
  10.             )
  11.         )
  12.     );
  13. }
复制代码
此中Transporters.bind会先获取当前可用的此中Transporters，默认也是NettyTransporter对象，调用其bind方法
new NettyServer(url, listener)，来到了AbstractServer类
  1. public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
  2.     super(url, handler);
  3.     localAddress = getUrl().toInetSocketAddress();
  4.     // 获取本地套接字地址，也就是IP端口
  5.     String host = url.getParameter(Constants.ANYHOST_KEY, false)
  6.                     || NetUtils.isInvalidLocalHost(getUrl().getHost())
  7.                     ? NetUtils.ANYHOST : getUrl().getHost();
  8.     bindAddress = new InetSocketAddress(host, getUrl().getPort());
  9.     // 绑定地址，如果是本地项目，host为"0.0.0.0"
  10.     this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
  11.     this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
  12.     try {
  13.         doOpen();
  14.         // 连接打开，当前默认是使用的NettyServer里的方法
  15.         // 关于Netty的操作细节目前也不是很理解，后续补充
  16.         if (logger.isInfoEnabled()) {
  17.             logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
  18.         }
  19.     } catch (Throwable t) {
  20.         throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
  21.                                     + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
  22.     }
  23.     if (handler instanceof WrappedChannelHandler ){
  24.         executor = ((WrappedChannelHandler)handler).getExecutor();
  25.     }
  26. }
复制代码
其终极返回一个NettyService,服务端已经开启了，客户端可以毗连了，继承跳入到HeaderExchangeServer中
开启心跳检测

HeaderExchangeServer 类
  1. public HeaderExchangeServer(Server server) {
  2.     // 此处传递的service就是上面生成的NettyService
  3.     if (server == null) {
  4.         throw new IllegalArgumentException("server == null");
  5.     }
  6.     this.server = server;
  7.     // 开始心跳检测
  8.     this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
  9.     // 默认是0（可是没找到在哪设置为了6000，也就是6s）
  10.     this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);
  11.     // 定时任务间隔期 6*3 = 18s
  12.     // 如果没有设置心跳检测的间隔期，则设置为心跳延迟时间的3倍
  13.     if (heartbeatTimeout < heartbeat * 2) {
  14.         // 设置的时间不符合要求
  15.         throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
  16.     }
  17.     startHeatbeatTimer();
  18.     // 开启心跳检测
  19. }
复制代码
心跳检测末了真正实验的任务是如下代码
HeartBeatTask 类
  1. public void run() {
  2.     try {
  3.         long now = System.currentTimeMillis();
  4.         // 具体所有IO的channel是通过HeaderExchangeServer.this.getChannels()获取到的
  5.         for ( Channel channel : channelProvider.getChannels() ) {
  6.             if (channel.isClosed()) {
  7.                 // 确保可用的channel
  8.                 continue;
  9.             }
  10.             try {
  11.                 Long lastRead = ( Long ) channel.getAttribute(
  12.                         HeaderExchangeHandler.KEY_READ_TIMESTAMP );
  13.                 Long lastWrite = ( Long ) channel.getAttribute(
  14.                         HeaderExchangeHandler.KEY_WRITE_TIMESTAMP );
  15.                 if ( ( lastRead != null && now - lastRead > heartbeat )
  16.                         || ( lastWrite != null && now - lastWrite > heartbeat ) ) {
  17.                     Request req = new Request();
  18.                     req.setVersion( "2.0.0" );
  19.                     req.setTwoWay( true );
  20.                     req.setEvent( Request.HEARTBEAT_EVENT );
  21.                     channel.send( req );
  22.                     // 发生心跳检测包 req是内容
  23.                     if ( logger.isDebugEnabled() ) {
  24.                         logger.debug( "Send heartbeat to remote channel " + channel.getRemoteAddress()
  25.                                               + ", cause: The channel has no data-transmission exceeds a heartbeat period: " + heartbeat + "ms" );
  26.                     }
  27.                 }
  28.                 if ( lastRead != null && now - lastRead > heartbeatTimeout ) {
  29.                     // channel因为超时可能存在关闭的情况
  30.                     logger.warn( "Close channel " + channel
  31.                                          + ", because heartbeat read idle time out: " + heartbeatTimeout + "ms" );
  32.                     if (channel instanceof Client) {
  33.                         try {
  34.                             ((Client)channel).reconnect();
  35.                         }catch (Exception e) {
  36.                             //do nothing
  37.                         }
  38.                     } else {
  39.                         channel.close();
  40.                     }
  41.                 }
  42.             } catch ( Throwable t ) {
  43.                 logger.warn( "Exception when heartbeat to remote channel " + channel.getRemoteAddress(), t );
  44.             }
  45.         }
  46.     } catch ( Throwable t ) {
  47.         logger.warn( "Unhandled exception when heartbeat, cause: " + t.getMessage(), t );
  48.     }
  49. }
复制代码
到此整个的服务袒露就全部竣事了
总结

Spring

基于spring开发，做到无缝对接spring，在利用上只有少少数的xml设置学习本钱，在Dubbo内部继承了多少个类，自界说实现的xsd和配套的分析方法等。如果以后须要接入自界说的NameSpaceContext，同样须要类似的处置处罚

  • 通过BeanNameAware修改bean的名称
  • ApplicationContextAware去获取Spring IOC的容器
  • IntializingBean的afterPropertiesSet去自界说实现bean的实例对象
  • ApplicationListener的onApplicationEvent罗致各种变乱
  • DisposableBean的destroy去烧毁bean
获取Spring IOC容器之后，在自界说实现dubbo的服务bean时，就可以获取到体系设置的注册中央、利用的协议、平衡负责等内容（对注册中央、协议等对Spring而言只是个平常的bean而言），如许就可以完成对ServiceBean的属性输入。
SPI

固然也先容过SPI的整个的处置处罚细节，在整个的dubbo中也确实大量利用中。但是真正的在调试服务袒露中还是有不少细节被忽略掉了。

  • 根本上全部的类都有通过拼接代码字符串再编译天生其对象
  • 动态天生的对象着实并没有做许多变乱，只是根据设置的xml参数，动态决定选取符合的实现类，再调用其方法
  • 参数注入和spring的参数注入方法不一样的是，dubbo是获取类的全部set开头同时参数只有1个的公共方法，通过SPI取到参数的值，然后反射invoke注入的
  • Dubbo为了一些附加的功能，存在包装类的情况，每一个接口都有wrapper的存在，如果存在则末了返回的对象不是对象自己，而是颠末层层包装产生的对象，比方Protocol
服务袒露流程

如果是我们去完成这个任务会怎么去做呢？有两点肯定是要实现的

  • 注册中央，须要把服务提供方的信息推送给注册中央同一管理，服务调用方可以或许感知到存在这个服务
  • 对外的网络端口，可通过套接字和外界（服务调用方）发生信息互换
服务调用方感知到服务的存在，然后获取其服务的套接字信息，然后直接通过网络IO完成信息互换
Dubbo着实也根本上按照这种思绪来实现的，只是支持了多种协议，而且另有监控、统计等功能，井然有序。
获取注册中央

注册中央的根本设置就是在xml设置的dubbo:registry,天生的URL也是注册协议的URL，比方，registry://127.0.0.1:2182/com.alibaba.dubbo.registry.RegistryService?application=dubbo-demo&client=zkclient&dubbo=2.5.3&group=dubbo-demo&owner=jwfy&pid=2772&registry=zookeeper&timestamp=1525276569763 会把一个dubbo的xml设置的根本信息给提取出去
服务信息的组装

既然注册中央已经准备好了，那如今就须要处置处罚我们对外袒露的service信息，对外袒露服务也须要设置的，这个设置就是从dubbo:protocol来的。
这个设置会告诉体系，全部对外袒露的IO读取协议是什么，默认的是dubbo，着实也支持redis、http等
是处置处罚IO数据流方式的协议
不外，如果没有IP、端口等数据，则从dubbo:provider提取
再设置其他属性数据，天生的URL，比方dubbo://172.16.109.110:20880/com.jwfy.dubbo.product.ProductService?anyhost=true&application=dubbo-demo&default.loadbalance=random&dubbo=2.5.3&interface=com.jwfy.dubbo.product.ProductService&methods=print,getStr&owner=jwfy &pid=2772&side=provider&timestamp=1525313423899
这个对外的服务利用的袒露的接口、具体可以被调用的函数 属于提供方
服务IO端口提供

既然网络IO端的协议数据已经准备好了，那就可以开启，等候服务方的调用
这点可以直接跳到DubboProtocol类的export操纵了（以Dubbo协议为例子）
针对每一个URL都提取出一个key，然后存储对应的invoker、url等信息
真正操纵中取的是服务名+协议端口为key，DubboExporter为value的一个map，后续就可以通过key反取出invoke对象
接下来就是利用哪种Transporter操纵了，默认是利用了netty的NettyTransporter（netty是一种基于java开发的异步的、变乱驱动的高效的网络IO框架），如果须要利用其他的IO框架，则须要在dubbo:provider设置service字段
如今网络对外的协议Netty已经准备好了，服务信息也已经准备好了，那接下来就按照特定的格式decode操纵，把数据变化为字节流，由netty向外开放出来即可
由此服务调用方就可以通过网络IO 链接到服务提供方上
服务的注册

服务的注册是另一个操纵，也就是把当条件供的服务告诉给注册中央，后续服务调用方就可以订阅注册中央，从而知道有哪些服务了，当前例子是利用了zookeeper作为注册中央
如今我们知道的是对外袒露服务的url信息（也就是invoke对象中的url）
以是操纵也就很显着了，先获取到注册中央的设置信息，然后把服务url信息转为zk协议的url信息，末了毗连到注册中央，注册生存即可
这内里就有利用zookeeper的jar包功能，须要和注册中央交互的操纵
别的另有些附属的功能，比方超时重试、毗连信息的生存和读取（如果注册中央忽然出现标题，服务端和客户端还可以利用本地文件缓存继承工作，只是不能再及时获取最新的订阅信息罢了）
一样平常本地的毗连信息存储在/Users/XXX/.dubbo中，如果出现了无法注册无法调用的情况，可以思量删除该文件重启服务
回复

使用道具 举报

返回列表

麻花痒
+ 我要发帖

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表