Skip to content

42加餐:服务引用流程全解析

Dubbo 作为一个 RPC 框架,暴露给用户最基本的功能就是服务发布和服务引用。在上一课时,我们已经分析了服务发布的核心流程。那么在本课时,我们就接着深入分析服务引用的核心流程

Dubbo 支持两种方式引用远程的服务:

  • 服务直连的方式 ,仅适合在调试服务的时候使用;

  • 基于注册中心引用服务 ,这是生产环境中使用的服务引用方式。

DubboBootstrap 入口

在上一课时介绍服务发布的时候,我们介绍了 DubboBootstrap.start() 方法的核心流程,其中除了会调用 exportServices() 方法完成服务发布 之外,还会调用 referServices() 方法完成服务引用 ,这里就不再贴出 DubboBootstrap.start() 方法的具体代码,你若感兴趣的话可以参考源码进行学习。

在 DubboBootstrap.referServices() 方法中,会从 ConfigManager 中获取所有 ReferenceConfig 列表,并根据 ReferenceConfig 获取对应的代理对象,入口逻辑如下:

java
private void referServices() {
    if (cache == null) { // 初始ReferenceConfigCache
        cache = ReferenceConfigCache.getCache();
    }
    configManager.getReferences().forEach(rc -> {
        // 遍历ReferenceConfig列表
        ReferenceConfig referenceConfig = (ReferenceConfig) rc;
        referenceConfig.setBootstrap(this);
        if (rc.shouldInit()) { // 检测ReferenceConfig是否已经初始化
            if (referAsync) { // 异步
                CompletableFuture<Object> future = ScheduledCompletableFuture.submit(
                        executorRepository.getServiceExporterExecutor(),
                        () -> cache.get(rc)
                );
                asyncReferringFutures.add(future);
            } else { // 同步
                cache.get(rc);
            }
        }
    });
}

这里的 ReferenceConfig 是哪里来的呢?在第 01 课时dubbo-demo-api-consumer 示例中,我们可以看到构造 ReferenceConfig 对象的逻辑,这些新建的 ReferenceConfig 对象会通过 DubboBootstrap.reference() 方法添加到 ConfigManager 中进行管理,如下所示:

java
public DubboBootstrap reference(ReferenceConfig<?> referenceConfig) {
    configManager.addReference(referenceConfig);
    return this;
}

ReferenceConfigCache

服务引用的核心实现在 ReferenceConfig 之中,一个 ReferenceConfig 对象对应一个服务接口,每个 ReferenceConfig 对象中都封装了与注册中心的网络连接,以及与 Provider 的网络连接,这是一个非常重要的对象。

为了避免底层连接泄漏造成性能问题,从 Dubbo 2.4.0 版本开始,Dubbo 提供了 ReferenceConfigCache 用于缓存 ReferenceConfig 实例。

在 dubbo-demo-api-consumer 示例中,我们可以看到 ReferenceConfigCache 的基本使用方式:

java
ReferenceConfig<DemoService> reference = new ReferenceConfig<>();
reference.setInterface(DemoService.class);
... 
// 这一步在DubboBootstrap.start()方法中完成
ReferenceConfigCache cache = ReferenceConfigCache.getCache();
...
DemoService demoService = ReferenceConfigCache.getCache().get(reference);

在 ReferenceConfigCache 中维护了一个静态的 Map(CACHE_HOLDER)字段,其中 Key 是由 Group、服务接口和 version 构成,Value 是一个 ReferenceConfigCache 对象。在 ReferenceConfigCache 中可以传入一个 KeyGenerator 用来修改缓存 Key 的生成逻辑,KeyGenerator 接口的定义如下:

java
public interface KeyGenerator {
    String generateKey(ReferenceConfigBase<?> referenceConfig);
}

默认的 KeyGenerator 实现是 ReferenceConfigCache 中的匿名内部类,其对象由 DEFAULT_KEY_GENERATOR 这个静态字段引用,具体实现如下:

java
public static final KeyGenerator DEFAULT_KEY_GENERATOR = referenceConfig -> {
    String iName = referenceConfig.getInterface();
    if (StringUtils.isBlank(iName)) { // 获取服务接口名称
        Class<?> clazz = referenceConfig.getInterfaceClass();
        iName = clazz.getName();
    }
    if (StringUtils.isBlank(iName)) {
        throw new IllegalArgumentException("No interface info in ReferenceConfig" + referenceConfig);
    }
    // Key的格式是group/interface:version
    StringBuilder ret = new StringBuilder();
    if (!StringUtils.isBlank(referenceConfig.getGroup())) {
        ret.append(referenceConfig.getGroup()).append("/");
    }
    ret.append(iName);
    if (!StringUtils.isBlank(referenceConfig.getVersion())) {
        ret.append(":").append(referenceConfig.getVersion());
    }
    return ret.toString();
};

在 ReferenceConfigCache 实例对象中,会维护下面两个 Map 集合。

  • proxies(ConcurrentMap<Class<?>, ConcurrentMap<String, Object>>类型):该集合用来存储服务接口的全部代理对象,其中第一层 Key 是服务接口的类型,第二层 Key 是上面介绍的 KeyGenerator 为不同服务提供方生成的 Key,Value 是服务的代理对象。

  • referredReferences(ConcurrentMap<String, ReferenceConfigBase<?>> 类型):该集合用来存储已经被处理的 ReferenceConfig 对象。

我们回到 DubboBootstrap.referServices() 方法中,看一下其中与 ReferenceConfigCache 相关的逻辑。

首先是 ReferenceConfigCache.getCache() 这个静态方法,会在 CACHE_HOLDER 集合中添加一个 Key 为"DEFAULT"的 ReferenceConfigCache 对象(使用默认的 KeyGenerator 实现),它将作为默认的 ReferenceConfigCache 对象。

接下来,无论是同步服务引用还是异步服务引用,都会调用 ReferenceConfigCache.get() 方法,创建并缓存代理对象。下面就是 ReferenceConfigCache.get() 方法的核心实现:

java
public <T> T get(ReferenceConfigBase<T> referenceConfig) {
    // 生成服务提供方对应的Key
    String key = generator.generateKey(referenceConfig);
    // 获取接口类型
    Class<?> type = referenceConfig.getInterfaceClass();
    // 获取该接口对应代理对象集合
    proxies.computeIfAbsent(type, _t -> new ConcurrentHashMap<>());
    ConcurrentMap<String, Object> proxiesOfType = proxies.get(type);
    // 根据Key获取服务提供方对应的代理对象
    proxiesOfType.computeIfAbsent(key, _k -> {
        // 服务引用
        Object proxy = referenceConfig.get();
        // 将ReferenceConfig记录到referredReferences集合
        referredReferences.put(key, referenceConfig);            
        return proxy;
    });
    return (T) proxiesOfType.get(key);
}

ReferenceConfig

通过前面的介绍我们知道,ReferenceConfig 是服务引用的真正入口,其中会创建相关的代理对象。下面先来看 ReferenceConfig.get() 方法:

java
public synchronized T get() {
    if (destroyed) { // 检测当前ReferenceConfig状态
        throw new IllegalStateException("...");
    }
    if (ref == null) {// ref指向了服务的代理对象
        init(); // 初始化ref字段
    }
    return ref;
}

在 ReferenceConfig.init() 方法中,首先会对服务引用的配置进行处理,以保证配置的正确性。这里的具体实现其实本身并不复杂,但由于涉及很多的配置解析和处理逻辑,代码就显得非常长,我们就不再一一展示,你若感兴趣的话可以参考源码进行学习。

ReferenceConfig.init() 方法的核心逻辑是调用 createProxy() 方法,调用之前会从配置中获取 createProxy() 方法需要的参数:

java
public synchronized void init() {
    if (initialized) { // 检测ReferenceConfig的初始化状态
        return;
    }
    if (bootstrap == null) { // 检测DubboBootstrap的初始化状态
        bootstrap = DubboBootstrap.getInstance();
        bootstrap.init();
    }
    ... // 省略其他配置的检查
    Map<String, String> map = new HashMap<String, String>();
    map.put(SIDE_KEY, CONSUMER_SIDE); // 添加side参数
    // 添加Dubbo版本、release参数、timestamp参数、pid参数
    ReferenceConfigBase.appendRuntimeParameters(map);
    // 添加interface参数
    map.put(INTERFACE_KEY, interfaceName);
    ... // 省略其他参数的处理
    String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);
    if (StringUtils.isEmpty(hostToRegistry)) {
        hostToRegistry = NetUtils.getLocalHost();
    } else if (isInvalidLocalHost(hostToRegistry)) {
        throw new IllegalArgumentException("...");
    }
    // 添加ip参数
    map.put(REGISTER_IP_KEY, hostToRegistry);
    // 调用createProxy()方法
    ref = createProxy(map);
    ...// 省略其他代码
    initialized = true;
    // 触发ReferenceConfigInitializedEvent事件
    dispatch(new ReferenceConfigInitializedEvent(this, invoker));
}

ReferenceConfig.createProxy() 方法中处理了多种服务引用的场景,例如,直连单个/多个Provider、单个/多个注册中心。下面是 createProxy() 方法的核心流程,大致可以梳理出这么 5 个步骤。

  1. 根据传入的参数集合判断协议是否为 injvm 协议,如果是,直接通过 InjvmProtocol 引用服务。

  2. 构造 urls 集合。Dubbo 支持直连 Provider依赖注册中心两种服务引用方式。如果是直连服务的模式,我们可以通过 url 参数指定一个或者多个 Provider 地址,会被解析并填充到 urls 集合;如果通过注册中心的方式进行服务引用,则会调用 AbstractInterfaceConfig.loadRegistries() 方法加载所有注册中心。

  3. 如果 urls 集合中只记录了一个 URL,通过 Protocol 适配器选择合适的 Protocol 扩展实现创建 Invoker 对象。如果是直连 Provider 的场景,则 URL 为 dubbo 协议,这里就会使用 DubboProtocol 这个实现;如果依赖注册中心,则使用 RegistryProtocol 这个实现。

  4. 如果 urls 集合中有多个注册中心,则使用 ZoneAwareCluster 作为 Cluster 的默认实现,生成对应的 Invoker 对象;如果 urls 集合中记录的是多个直连服务的地址,则使用 Cluster 适配器选择合适的扩展实现生成 Invoker 对象。

  5. 通过 ProxyFactory 适配器选择合适的 ProxyFactory 扩展实现,将 Invoker 包装成服务接口的代理对象。

通过上面的流程我们可以看出createProxy() 方法中有两个核心:一是通过 Protocol 适配器选择合适的 Protocol 扩展实现创建 Invoker 对象;二是通过 ProxyFactory 适配器选择合适的 ProxyFactory 创建代理对象。

下面我们来看 createProxy() 方法的具体实现:

java
private T createProxy(Map<String, String> map) {
    if (shouldJvmRefer(map)) { // 根据url的协议、scope以及injvm等参数检测是否需要本地引用
        // 创建injvm协议的URL
        URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
        // 通过Protocol的适配器选择对应的Protocol实现创建Invoker对象
        invoker = REF_PROTOCOL.refer(interfaceClass, url);
        if (logger.isInfoEnabled()) {
            logger.info("Using injvm service " + interfaceClass.getName());
        }
    } else {
        urls.clear();
        if (url != null && url.length() > 0) {
            String[] us = SEMICOLON_SPLIT_PATTERN.split(url); // 配置多个URL的时候,会用分号进行切分
            if (us != null && us.length > 0) { // url不为空,表明用户可能想进行点对点调用
                for (String u : us) {
                    URL url = URL.valueOf(u);
                    if (StringUtils.isEmpty(url.getPath())) {
                        url = url.setPath(interfaceName);  // 设置接口完全限定名为URL Path
                    }
                    if (UrlUtils.isRegistry(url)) { // 检测URL协议是否为registry,若是,说明用户想使用指定的注册中心
                        // 这里会将map中的参数整理成一个参数添加到refer参数中
                        urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                    } else {
                        // 将map中的参数添加到url中
                        urls.add(ClusterUtils.mergeUrl(url, map));
                    }
                }
            }
        } else { 
            if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
                checkRegistry();
                // 加载注册中心的地址RegistryURL
                List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
                if (CollectionUtils.isNotEmpty(us)) {
                    for (URL u : us) {
                        URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
                        if (monitorUrl != null) {
                            map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                        }
                        // 将map中的参数整理成refer参数,添加到RegistryURL中
                        urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                    }
                }
                if (urls.isEmpty()) { // 既不是服务直连,也没有配置注册中心,抛出异常
                    throw new IllegalStateException("...");
                }
            }
        }
        if (urls.size() == 1) {
            // 在单注册中心或是直连单个服务提供方的时候,通过Protocol的适配器选择对应的Protocol实现创建Invoker对象
            invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
        } else {
            // 多注册中心或是直连多个服务提供方的时候,会根据每个URL创建Invoker对象
            List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null;
            for (URL url : urls) {
                invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                if (UrlUtils.isRegistry(url)) { // 确定是多注册中心,还是直连多个Provider
                    registryURL = url;
                }
            }
            if (registryURL != null) {
                // 多注册中心的场景中,会使用ZoneAwareCluster作为Cluster默认实现,多注册中心之间的选择
                URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
                invoker = CLUSTER.join(new StaticDirectory(u, invokers));
            } else {
                // 多个Provider直连的场景中,使用Cluster适配器选择合适的扩展实现
                invoker = CLUSTER.join(new StaticDirectory(invokers));
            }
        }
    }
    if (shouldCheck() && !invoker.isAvailable()) {
        // 根据check配置决定是否检测Provider的可用性
        invoker.destroy();
        throw new IllegalStateException("...");
    }
    ...// 元数据处理相关的逻辑
    // 通过ProxyFactory适配器选择合适的ProxyFactory扩展实现,创建代理对象
    return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
}

RegistryProtocol

在直连 Provider 的场景中,会使用 DubboProtocol.refer() 方法完成服务引用,DubboProtocol.refer() 方法的具体实现在前面第 25 课时中已经详细介绍过了,这里我们重点来看存在注册中心的场景中,Dubbo Consumer 是如何通过 RegistryProtocol 完成服务引用的。

在 RegistryProtocol.refer() 方法中,会先根据 URL 获取注册中心的 URL,再调用 doRefer 方法生成 Invoker,在 refer() 方法中会使用 MergeableCluster 处理多 group 引用的场景。

java
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    url = getRegistryUrl(url); // 从URL中获取注册中心的URL
    // 获取Registry实例,这里的RegistryFactory对象是通过Dubbo SPI的自动装载机制注入的
    Registry registry = registryFactory.getRegistry(url);
    if (RegistryService.class.equals(type)) {
        return proxyFactory.getInvoker((T) registry, type, url);
    }
    // 从注册中心URL的refer参数中获取此次服务引用的一些参数,其中就包括group
    Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
    String group = qs.get(GROUP_KEY);
    if (group != null && group.length() > 0) {
        if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
            // 如果此次可以引用多个group的服务,则Cluser实现使用MergeableCluster实现,
            // 这里的getMergeableCluster()方法就会通过Dubbo SPI方式找到MergeableCluster实例
            return doRefer(getMergeableCluster(), registry, type, url);
        }
    }
    // 如果没有group参数或是只指定了一个group,则通过Cluster适配器选择Cluster实现
    return doRefer(cluster, registry, type, url);
}

在 doRefer() 方法中,首先会根据 URL 初始化 RegistryDirectory 实例,然后生成 Subscribe URL 并进行注册,之后会通过 Registry 订阅服务,最后通过 Cluster 将多个 Invoker 合并成一个 Invoker 返回给上层,具体实现如下:

java
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    // 创建RegistryDirectory实例
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    // 生成SubscribeUrl,协议为consumer,具体的参数是RegistryURL中refer参数指定的参数
    Map<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());
    URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    if (directory.isShouldRegister()) {
        directory.setRegisteredConsumerUrl(subscribeUrl); // 在SubscribeUrl中添加category=consumers和check=false参数
        registry.register(directory.getRegisteredConsumerUrl()); // 服务注册,在Zookeeper的consumers节点下,添加该Consumer对应的节点
    }
    directory.buildRouterChain(subscribeUrl); // 根据SubscribeUrl创建服务路由
    // 订阅服务,toSubscribeUrl()方法会将SubscribeUrl中category参数修改为"providers,configurators,routers"
    // RegistryDirectory的subscribe()在前面详细分析过了,其中会通过Registry订阅服务,同时还会添加相应的监听器
    directory.subscribe(toSubscribeUrl(subscribeUrl));
    // 注册中心中可能包含多个Provider,相应地,也就有多个Invoker,
    // 这里通过前面选择的Cluster将多个Invoker对象封装成一个Invoker对象
    Invoker<T> invoker = cluster.join(directory);
    // 根据URL中的registry.protocol.listener参数加载相应的监听器实现
    List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
    if (CollectionUtils.isEmpty(listeners)) {
        return invoker;
    }
    // 为了方便在监听器中回调,这里将此次引用使用到的Directory对象、Cluster对象、Invoker对象以及SubscribeUrl
    // 封装到一个RegistryInvokerWrapper中,传递给监听器
    RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker, subscribeUrl);
    for (RegistryProtocolListener listener : listeners) {
        listener.onRefer(this, registryInvokerWrapper);
    }
    return registryInvokerWrapper;
}

这里涉及的 RegistryDirectory、Router 接口、Cluster 接口及其相关的扩展实现,我们都已经在前面的课时详细分析过了,这里不再重复。

总结

本课时,我们重点介绍了 Dubbo 服务引用的整个流程。

  • 首先,我们介绍了 DubboBootStrap 这个入口门面类与服务引用相关的方法,其中涉及 referServices()、reference() 等核心方法。

  • 接下来,我们分析了 ReferenceConfigCache 这个 ReferenceConfig 对象缓存,以及 ReferenceConfig 实现服务引用的核心流程。

  • 最后,我们还讲解了 RegistryProtocol 从注册中心引用服务的核心实现。

下一课时,我们将开始介绍 Dubbo 服务自省的相关内容,记得按时来听课。