Dubbo服务消费者Consumer启动流程(五)
作者：互联网
在看完上篇的服务提供者启动流程之后，再来看消费者的启动流程就简单很多了，其大体的设计流程是差不多的。服务消费者的启动主要调用ReferenceConfig#get(), get方法跟服务提供者的export方法类似，主要关注checkAndUpdateSubConfigs和init方法。
ReferenceConfig#checkAndUpdateSubConfigs
这个checkAndUpdateSubConfigs方法这边不详细深入，也不是特别复杂，读者可以自己看每个步骤细节，这边只是大体上讲解流程。
public void checkAndUpdateSubConfigs() {
if (StringUtils.isEmpty(interfaceName)) {
throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
}
// 配置优先级 consumer > module > application
completeCompoundConfigs();
// 将外部配置中心的配置刷新到本地配置
startConfigCenter();
// get consumer's global configuration
// 设置comsumer配置
checkDefault();
this.refresh();
if (getGeneric() == null && getConsumer() != null) {
setGeneric(getConsumer().getGeneric());
}
if (ProtocolUtils.isGeneric(getGeneric())) {
interfaceClass = GenericService.class;
} else {
try {
interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
.getContextClassLoader());
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
// 检查<dubbo:method> 配置的方法是否包含在interfaceClass的方法中
checkInterfaceAndMethods(interfaceClass, methods);
}
// 处理dubbo服务消费端resolve机制，也就是说消息消费者只连服务提供者，绕过注册中心。
// 这个方法主要是获取直连服务的配置信息
resolveFile();
// 校验ReferenceBean的application是否为空,如果为空，new 一个application
checkApplication();
checkMetadataReport();
}
接下来就是int方法了，初始化ref，即获取到的实际上是一个由proxyFactory来代理invoker对象，通过其Proxy InvokerInvocationHandler的invoke方法，即调用invoker的invoker方法返回的对象。整个对象类型就是远程服务接口类型。
ReferenceConfig#init
private void init() {
if (initialized) {
return;
}
initialized = true;
// 校验stub、mock实现类与interface的兼容性
checkStubAndLocal(interfaceClass);
checkMock(interfaceClass);
Map<String, String> map = new HashMap<String, String>();
map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
/**
* 构建Map,封装服务消费者引用服务提供者URL的属性，这里主要填充side:consume（消费端)、dubbo：2.0.0(版本)、timestamp、pid:进程ID。
*/
appendRuntimeParameters(map);
// 如果不是泛化引用，增加methods:interface的所有方法名，多个用逗号隔开。
if (!isGeneric()) {
String revision = Version.getVersion(interfaceClass, version);
if (revision != null && revision.length() > 0) {
map.put("revision", revision);
}
// 使用Wrapper 目的是为了缓存interfaceClass的类元数据
String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
if (methods.length == 0) {
logger.warn("No method found in service interface " + interfaceClass.getName());
map.put("methods", Constants.ANY_VALUE);
} else {
map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
}
}
// 用Map存储application配置、module配置、默认消费者参数(ConsumerConfig)、服务消费者dubbo:reference的属性。
map.put(Constants.INTERFACE_KEY, interfaceName);
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, this);
Map<String, Object> attributes = null;
/**
* 获取服务键值 {group}/interface:版本，如果group为空，则为interface:版本,其值存为prifex，
* 然后将dubbo:method的属性名称也填入map中，键前缀为dubbo.method.methodname.属性名。
* dubbo:method的子标签dubbo:argument标签的属性也追加到attributes map中，键为 prifex + methodname.属性名。
*/
if (CollectionUtils.isNotEmpty(methods)) {
attributes = new HashMap<String, Object>();
for (MethodConfig methodConfig : methods) {
appendParameters(map, methodConfig, methodConfig.getName());
String retryKey = methodConfig.getName() + ".retry";
if (map.containsKey(retryKey)) {
String retryValue = map.remove(retryKey);
if ("false".equals(retryValue)) {
map.put(methodConfig.getName() + ".retries", "0");
}
}
attributes.put(methodConfig.getName(), convertMethodConfig2AyncInfo(methodConfig));
}
}
//填充register.ip属性，该属性是消息消费者连接注册中心的IP，并不是注册中心自身的IP。
String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
if (StringUtils.isEmpty(hostToRegistry)) {
hostToRegistry = NetUtils.getLocalHost();
}
map.put(Constants.REGISTER_IP_KEY, hostToRegistry);
ref = createProxy(map);
// 获取服务键值 {group}/interface:版本
String serviceKey = URL.buildKey(interfaceName, group, version);
// 将消息消费者缓存在ApplicationModel中。
ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes));
}
init的作用还是根据配置构建消费者map中的查询参数URL，关键在粗体部分，ref = createProxy(map);本章重点也是在createProxy方法上详解。
ReferenceConfig#createProxy
createProxy 首先判断是否是injvm协议的本地引用，还是revoke机制的直连模式，还是普通消费者，从注册中心获取订阅服务，目的就是为了初始化引用服务的urls,其中url的subscribe设置为true。然后根据url来获取对应协议的invoker，最后根据invoker来代理远程服务。
判断是否是injvm服务？通过shouldJvmRefer方法，第一是根据配置，如果没有配置，发现是直连服务，那么久不是injvm。或者如果scope=local或injvm=true，那么isJvmRefer=true
如何判断是否是直连模式呢？根据前面的resolveFile，如果是直连，那么就会初始化这个点对点的url。
剩下的就是普通消费者模式了，依赖注册中心。
private T createProxy(Map<String, String> map) {
// 判断该消费者是否是引用本(JVM)内提供的服务。
if (shouldJvmRefer(map)) {
/**
* 如果消费者引用本地JVM中的服务，则利用InjvmProtocol创建Invoker，dubbo中的invoker主要负责服务调用的功能
* injvm://127.0.0.1/org.apache.dubbo.demo.DemoService?application=dubbo-demo-api-consumer
* &default.lazy=false&default.sticky=false&dubbo=2.0.2&interface=org.apache.dubbo.demo.DemoService&lazy=false&methods=sayHello
* &pid=36176®ister.ip=192.168.0.102&release=&side=consumer&sticky=false×tamp=1640070027986
*/
URL url = new URL(Constants.LOCAL_PROTOCOL, Constants.LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
invoker = refprotocol.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
} else { // 处于直连模式
/**
* 对直连URL进行分割，多个直连URL用分号隔开，如果URL中不包含path属性，则为URL设置path属性为interfaceName。
* 如果直连提供者的协议为registry，则对url增加refer属性，其值为消息消费者所有的属性。(表示从注册中心发现服务提供者)
* 如果是其他协议提供者，则合并服务提供者与消息消费者的属性，并移除服务提供者默认属性。以default开头的属性。
*/
if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
String[] us = Constants.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.setPath(interfaceName);
}
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else {
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else { // assemble URL from register center's configuration 普通消息消费者，从注册中心订阅服务
/**
* ：获取所有注册中心URL，其中参数false表示消费端，需要排除dubbo:registry subscribe=false的注册中心，其值为false表示不接受订阅。
* 根据注册中心URL，构建监控中心URL。
* 如果监控中心不为空，在注册中心URL后增加属性monitor。
* 在注册中心URL中，追加属性refer，其值为消费端的所有配置组成的URL。
*/
checkRegistry();
List<URL> us = loadRegistries(false);
if (CollectionUtils.isNotEmpty(us)) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if (monitorUrl != null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
urls.add(u.addParameterAndEncoded(Constants.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) {
// 根据URL获取对应协议的Invoker。
//如果只有一个服务提供者URL,则直接根据协议构建Invoker，具体有如下协议：
// Protocal$Adaptive.refer -> RegisterProtocal.refer
invoker = refprotocol.refer(interfaceClass, urls.get(0));
} else {
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(refprotocol.refer(interfaceClass, url));
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // use last registry url
}
}
if (registryURL != null) { // registry url is available
// use RegistryAwareCluster only when register's cluster is available
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, RegistryAwareCluster.NAME);
// The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
invoker = cluster.join(new StaticDirectory(u, invokers));
} else { // not a registry url, must be direct invoke.
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
// 如果dubbo:referecnce的check=true或默认为空，则需要判断服务提供者是否存在。
if (shouldCheck() && !invoker.isAvailable()) {
// make it possible for consumer to retry later if provider is temporarily unavailable
initialized = false;
throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
/**
* @since 2.7.0
* ServiceData Store
*/
MetadataReportService metadataReportService = null;
if ((metadataReportService = getMetadataReportService()) != null) {
URL consumerURL = new URL(Constants.CONSUMER_PROTOCOL, map.remove(Constants.REGISTER_IP_KEY), 0, map.get(Constants.INTERFACE_KEY), map);
metadataReportService.publishConsumer(consumerURL);
}
// create service proxy
/**
* 根据invoker获取代理类，其实现逻辑如下：
* 从消费者URL中获取interfaces的值，用,分隔出单个服务应用接口。
* 增加默认接口EchoService接口。
* 根据需要实现的接口，使用jdk或Javassist创建代理类。
*/
return (T) proxyFactory.getProxy(invoker);
}
这里面的checkRegister跟服务者类似。 这里边就不展开了。urls都是register://开头，其中参数register对应具体的注册类型，比如向zookeeper或者redis等。
根据urls的size，我们就可以初始化invoker了
invoker = refprotocol.refer(interfaceClass, url);
如果多个url，则会产生多个invoker，最后通过cluser的join来返回一个invoker对象 invoker = cluster.join(new StaticDirectory(invokers));
最后通过Javassist来代理。具体细节将会在下一章消费者调用流程中介绍。
标签：Dubbo,map,URL,流程,url,invoker,interfaceClass,Consumer,Constants 来源： https://www.cnblogs.com/gaojy/p/15709137.html