Dubbo 作为一个 RPC 框架，暴露给用户最基本的功能就是服务发布和服务引用。在上一章，我已经分析了服务发布的核心流程。那么在本章，我就接着深入分析 服务引用的核心流程 。
Dubbo 支持两种方式引用远程的服务：
- 服务直连：仅适合在测试场景调试服务时使用；
- 基于注册中心引用：这是生产环境中使用的服务引用方式。
一、DubboBootstrap 入口
在上一章介绍服务发布的时候，我介绍了 DubboBootstrap.start() 方法的核心流程，其中除了会调用 exportServices() 方法完成服务发布之外，还会调用 referServices() 方法完成服务引用。
在 DubboBootstrap.referServices() 方法中，会从 ConfigManager 中获取所有 ReferenceConfig 列表，并根据 ReferenceConfig 获取对应的代理对象，入口逻辑如下：
// DubboBootstrap.java
private void referServices() {
// 初始化ReferenceConfigCache
if (cache == null) {
cache = ReferenceConfigCache.getCache();
}
configManager.getReferences().forEach(rc -> {
// 遍历ReferenceConfig列表
ReferenceConfig referenceConfig = (ReferenceConfig) rc;
referenceConfig.setBootstrap(this);
// 检测ReferenceConfig是否已经初始化
if (rc.shouldInit()) {
// 异步
if (referAsync) {
CompletableFuture<Object> future = ScheduledCompletableFuture.submit(
executorRepository.getServiceExporterExecutor(),
() -> cache.get(rc)
);
asyncReferringFutures.add(future);
}
// 同步
else {
cache.get(rc);
}
}
});
}
上述的 ReferenceConfig 是哪里来的呢？在前面章节讲解 dubbo-demo-api-consumer 的示例中，我们可以看到构造 ReferenceConfig 对象的逻辑，这些新建的 ReferenceConfig 对象会通过 DubboBootstrap.reference() 方法添加到 ConfigManager 中进行管理，如下所示：
// DubboBootstrap.java
public DubboBootstrap reference(ReferenceConfig<?> referenceConfig) {
configManager.addReference(referenceConfig);
return this;
}
1.1 ReferenceConfigCache
服务引用的核心实现在 ReferenceConfig 之中，一个 ReferenceConfig 对象对应一个服务接口，每个 ReferenceConfig 对象中都封装了与注册中心的网络连接，以及与 Provider 的网络连接，这是一个非常重要的对象。
为了避免底层连接泄漏造成性能问题，从 Dubbo 2.4.0 版本开始，Dubbo 提供了 ReferenceConfigCache 用于缓存 ReferenceConfig 实例。
在 dubbo-demo-api-consumer 示例中，我们可以看到 ReferenceConfigCache 的基本使用方式：
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 接口的定义如下：
// KeyGenerator.java
public interface KeyGenerator {
String generateKey(ReferenceConfigBase<?> referenceConfig);
}
默认的 KeyGenerator 实现是 ReferenceConfigCache 中的匿名内部类，其对象由 DEFAULT_KEY_GENERATOR 这个静态字段引用，具体实现如下：
// ReferenceConfigCache.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() 方法的核心实现：
// ReferenceConfigCache.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);
}
1.2 ReferenceConfig
通过前面的介绍我们知道， ReferenceConfig 是服务引用的真正入口 ，其中会创建相关的代理对象。下面先来看 ReferenceConfig.get() 方法：
// ReferenceConfig.java
public synchronized T get() {
// 检测当前ReferenceConfig状态
if (destroyed) {
throw new IllegalStateException("...");
}
// ref指向了服务的代理对象
if (ref == null) {
// 初始化ref字段
init();
}
return ref;
}
在 ReferenceConfig.init() 方法中，首先会对服务引用的配置进行处理，以保证配置的正确性。这里的具体实现其实本身并不复杂，但由于涉及很多的配置解析和处理逻辑，代码就显得非常长，我就不再一一展示，感兴趣童鞋可以参考源码进行学习。
ReferenceConfig.init() 方法的核心逻辑是调用 createProxy() 方法 ，调用之前会从配置中获取 createProxy() 方法需要的参数：
// ReferenceConfig.java
public synchronized void init() {
// 检测ReferenceConfig的初始化状态
if (initialized) {
return;
}
// 检测DubboBootstrap的初始化状态
if (bootstrap == null) {
bootstrap = DubboBootstrap.getInstance();
bootstrap.init();
}
// ...省略其他配置的检查
Map<String, String> map = new HashMap<String, String>();
// 添加side参数
map.put(SIDE_KEY, CONSUMER_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 个步骤。
- 根据传入的参数集合判断协议是否为 injvm 协议，如果是，直接通过 InjvmProtocol 引用服务；
- 构造 urls 集合。Dubbo 支持 直连 Provider 和 依赖注册中心 两种服务引用方式。如果是直连服务的模式，我们可以通过 url 参数指定一个或者多个 Provider 地址，会被解析并填充到 urls 集合；如果通过注册中心的方式进行服务引用，则会调用 AbstractInterfaceConfig.loadRegistries() 方法加载所有注册中心；
- 如果 urls 集合中只记录了一个 URL，通过 Protocol 适配器选择合适的 Protocol 扩展实现创建 Invoker 对象。如果是直连 Provider 的场景，则 URL 为 dubbo 协议，这里就会使用 DubboProtocol 这个实现；如果依赖注册中心，则使用 RegistryProtocol 这个实现；
- 如果 urls 集合中有多个注册中心，则使用 ZoneAwareCluster 作为 Cluster 的默认实现，生成对应的 Invoker 对象；如果 urls 集合中记录的是多个直连服务的地址，则使用 Cluster 适配器选择合适的扩展实现生成 Invoker 对象；
- 通过 ProxyFactory 适配器选择合适的 ProxyFactory 扩展实现，将 Invoker 包装成服务接口的代理对象；
通过上面的流程我们可以看出 createProxy() 方法中有两个核心 ：
- 通过 Protocol 适配器选择合适的 Protocol 扩展实现创建 Invoker 对象；
- 通过 ProxyFactory 适配器选择合适的 ProxyFactory 创建代理对象。
下面我们来看 createProxy() 方法的具体实现：
// ReferenceConfig.java
private T createProxy(Map<String, String> map) {
// 根据url的协议、scope以及injvm等参数检测是否需要本地引用
if (shouldJvmRefer(map)) {
// 创建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) {
// 配置多个URL的时候，会用分号进行切分
String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
// url不为空，表明用户可能想进行点对点调用
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (StringUtils.isEmpty(url.getPath())) {
// 设置接口完全限定名为URL Path
url = url.setPath(interfaceName);
}
// 检测URL协议是否为registry，若是，说明用户想使用指定的注册中心
if (UrlUtils.isRegistry(url)) {
// 这里会将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));
// 确定是多注册中心，还是直连多个Provider
if (UrlUtils.isRegistry(url)) {
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() 方法的具体实现在前面章节已经详细介绍过了，这里我们重点来看存在注册中心的场景中，Dubbo Consumer 是如何通过 RegistryProtocol 完成服务引用的。
在 RegistryProtocol.refer() 方法中，会先根据 URL 获取注册中心的 URL，再调用 doRefer 方法生成 Invoker，在 refer() 方法中会使用 MergeableCluster 处理多 group 引用的场景。
// RegistryProtocol.java
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 从URL中获取注册中心的URL
url = getRegistryUrl(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 返回给上层，具体实现如下：
// RegistryProtocol.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()) {
// 在SubscribeUrl中添加category=consumers和check=false参数
directory.setRegisteredConsumerUrl(subscribeUrl);
// 服务注册，在Zookeeper的consumers节点下，添加该Consumer对应的节点
registry.register(directory.getRegisteredConsumerUrl());
}
// 根据SubscribeUrl创建服务路由
directory.buildRouterChain(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 从注册中心引用服务的核心实现。
