蒋先森のBlog

一个刚起航的后端菜鸟

0%

Dubbo源码分析之服务导出

Posted on Edited on In

源码版本是 2.7.8

Dubbo 服务导出过程始于 Spring 容器发布刷新事件，Dubbo 在接收到事件后，会立即执行服务导出逻辑。整个逻辑大致可分为三个部分，第一部分是前置工作，主要用于检查参数，组装 URL。第二部分是导出服务，包含导出服务到本地 (JVM)，和导出服务到远程两个过程。第三部分是向注册中心注册服务，用于服务发现。本篇文章将会对这三个部分代码进行详细的分析。

在 2.7.8 版本中，服务导出的入口已经在 DubboBootstrapApplicationListener onApplicationContextEvent方法是在 onApplicationEvent中引用的，onApplicationEvent 是一个事件响应方法，该方法会在收到 Spring 上下文刷事件后执行服务导出操作。

在DubboBootstrap.start调用中，会调用一个exportServices方法，这个方法中，会调用 export 方法，这是就开始服务导出流程了。

前置工作

export

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public synchronized void export() {
    if (bootstrap == null) {
        bootstrap = DubboBootstrap.getInstance();
        bootstrap.initialize();
    }

    checkAndUpdateSubConfigs();

    //init serviceMetadata 初始化服务元数据
    serviceMetadata.setVersion(getVersion());
    serviceMetadata.setGroup(getGroup());
    serviceMetadata.setDefaultGroup(getGroup());
    serviceMetadata.setServiceType(getInterfaceClass());
    serviceMetadata.setServiceInterfaceName(getInterface());
    serviceMetadata.setTarget(getRef());

    //如果不允许暴露，直接放过 <dubbo:provider export="false" />
    if (!shouldExport()) {
        return;
    }
    // 如果需要延迟，则延迟暴露
    if (shouldDelay()) {
        DELAY_EXPORT_EXECUTOR.schedule(this::doExport, getDelay(), TimeUnit.MILLISECONDS);
    } else {
        // 直接暴露
        doExport();
    }
    exported();
}

ServiceConfig.export是继承重写了父类ServiceConfigBase.export方法。在export方法中，主要是以下逻辑：

  • 检测 dubbo:service 标签的 interface 属性合法性，不合法则抛出异常
  • 检测 ProviderConfig、ApplicationConfig 等核心配置类对象是否为空，若为空，则尝试从其他配置类对象中获取相应的实例。
  • 检测并处理泛化服务和普通服务类
  • 检测本地存根配置，并进行相应的处理
  • 对 ApplicationConfig、RegistryConfig 等配置类进行检测，为空则尝试创建，若无法创建则抛出异常
  • 主要是初始化服务元数据
  • 对export配置项进行检查，判断是否需要导出服务
  • 如果需要延迟，则延迟导出，否则直接导出服务

doExport

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
protected synchronized void doExport() {
    if (unexported) {
        throw new IllegalStateException("The service " + interfaceClass.getName() + " has already unexported!");
    }
    if (exported) {
        return;
    }
    exported = true;

    if (StringUtils.isEmpty(path)) {
        path = interfaceName;
    }
    // 重点关注
    doExportUrls();
}

这个方法没啥逻辑，前面是一些校验，重点关注doExportUrls方法。

doExportUrls

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
private void doExportUrls() {
    ServiceRepository repository = ApplicationModel.getServiceRepository();
    ServiceDescriptor serviceDescriptor = repository.registerService(getInterfaceClass());
    repository.registerProvider(
            getUniqueServiceName(),
            ref,
            serviceDescriptor,
            this,
            serviceMetadata
    );
    // 获取当前服务的注册中心，可以有多个服务中心
    List<URL> registryURLs = ConfigValidationUtils.loadRegistries(this, true);
    // // 遍历 protocols，并在每个协议下导出服务
    for (ProtocolConfig protocolConfig : protocols) {
        // 凭借key，全限定名+group+version。类似 com.dubbo.api.service.AsyncService:1.0.0
        String pathKey = URL.buildKey(getContextPath(protocolConfig)
                .map(p -> p + "/" + path)
                .orElse(path), group, version);
        // In case user specified path, register service one more time to map it to path.
        repository.registerService(pathKey, interfaceClass);
        // TODO, uncomment this line once service key is unified
        serviceMetadata.setServiceKey(pathKey);
        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
    }
}

Dubbo 允许我们使用不同的协议导出服务，也允许我们向多个注册中心注册服务。Dubbo 在 doExportUrls 方法中对多协议，多注册中心进行了支持。从上面的代码中，可以知道，通过 loadRegistries 加载注册中心链接，然后再遍历 ProtocolConfig 集合导出每个服务。并在导出服务的过程中，将服务注册到注册中心。

ConfigValidationUtils.loadRegistries 方法主要功能：

  • 检测是否存在注册中心配置类，不存在则抛出异常
  • 构建参数映射集合，也就是 map
  • 构建注册中心链接列表
  • 遍历链接列表，并根据条件决定是否将其添加到 registryList 中

组装URL

这个方法主要的功能是组装 URL。URL 是 Dubbo 配置的载体，通过 URL 可让 Dubbo 的各种配置在各个模块之间传递。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
    String name = protocolConfig.getName();
    if (StringUtils.isEmpty(name)) {
        name = DUBBO;
    }
    // 根据配置往map中塞值，构建URL
    Map<String, String> map = new HashMap<String, String>();
    map.put(SIDE_KEY, PROVIDER_SIDE);

    ServiceConfig.appendRuntimeParameters(map);
    AbstractConfig.appendParameters(map, getMetrics());
    AbstractConfig.appendParameters(map, getApplication());
    AbstractConfig.appendParameters(map, getModule());
    // remove 'default.' prefix for configs from ProviderConfig
    // appendParameters(map, provider, Constants.DEFAULT_KEY);
    AbstractConfig.appendParameters(map, provider);
    AbstractConfig.appendParameters(map, protocolConfig);
    AbstractConfig.appendParameters(map, this);
    MetadataReportConfig metadataReportConfig = getMetadataReportConfig();
    if (metadataReportConfig != null && metadataReportConfig.isValid()) {
        map.putIfAbsent(METADATA_KEY, REMOTE_METADATA_STORAGE_TYPE);
    }
    // methods 为 MethodConfig 集合，MethodConfig 中存储了 <dubbo:method> 标签的配置信息
    if (CollectionUtils.isNotEmpty(getMethods())) {
        // 省略代码
    }

    if (ProtocolUtils.isGeneric(generic)) {
        map.put(GENERIC_KEY, generic);
        map.put(METHODS_KEY, ANY_VALUE);
    } else {
        String revision = Version.getVersion(interfaceClass, version);
        if (revision != null && revision.length() > 0) {
            map.put(REVISION_KEY, revision);
        }

        String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
        if (methods.length == 0) {
            logger.warn("No method found in service interface " + interfaceClass.getName());
            map.put(METHODS_KEY, ANY_VALUE);
        } else {
            map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
        }
    }

    /**
     * Here the token value configured by the provider is used to assign the value to ServiceConfig#token
     */
    if (ConfigUtils.isEmpty(token) && provider != null) {
        token = provider.getToken();
    }

    if (!ConfigUtils.isEmpty(token)) {
        if (ConfigUtils.isDefault(token)) {
            map.put(TOKEN_KEY, UUID.randomUUID().toString());
        } else {
            map.put(TOKEN_KEY, token);
        }
    }
    //init serviceMetadata attachments
    serviceMetadata.getAttachments().putAll(map);

    // export service
    String host = findConfigedHosts(protocolConfig, registryURLs, map);
    Integer port = findConfigedPorts(protocolConfig, name, map);
    // 构建URL
    URL url = new URL(name, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), map);

    // 省略代码 ...
}

上面的代码首先是将一些信息，比如版本、时间戳、方法名以及各种配置对象的字段信息放入到 map 中，map 中的内容将作为 URL 的查询字符串。构建好 map 后，紧接着是获取上下文路径、主机名以及端口号等信息。最后将 map 和主机名等数据传给 URL 构造方法创建 URL 对象。需要注意的是，这里出现的 URL 并非 java.net.URL，而是 com.alibaba.dubbo.common.URL。

上面在if判断中省略了一段代码。这段代码用于检测 dubbo:method 标签中的配置信息，并将相关配置添加到 map 中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
for (MethodConfig method : getMethods()) {
    AbstractConfig.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 (CollectionUtils.isNotEmpty(arguments)) {
        for (ArgumentConfig argument : arguments) {
            // convert argument type
            if (argument.getType() != null && argument.getType().length() > 0) {
                Method[] methods = interfaceClass.getMethods();
                // visit all methods
                if (methods.length > 0) {
                    for (int i = 0; i < methods.length; i++) {
                        String methodName = methods[i].getName();
                        // target the method, and get its signature
                        if (methodName.equals(method.getName())) {
                            Class<?>[] argtypes = methods[i].getParameterTypes();
                            // one callback in the method
                            if (argument.getIndex() != -1) {
                                if (argtypes[argument.getIndex()].getName().equals(argument.getType())) {
                                    AbstractConfig.appendParameters(map, argument, method.getName() + "." + argument.getIndex());
                                } else {
                                    throw new IllegalArgumentException("Argument config error : the index attribute and type attribute not match :index :" + argument.getIndex() + ", type:" + argument.getType());
                                }
                            } else {
                                // multiple callbacks in the method
                                for (int j = 0; j < argtypes.length; j++) {
                                    Class<?> argclazz = argtypes[j];
                                    if (argclazz.getName().equals(argument.getType())) {
                                        AbstractConfig.appendParameters(map, argument, method.getName() + "." + j);
                                        if (argument.getIndex() != -1 && argument.getIndex() != j) {
                                            throw new IllegalArgumentException("Argument config error : the index attribute and type attribute not match :index :" + argument.getIndex() + ", type:" + argument.getType());
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            } else if (argument.getIndex() != -1) {
                AbstractConfig.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

导出服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
    // 省略无关代码 ...
    // You can customize Configurator to append extra parameters
    if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
            .hasExtension(url.getProtocol())) {
        url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
    }

    String scope = url.getParameter(SCOPE_KEY);
    // don't export when none is configured 当scope=none时，不进行服务暴露
    if (!SCOPE_NONE.equalsIgnoreCase(scope)) {
        // 如果不是远程暴露，进行本地暴露。如果scope是null，则会进入分支，进行本地暴露
        if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
            exportLocal(url);
        }
        // 如果不是本地暴露，进行远程暴露
        if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
            // 如果有注册中心，进行远程暴露
            if (CollectionUtils.isNotEmpty(registryURLs)) {
                for (URL registryURL : registryURLs) {
                    // 判断协议是否是 injvm，是的话，不进行注册了
                    if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
                        continue;
                    }
                    url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));
                    // 如果有监控中心，则会添加之后向其汇报
                    URL monitorUrl = ConfigValidationUtils.loadMonitor(this, registryURL);
                    if (monitorUrl != null) {
                        url = url.addParameterAndEncoded(MONITOR_KEY, monitorUrl.toFullString());
                    }
                    if (logger.isInfoEnabled()) {
                        if (url.getParameter(REGISTER_KEY, true)) {
                            logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                        } else {
                            logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                        }
                    }
                    // For providers, this is used to enable custom proxy to generate invoker
                    String proxy = url.getParameter(PROXY_KEY);
                    if (StringUtils.isNotEmpty(proxy)) {
                        registryURL = registryURL.addParameter(PROXY_KEY, proxy);
                    }
                    // 将具体实现类封装成 invoker
                    Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
                    // 包装转换成 exporter
                    Exporter<?> exporter = PROTOCOL.export(wrapperInvoker);
                    exporters.add(exporter);
                }
            } else {
                if (logger.isInfoEnabled()) {
                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                }
                Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, url);
                DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                Exporter<?> exporter = PROTOCOL.export(wrapperInvoker);
                exporters.add(exporter);
            }
            MetadataUtils.publishServiceDefinition(url);
        }
    }
    this.urls.add(url);
}

上面代码根据 url 中的 scope 参数决定服务导出方式，分别如下：

  • scope = none，不导出服务
  • scope != remote，导出到本地
  • scope != local，导出到远程

Invoker 创建过程

导出服务到本地

导出服务到远程

注册服务

Reference

客官，赏一杯coffee嘛~~~~