Dubbo源码理解

2.2 Dubbo 远程调用细节

2.2.1 服务提供者暴露一个服务的概要过程

下面我们结合时序图来讲解 ExtensionLoader 的 getAdaptiveExtension() 方法是如何动态生成扩展接口对应的适配器类，以及 getExtension() 方法如何根据扩展实现类的名称找到对应的扩展实现类的。

上面的时序图中看起来可能比较迷茫，结合代码debug可能更加直观，如下手写了一个dubbo的服务提供者的代码：

public class ApiProvider {

    public static void main(String[] args) throws IOException {
        // 1.创建ServiceConfig实例
        ServiceConfig<GreetingService> serviceConfig = new ServiceConfig<GreetingService>();
        // 2.设置应用程序配置
        serviceConfig.setApplication(new ApplicationConfig("first-dubbo-provider"));

        // 3.设置服务注册中心信息
        RegistryConfig registryConfig = new RegistryConfig("zookeeper://127.0.0.1:2181");
        serviceConfig.setRegistry(registryConfig);
        // 4.设置接口与实现类
        serviceConfig.setInterface(GreetingService.class);
        serviceConfig.setRef(new GreetingServiceImpl());

        // 5.设置服务分组与版本
        serviceConfig.setVersion("1.0.0");
        serviceConfig.setGroup("dubbo");
        
        // 7.导出服务
        serviceConfig.export();

        // 8.挂起线程，避免服务停止
        System.out.println("server is started");
        System.in.read();
    }
}
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

可以通过debug第一行，来观察

断点断到第一行，进入：

这里可以看到在创建 ServiceConfig 对象的时候，初始化了 Protocol 对象，这里就是通过ExtensionLoader 对象来初始化的。所以得出结论，在dubbo中，要想初始化一个类，需要通过ExtensionLoader 的静态方法 getExtensionLoader 方法来进行加载，之所以需要这个方法来加载，是因为可以通过 dubbo 特殊的 SPI机制，对扩展接口实现适配器类，这样比较方便扩展。
在 Dubbo中的可扩展接口，都可以通过ExtensionLoader 对象，来扩展。
代码如下：

    public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) {
        if (type == null) {
            throw new IllegalArgumentException("Extension type == null");
        }
        if (!type.isInterface()) {
            throw new IllegalArgumentException("Extension type (" + type + ") is not an interface!");
        }
        if (!withExtensionAnnotation(type)) {
            throw new IllegalArgumentException("Extension type (" + type +
                    ") is not an extension, because it is NOT annotated with @" + SPI.class.getSimpleName() + "!");
        }

        ExtensionLoader<T> loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
        if (loader == null) {
            EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader<T>(type));
            loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
        }
        return loader;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

可以看到源码中，有个EXTENSION_LOADERS，它的定义如下：

private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
1

这是一个map，而且是线程安全的并发map，它通过这个map来获取对应接口的ExtensionLoader映射，可以看到它的key是 Class 对象，value 就是ExtensionLoader 的对象。
可以看到这里有个判断，如果loader是空的，那么旧会新建一个ExtensionLoader 的对象，放到这个缓存中，下次就可以直接用了。
上面的时许图中的第二步，是 getAdaptiveExtension 方法，该方法是用来获取当前扩展接口对应的适配器对象的。代码如下：

public T getAdaptiveExtension() {
        Object instance = cachedAdaptiveInstance.get();
        if (instance == null) {
            if (createAdaptiveInstanceError == null) {
                synchronized (cachedAdaptiveInstance) {
                    instance = cachedAdaptiveInstance.get();
                    if (instance == null) {
                        try {
                            instance = createAdaptiveExtension();
                            cachedAdaptiveInstance.set(instance);
                        } catch (Throwable t) {
                            createAdaptiveInstanceError = t;
                            throw new IllegalStateException("Failed to create adaptive instance: " + t.toString(), t);
                        }
                    }
                }
            } else {
                throw new IllegalStateException("Failed to create adaptive instance: " + createAdaptiveInstanceError.toString(), createAdaptiveInstanceError);
            }
        }

        return (T) instance;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

上面的代码通过双重锁检查创建 cachedAdaptiveInstance 对象，接口对应的适配器对象就保存到这个对象里。
对于双重锁检查如果不了解可以查看其他资料说明：https://blog.csdn.net/weixin_39158966/article/details/130314709
可以参考这篇文章，这里不过多赘述，总之就是经过一系列的线程安全性的设置以后，我们得到了通过 createAdaptiveExtension 方法创建的对象，所以现在的关键点就在于 createAdaptiveExtension 方法的实现，可以看出，具体创建出适配器对象就是通过这个方法实现的。代码如下：

    private T createAdaptiveExtension() {
        try {
            return injectExtension((T) getAdaptiveExtensionClass().newInstance());
        } catch (Exception e) {
            throw new IllegalStateException("Can't create adaptive extension " + type + ", cause: " + e.getMessage(), e);
        }
    }
1
2
3
4
5
6
7

上面的代码通过 getAdaptiveExtensionClass().newInstance() 来获取适配器对象的一个实例，然后调用 injectExtension 方法进行扩展点相互依赖注入（扩展点之间的依赖自动注入）。可以简单的理解为，通过这个方法，可以把我们新创建出来的bean，依赖的一些对象，自动的帮我们注入到这个对象中。
这里可以看出 getAdaptiveExtensionClass 方法返回的是 Class 对象，因为它使用了 newInstance 这个 API 来进行对象的创建。
下面是 getAdaptiveExtensionClass 方法的代码，来看看它是怎么创建的适配器的 Class 对象：

private Class<?> getAdaptiveExtensionClass() {
        getExtensionClasses();
        if (cachedAdaptiveClass != null) {
            return cachedAdaptiveClass;
        }
        return cachedAdaptiveClass = createAdaptiveExtensionClass();
    }
1
2
3
4
5
6
7

可以看到，先调用了一个 getExtensionClasses 方法，对应上面的时序图中的第五步，这个方法的含义是获取所有扩展了这个接口的实现类，比如说我们现在看的：

private static final Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
1

那么这里获取到的类就是所有扩展了 Protocol 接口的实现类的 Class 对象，如下是我本地 debug的结果：

可以看到这里获取到了所有实现了Procotol接口的类的Class对象。
接下来就调用了 createAdaptiveExtensionClass 方法，这个方法就是创建具体的适配器对象的 Class 对象。
该方法内部会根据字符串代码生成适配器的 Class 对象并返回，然后通过 newInstance 方法创建适配器类的一个对象实例。
到这里，一个扩展接口的适配器对象就创建完毕了。
这里再详细的说明一下 getExtensionClasses 方法，这个方法刚刚是说用来获取接口的实现类的 Class 对象。代码如下：

    private Map<String, Class<?>> getExtensionClasses() {
        Map<String, Class<?>> classes = cachedClasses.get();
        if (classes == null) {
            synchronized (cachedClasses) {
                classes = cachedClasses.get();
                if (classes == null) {
                    classes = loadExtensionClasses();
                    cachedClasses.set(classes);
                }
            }
        }
        return classes;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

如果cachedClasses方法中没有缓存到对象，那么就需要调用 loadExtensionClasses 方法。其代码如下：

    // synchronized in getExtensionClasses
    private Map<String, Class<?>> loadExtensionClasses() {
        // 获取默认扩展名
        cacheDefaultExtensionName();

        // 在指定目录的 Jar里面查找扩展点
        Map<String, Class<?>> extensionClasses = new HashMap<>();
        loadDirectory(extensionClasses, DUBBO_INTERNAL_DIRECTORY, type.getName());
        loadDirectory(extensionClasses, DUBBO_INTERNAL_DIRECTORY, type.getName().replace("org.apache", "com.alibaba"));
        loadDirectory(extensionClasses, DUBBO_DIRECTORY, type.getName());
        loadDirectory(extensionClasses, DUBBO_DIRECTORY, type.getName().replace("org.apache", "com.alibaba"));
        loadDirectory(extensionClasses, SERVICES_DIRECTORY, type.getName());
        loadDirectory(extensionClasses, SERVICES_DIRECTORY, type.getName().replace("org.apache", "com.alibaba"));
        return extensionClasses;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

cacheDefaultExtensionName 方法的详细代码如下：

    /**
     * extract and cache default extension name if exists
     */
    private void cacheDefaultExtensionName() {
        // 获取接口上的 SPI 注解
        final SPI defaultAnnotation = type.getAnnotation(SPI.class);
        // 检查是否存在注解
        if (defaultAnnotation != null) {
            String value = defaultAnnotation.value();
            if ((value = value.trim()).length() > 0) {
                String[] names = NAME_SEPARATOR.split(value);
                if (names.length > 1) {
                    throw new IllegalStateException("More than 1 default extension name on extension " + type.getName()
                            + ": " + Arrays.toString(names));
                }
                if (names.length == 1) {
                    cachedDefaultName = names[0];
                }
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

接下来就会在META-INF/dubbo/internal/、META-INF/dubbo/、META-INF/services/ 目录下去加载具体的扩展实现类，如图所示：

接下来第7步就是调用 injectExtension() 方法，该方法时对扩展点实现类进行依赖注入动作，也就是我们所熟知的 IoC。

    private T injectExtension(T instance) {
        try {
            if (objectFactory != null) {
            	// 遍历扩展点实现类所有的方法
                for (Method method : instance.getClass().getMethods()) {
                    if (isSetter(method)) {
                        /**
                         * Check {@link DisableInject} to see if we need auto injection for this property
                         * 如果方法含有 DisableInject 注解，说明该属性不需要自动注入
                         */
                        if (method.getAnnotation(DisableInject.class) != null) {
                            continue;
                        }
                        // 第一个参数是原始类型，则不需要自动注入
                        Class<?> pt = method.getParameterTypes()[0];
                        if (ReflectUtils.isPrimitives(pt)) {
                            continue;
                        }
                        // 查看 set() 方法设置的变量是不是有扩展接口实现
                        try {
                        	// 获取 setter 对应的属性名，比如 setVersion，则返回 "version"
                            String property = getSetterProperty(method);
                            // 查看该属性类似是否存在扩展实现
                            Object object = objectFactory.getExtension(pt, property);
                            // 如果存在则反射调用setter()方法进行属性注入
                            if (object != null) {
                                method.invoke(instance, object);
                            }
                        } catch (Exception e) {
                            logger.error("Failed to inject via method " + method.getName()
                                    + " of interface " + type.getName() + ": " + e.getMessage(), e);
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return instance;
    }

    /**
     * 获取setter对应的属性名，比如setVersion则返回"version"
     * get properties name for setter, for instance: setVersion, return "version"
     * 

     * return "", if setter name with length less than 3
     */
    private String getSetterProperty(Method method) {
        return method.getName().length() > 3 ? method.getName().substring(3, 4).toLowerCase() + method.getName().substring(4) : "";
    }
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

到这里为止，基本上就把ExtensionLoader 的 getAdaptiveExtension() 方法是如何动态生成扩展接口对应的适配器类以及如何加载扩展接口的实现类的 Class 对象就说完了。下面需要再看看 getExtension() 方法，这个方法是在 org.apache.dubbo.common.extension.ExtensionLoader#getDefaultExtension 方法中，这个方法的作用就是根据扩展实现类的名称找到对应的实现类的，具体代码如下：

/**
     * Find the extension with the given name. If the specified name is not found, then {@link IllegalStateException}
     * will be thrown.
     */
    @SuppressWarnings("unchecked")
    public T getExtension(String name) {
    	// 扩展名称合法性校验
        if (StringUtils.isEmpty(name)) {
            throw new IllegalArgumentException("Extension name == null");
        }
        // 如果为true则加载默认扩展
        if ("true".equals(name)) {
            return getDefaultExtension();
        }
        // 根据name获取实例
        Holder<Object> holder = getOrCreateHolder(name);
        Object instance = holder.get();
        if (instance == null) {
            synchronized (holder) {
                instance = holder.get();
                if (instance == null) {
                	// 不存在则创建
                    instance = createExtension(name);
                    holder.set(instance);
                }
            }
        }
        return (T) instance;
    }
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

其中 createExtension 方法的代码如下：

 @SuppressWarnings("unchecked")
    private T createExtension(String name) {
    	// 根据 name 查找对应的扩展实现的 Class 对象
        Class<?> clazz = getExtensionClasses().get(name);
        if (clazz == null) {
            throw findException(name);
        }
        // 如果缓存里不存在实例，则使用Class创建实例
        try {
            T instance = (T) EXTENSION_INSTANCES.get(clazz);
            if (instance == null) {
                EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance());
                instance = (T) EXTENSION_INSTANCES.get(clazz);
            }
            // IOC依赖注入
            injectExtension(instance);
            // Wrapper对扩展实现进行功能增强
            Set<Class<?>> wrapperClasses = cachedWrapperClasses;
            if (CollectionUtils.isNotEmpty(wrapperClasses)) {
                for (Class<?> wrapperClass : wrapperClasses) {
                    instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));
                }
            }
            return instance;
        } catch (Throwable t) {
            throw new IllegalStateException("Extension instance (name: " + name + ", class: " +
                    type + ") couldn't be instantiated: " + t.getMessage(), t);
        }
    }
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

第三章 远程服务发布与引用流程剖析

3.1 Dubbo 服务发布端启动流程剖析

如果要发布一个服务，需要调用 export 方法来发布服务。export的核心代码如下所示：

public synchronized void export() {
        checkAndUpdateSubConfigs();

        // 是否需要导出服务
        if (!shouldExport()) {
            return;
        }

        // 这里是延迟发布
        if (shouldDelay()) {
            delayExportExecutor.schedule(this::doExport, delay, TimeUnit.MILLISECONDS);
        } else {
            // 直接发布
            doExport();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

其中 delayExportExecutor 的定义如下：

/**
     * A delayed exposure service timer
     */
    private static final ScheduledExecutorService delayExportExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("DubboServiceDelayExporter", true));

1
2
3
4
5

所以通过上面可以得知，dubbo的延迟发布时通过 ScheduledExecutorService 来发布的，延迟时间可以看到第二个参数是delay，这个就是延迟时间，时间的单位毫秒，那么如何设置这个delay的参数呢？需要通过：

serviceConfig.setDelay(1);
1

来设置延迟时间。可以看到 shouldDelay() 方法被用来判断是否需要进行延迟发布，其代码如下：

    private boolean shouldDelay() {
        Integer delay = getDelay();
        if (delay == null && provider != null) {
            delay = provider.getDelay();
        }
        return delay != null && delay > 0;
    }
1
2
3
4
5
6
7

如果经过判断，需要进行延迟发布，则不会发布，否则就会直接进行发布操作。
进入 doExport 方法内部，如下：

    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();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

可以看到基本上除了 doExportUrls 方法外，都是一些检查，在此不做解释。这里主要看 doExportUrls 方法的内部。
上面时序图中可以看到第四步，是 loadRegistries 方法，在doExportUrls方法内部也可以看到，如下：

    private void doExportUrls() {
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            String pathKey = URL.buildKey(getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), group, version);
            ProviderModel providerModel = new ProviderModel(pathKey, ref, interfaceClass);
            ApplicationModel.initProviderModel(pathKey, providerModel);
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }
1
2
3
4
5
6
7
8
9

这个方法的第一行就是 loadRegistries 方法。 loadRegistries 方法会加载所有的服务注册中心对象，如下图所示：

这是我自己本地起了一个zk服务，可以看到，注册中心就是我本地的zk，当然，这个服务也可以注册到很多个不同的注册中心中。
可以看到上面的时序图，下一步是 doExportUrlsFor1Protocol 方法，这个方法的代码比较多，这里只摘取比较关键的部分。

 private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        // 1) 解析 MethodConfig配置
        // 省略...

		// 2) 如果为泛型，设置泛型类型
        if (ProtocolUtils.isGeneric(generic)) {
            map.put(Constants.GENERIC_KEY, generic);
            map.put(Constants.METHODS_KEY, Constants.ANY_VALUE);
        } else {
            // 3) 正常调用设置拼接 URL 参数
            // 省略...
        }
        // 4) 拼接 URL 对象
        if (!ConfigUtils.isEmpty(token)) {
            if (ConfigUtils.isDefault(token)) {
                map.put(Constants.TOKEN_KEY, UUID.randomUUID().toString());
            } else {
                map.put(Constants.TOKEN_KEY, token);
            }
        }
        // export service
        String host = this.findConfigedHosts(protocolConfig, registryURLs, map);
        Integer port = this.findConfigedPorts(protocolConfig, name, map);
        URL url = new URL(name, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), map);

        if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                .hasExtension(url.getProtocol())) {
            url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                    .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
        }

		// 5) 导出服务，本地服务，远程服务
        String scope = url.getParameter(Constants.SCOPE_KEY);
        // don't export when none is configured
        // 如果scope为SCOPE_NONE则不导出服务
        if (!Constants.SCOPE_NONE.equalsIgnoreCase(scope)) {

            // export to local if the config is not remote (export to remote only when config is remote)
            // 5.1) 如果scope不是SCOPE_REMOTE，则导出本地服务
            if (!Constants.SCOPE_REMOTE.equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            // export to remote if the config is not local (export to local only when config is local)
            // 5.2) 如果scope不是SCOPE_LOCAL，则导出远程服务
            if (!Constants.SCOPE_LOCAL.equalsIgnoreCase(scope)) {
                // 省略...
                // 5.2.1) 如果有服务注册中心地址
                if (CollectionUtils.isNotEmpty(registryURLs)) {
                    for (URL registryURL : registryURLs) {
                        // 省略...

                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                        Exporter<?> exporter = protocol.export(wrapperInvoker);
                        exporters.add(exporter);
                    }
                } else {
                	// 5.2.2) 直连方式
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                    Exporter<?> exporter = protocol.export(wrapperInvoker);
                    exporters.add(exporter);
                }
                /**
                 * @since 2.7.0
                 * ServiceData Store
                 */
                 // 5.2.3) 元数据存储
                MetadataReportService metadataReportService = null;
                if ((metadataReportService = getMetadataReportService()) != null) {
                    metadataReportService.publishProvider(url);
                }
            }
        }
        this.urls.add(url);
    }
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
71
72
73
74
75
76
77
78

其大致的过程就是先拼接成一个 URL对象，把参数都放到这个 URL 对象里面，然后再导出。
代码1解析 MethodConfig 对象设置的方法级别的配置并保存到参数map中；代码2用来判断调用类型，如果为泛型调用，则设置泛型类型（true、nativejava或bean方式）；代码5用来导出服务，Dubbo服务导出分本地导出与远程导出，本地导出使用了injvm协议，是一个伪协议，它不开启端口，不发起远程调用，只在 JVM 内直接关联，但执行 Dubbo 的 Filter 链；在默认情况下，Dubbo 同时支持本地导出与远程导出协议，可以通过ServiceConfig 的 setScope() 方法设置，其中配置为 none 表示不导出服务，为 remote 表示只导出远程服务，为 local 表示只导出本地服务。