Dubbo-服务消费者启动流程

参考：https://blog.csdn.net/prestigeding/article/details/80621535

Dubbo官网写的非常详细：https://dubbo.apache.org/zh/docs/v2.7/dev/source/refer-service/

设计图

ReferenceBean类结构图

　　ReferenceBean实现了InitializingBean接口，ReferenceBean#afterPropertiesSet方法

public void afterPropertiesSet() throws Exception {

	// Initializes Dubbo's Config Beans before @Reference bean autowiring
	prepareDubboConfigBeans();

	// lazy init by default. 默认懒汉
	if (init == null) {
		init = false;
	}

	// eager init if necessary. 可通过配置 <dubbo:reference> 的 init 属性开启饿汉模式
	if (shouldInit()) {
		getObject();
	}
}

　　ReferenceBean实现了FactoryBean接口，所以Spring的Application获得Bean时会调用ReferenceBean#getObject方法

/**
 * 懒汉式，在 ReferenceBean 对应的服务被注入到其他类中时引用服务
 */
@Override
public Object getObject() {
	return get();
}

public synchronized T get() {
	if (destroyed) {
		throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
	}
	// 检测 ref 是否为空，为空则通过 init 方法创建
	if (ref == null) {
		// init 方法主要用于处理配置，以及调用 createProxy 生成代理类
		init();
	}
	return ref;
}

　　ReferenceConfig#init源码分析

public synchronized void init() {
	// 避免重复初始化
	if (initialized) {
		return;
	}

	if (bootstrap == null) {
		bootstrap = DubboBootstrap.getInstance();
		bootstrap.init();
	}

	// -------------------------------✨ 分割线1 ✨------------------------------

	/**
	 * 1.检测 ConsumerConfig 实例是否存在，如不存在则创建一个新的实例，然后通过系统变量或 dubbo.properties 配置文件填充 ConsumerConfig 的字段
	 * 2.检测泛化配置，并根据配置设置 interfaceClass 的值
	 * 3.从系统属性或配置文件中加载与接口名相对应的配置，并将解析结果赋值给 url 字段。url 字段的作用一般是用于点对点调用
	 * 4.检测几个核心配置类是否为空，为空则尝试从其他配置类中获取
	 */
	checkAndUpdateSubConfigs();

	// -------------------------------✨ 分割线2 ✨------------------------------

	// 检测本地存根配置合法性
	checkStubAndLocal(interfaceClass);
	ConfigValidationUtils.checkMock(interfaceClass, this);

	// -------------------------------✨ 分割线3 ✨------------------------------

	// 添加 side、协议版本信息、时间戳和进程号等信息到 map 中
	Map<String, String> map = new HashMap<String, String>();
	map.put(SIDE_KEY, CONSUMER_SIDE);

	ReferenceConfigBase.appendRuntimeParameters(map);
	// 非泛化服务
	if (!ProtocolUtils.isGeneric(generic)) {
		// 获取版本
		String revision = Version.getVersion(interfaceClass, version);
		if (revision != null && revision.length() > 0) {
			map.put(REVISION_KEY, revision);
		}

		// 获取接口方法列表，并添加到 map 中
		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)), COMMA_SEPARATOR));
		}
	}
	map.put(INTERFACE_KEY, interfaceName);
	// 将 ApplicationConfig、ConsumerConfig、ReferenceConfig 等对象的字段信息添加到 map 中
	AbstractConfig.appendParameters(map, getMetrics());
	AbstractConfig.appendParameters(map, getApplication());
	AbstractConfig.appendParameters(map, getModule());
	// remove 'default.' prefix for configs from ConsumerConfig
	// appendParameters(map, consumer, Constants.DEFAULT_KEY);
	AbstractConfig.appendParameters(map, consumer);
	AbstractConfig.appendParameters(map, this);
	MetadataReportConfig metadataReportConfig = getMetadataReportConfig();
	if (metadataReportConfig != null && metadataReportConfig.isValid()) {
		map.putIfAbsent(METADATA_KEY, REMOTE_METADATA_STORAGE_TYPE);
	}

	// -------------------------------✨ 分割线4 ✨------------------------------

	Map<String, AsyncMethodInfo> attributes = null;
	if (CollectionUtils.isNotEmpty(getMethods())) {
		attributes = new HashMap<>();
		// 遍历 MethodConfig 列表
		for (MethodConfig methodConfig : getMethods()) {
			AbstractConfig.appendParameters(map, methodConfig, methodConfig.getName());
			String retryKey = methodConfig.getName() + ".retry";
			// 检测 map 是否包含 methodName.retry
			if (map.containsKey(retryKey)) {
				String retryValue = map.remove(retryKey);
				if ("false".equals(retryValue)) {
					// 添加重试次数配置 methodName.retries
					map.put(methodConfig.getName() + ".retries", "0");
				}
			}
			// 事件通知配置：onreturn、onthrow、oninvoke等
			AsyncMethodInfo asyncMethodInfo = AbstractConfig.convertMethodConfig2AsyncInfo(methodConfig);
			if (asyncMethodInfo != null) {
//                    consumerModel.getMethodModel(methodConfig.getName()).addAttribute(ASYNC_KEY, asyncMethodInfo);
				attributes.put(methodConfig.getName(), asyncMethodInfo);
			}
		}
	}

	// -------------------------------✨ 分割线5 ✨------------------------------

	// 获取服务消费者 ip 地址
	String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);
	if (StringUtils.isEmpty(hostToRegistry)) {
		hostToRegistry = NetUtils.getLocalHost();
	} else if (isInvalidLocalHost(hostToRegistry)) {
		throw new IllegalArgumentException("Specified invalid registry ip from property:" + DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
	}
	map.put(REGISTER_IP_KEY, hostToRegistry);

	serviceMetadata.getAttachments().putAll(map);

	// 创建代理类
	ref = createProxy(map);

	serviceMetadata.setTarget(ref);
	serviceMetadata.addAttribute(PROXY_CLASS_REF, ref);
	ConsumerModel consumerModel = repository.lookupReferredService(serviceMetadata.getServiceKey());
	consumerModel.setProxyObject(ref);
	consumerModel.init(attributes);

	initialized = true;

	// invoker 可用性检查
	checkInvokerAvailable();

	// dispatch a ReferenceConfigInitializedEvent since 2.7.4
	dispatch(new ReferenceConfigInitializedEvent(this, invoker));
}

　　ReferenceConfig#createProxy源码分析

private T createProxy(Map<String, String> map) {
	if (shouldJvmRefer(map)) { // 本地引用
		// 生成本地引用 URL，协议为 injvm
		URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
		// 调用 refer 方法构建 InjvmInvoker 实例
		invoker = REF_PROTOCOL.refer(interfaceClass, url);
		if (logger.isInfoEnabled()) {
			logger.info("Using injvm service " + interfaceClass.getName());
		}
	} else { // 远程引用
		urls.clear();
		// url 不为空，表明用户可能想进行点对点调用，绕过注册中心，消费者直连提供者
		if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
			// 当需要配置多个 url 时，可用分号进行分割，这里会进行切分
			String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
			if (us != null && us.length > 0) {
				for (String u : us) {
					URL url = URL.valueOf(u);
					if (StringUtils.isEmpty(url.getPath())) {
						// 设置接口全限定名为 url 路径
						url = url.setPath(interfaceName);
					}
					// 检测 url 协议是否为 registry，若是，表明用户想使用指定的注册中心
					if (UrlUtils.isRegistry(url)) {
						// 将 map 转换为查询字符串，并作为 refer 参数的值添加到 url 中
						urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
					} else {
						// 合并 url，移除服务提供者的一些配置（这些配置来源于用户配置的 url 属性），
						// 比如线程池相关配置。并保留服务提供者的部分配置，比如版本，group，时间戳等
						// 最后将合并后的配置设置为 url 查询字符串中。
						urls.add(ClusterUtils.mergeUrl(url, map));
					}
				}
			}
		} else { // assemble URL from register center's configuration
			// if protocols not injvm checkRegistry
			if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
				checkRegistry();
				// 加载注册中心 url
				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()));
						}
						// 添加 refer 参数到 url 中，并将 url 添加到 urls 中
						urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
					}
				}
				// 未配置注册中心，抛出异常
				if (urls.isEmpty()) {
					throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address="..." /> to your spring config.");
				}
			}
		}

		// 单个注册中心或服务提供者(服务直连，下同)
		if (urls.size() == 1) {
			// 调用 RegistryProtocol 的 refer 构建 Invoker 实例
			invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
		} else { // 多个注册中心或多个服务提供者，或者两者混合
			List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
			URL registryURL = null;
			// 获取所有的 Invoker
			for (URL url : urls) {
				// 通过 refprotocol 调用 refer 构建 Invoker，refprotocol 会在运行时
				// 根据 url 协议头加载指定的 Protocol 实例，并调用实例的 refer 方法
				invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
				if (UrlUtils.isRegistry(url)) {
					registryURL = url; // use last registry url
				}
			}
			if (registryURL != null) { // registry url is available
				// for multi-subscription scenario, use 'zone-aware' policy by default
				// 如果注册中心链接不为空，则将使用 ZoneAwareCluster
				String cluster = registryURL.getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME);
				// The invoker wrap sequence would be: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
				// 创建 StaticDirectory 实例，并由 Cluster 对多个 Invoker 进行合并
				invoker = Cluster.getCluster(cluster, false).join(new StaticDirectory(registryURL, invokers));
			} else { // not a registry url, must be direct invoke.
				String cluster = CollectionUtils.isNotEmpty(invokers)
						? (invokers.get(0).getUrl() != null ? invokers.get(0).getUrl().getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME) : Cluster.DEFAULT)
						: Cluster.DEFAULT;
				invoker = Cluster.getCluster(cluster).join(new StaticDirectory(invokers));
			}
		}
	}

	if (logger.isInfoEnabled()) {
		logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
	}
	/**
	 * @since 2.7.0
	 * ServiceData Store
	 */
	String metadata = map.get(METADATA_KEY);
	WritableMetadataService metadataService = WritableMetadataService.getExtension(metadata == null ? DEFAULT_METADATA_STORAGE_TYPE : metadata);
	if (metadataService != null) {
		URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
		metadataService.publishServiceDefinition(consumerURL);
	}
	// create service proxy 生成代理类
	return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
}