dubbo源码阅读之服务目录
服务目录
服务目录对应的接口是Directory，这个接口里主要的方法是
List<Invoker<T>> list(Invocation invocation) throws RpcException;
列出所有的Invoker，对于服务消费端而言，一个Invoker对应一个可用的服务提供者，底层封装了一个tcp连接。当然Invoker也可以是嵌套的，一个Invoker内包含了多个实际的Invoker。通过Cluster对象将一个服务目录封装成一个Invoker，内部包含了故障转移，服务路由，负载均衡，等等相关的集群逻辑。
回到服务目录，主要包括两种服务目录，StaticDirectory，RegistryDirectory。
- StaticDirectory。静态服务目录，顾名思义，这个目录在创建的时候就会通过构造方法传进一个Invoker列表，在之后过程中这个列表不再变化。
- RegistryDirectory。通过监听注册中心的服务提供者信息动态更新Invoker列表的服务目录。
从上节服务引入，我们知道，不论是StaticDirectory还是RegistryDirectory，最终都会通过Cluster.join方法封装为一个Invoker。由于静态服务目录的逻辑很简单，这里不再赘述，本节我们主要分析一下注册中心的服务目录。
RegistryDirectory概述
这个类除了继承了AbstractDirectory，还实现了NotifyListener接口。NotifyListener接口是一个监听类，用于监听注册中心配置信息的变更事件。我们首先简单看一下RegistryDirectory中实现Directory接口的部分代码。
AbstractDirectory.list
list方法的实现放在抽象类AbstractDirectory中，
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
return doList(invocation);
}
wishing就是一个状态的判断。doList是一个模板方法，由子类实现。
RegistryDirectory.doList
@Override
public List<Invoker<T>> doList(Invocation invocation) {
// 当状态量forbidden为true时，服务调用被禁止
// 什么时候forbidden为true呢？？当url只有一个，且协议名称为empty时，就以为这没有服务提供者可用。
if (forbidden) {
// 1. No service provider 2. Service providers are disabled
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "No provider available from registry " +
getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " +
NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() +
", please check status of providers(disabled, not registered or in blacklist).");
}
// 服务分组
if (multiGroup) {
return this.invokers == null ? Collections.emptyList() : this.invokers;
}
List<Invoker<T>> invokers = null;
try {
// Get invokers from cache, only runtime routers will be executed.
// 从缓存中取出Invoker列表，并经由服务路由获取相应的Invoker
invokers = routerChain.route(getConsumerUrl(), invocation);
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
// FIXME Is there any need of failing back to Constants.ANY_VALUE or the first available method invokers when invokers is null?
/*Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
String methodName = RpcUtils.getMethodName(invocation);
invokers = localMethodInvokerMap.get(methodName);
if (invokers == null) {
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
if (invokers == null) {
Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
invokers = iterator.next();
}
}
}*/
return invokers == null ? Collections.emptyList() : invokers;
}
这个方法的主要逻辑是，首先判断服务是否可用（根据forbidden状态变量）。然后从路由链中取出Invoker列表。由于服务路由并不是本节的重点，所以我们只是简单第看一下RouterChain.route方法
RouterChain.route
public List<Invoker<T>> route(URL url, Invocation invocation) {
List<Invoker<T>> finalInvokers = invokers;
for (Router router : routers) {
finalInvokers = router.route(finalInvokers, url, invocation);
}
return finalInvokers;
}
一次调用路由列表中的路由规则，最终返回经过多个路由规则路由过的Invoker列表。类似于责任链模式，有点像web容器的过滤器，或者是spring-mvc中的拦截器，都是一个链式的调用。
实际上我们平时一般较少使用到路由功能，所以这里routers列表实际上是空的，这种情况下不用经过任何路由，直接原样返回Invokers列表。而至于RouterChain内部的invokers成员是哪来的，RegistryDirectory监听注册中心发生变更后刷新本地缓存中的Invokers列表，并将其注入到RouterChain对象中，我们后面会讲到。
RegistryDirectory.notify
接下来我们分析RegistryDirectory中最重要的方法，也就是监听方法，用于监听注册中心的变更事件。
public synchronized void notify(List<URL> urls) {
// 将监听到的url分类，
// 按照协议名称或者category参数分为configurators，routers，providers三类
Map<String, List<URL>> categoryUrls = urls.stream()
.filter(Objects::nonNull)
.filter(this::isValidCategory)
.filter(this::isNotCompatibleFor26x)
.collect(Collectors.groupingBy(url -> {
if (UrlUtils.isConfigurator(url)) {
return CONFIGURATORS_CATEGORY;
} else if (UrlUtils.isRoute(url)) {
return ROUTERS_CATEGORY;
} else if (UrlUtils.isProvider(url)) {
return PROVIDERS_CATEGORY;
}
return "";
}));
// 如果有变化的configurators类别的url，那么将其转化为参数并设到成员变量configurators
List<URL> configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);
// 如果有变更的路由信息url，那么将其转化为Router对象并覆盖原先的路由信息
List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList());
toRouters(routerURLs).ifPresent(this::addRouters);
// providers
// 最后处理最重要的服务提供者变更信息，并用这些url刷新当前缓存的Invoker
List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());
refreshOverrideAndInvoker(providerURLs);
}
首先将从注册中心获取到的最新的url进行分类，根据协议名称或者category参数将url分为三类：configurators, routers, providers,
- configurators类型的url被转换为Configurator列表，覆盖本地缓存
- routers类型的url被转换为Router列表，并被设置到routerChain对象中
- providers类型的url则被用于接下来的创建Invoker
RegistryDirectory.refreshOverrideAndInvoker
private void refreshOverrideAndInvoker(List<URL> urls) {
// mock zookeeper://xxx?mock=return null
// 用变更的配置信息覆盖overrideDirectoryUrl成员变量
overrideDirectoryUrl();
// 刷新缓存中的Invokers
refreshInvoker(urls);
}
overrideDirectoryUrl方法的作用主要是用从注册中心以及配置中心监听到的变更的配置覆盖本地的overrideDirectoryUrl成员变量中的配置。我们接着往下走。
RegistryDirectory.refreshInvoker
// 入参invokerUrls是从注册中心拉取的服务提供者url
private void refreshInvoker(List<URL> invokerUrls) {
Assert.notNull(invokerUrls, "invokerUrls should not be null");
// 如果只有一个服务提供者，并且协议名称是empty，说明无提供者可用
// 将状态forbidden设为true， invokers设为空列表
if (invokerUrls.size() == 1
&& invokerUrls.get(0) != null
&& Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
this.forbidden = true; // Forbid to access
this.invokers = Collections.emptyList();
routerChain.setInvokers(this.invokers);
destroyAllInvokers(); // Close all invokers
} else {
this.forbidden = false; // Allow to access
// 记下旧的Invoker列表
Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
if (invokerUrls == Collections.<URL>emptyList()) {
invokerUrls = new ArrayList<>();
}
// 如果从注册中心没有拉取到服务提供者信息，那么使用之前缓存的服务提供者信息
// 这就是为什么dubbo在注册中心挂了之后消费者仍然能够调用提供者，因为消费者在本地进行了缓存
if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
invokerUrls.addAll(this.cachedInvokerUrls);
} else {
this.cachedInvokerUrls = new HashSet<>();
this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
}
// 如果注册中心没有提供者信息，并且本地也没有缓存，那么就没法进行服务调用了
if (invokerUrls.isEmpty()) {
return;
}
// 将服务提供者url转化为Invoker对象存放到map中
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
/**
* If the calculation is wrong, it is not processed.
*
* 1. The protocol configured by the client is inconsistent with the protocol of the server.
* eg: consumer protocol = dubbo, provider only has other protocol services(rest).
* 2. The registration center is not robust and pushes illegal specification data.
*
*/
if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) {
logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls
.toString()));
return;
}
List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
// pre-route and build cache, notice that route cache should build on original Invoker list.
// toMergeMethodInvokerMap() will wrap some invokers having different groups, those wrapped invokers not should be routed.
// 将生成的Invoker列表设置到routerChain的缓存中，
// routerChain将对这些Invoker进行路由
routerChain.setInvokers(newInvokers);
// 处理服务分组的情况
this.invokers = multiGroup ? toMergeInvokerList(newInvokers) : newInvokers;
// 将缓存的Invoker设置为新生成的
this.urlInvokerMap = newUrlInvokerMap;
try {
// 这里实际上求新的Invoker列表和旧的差集，将不再使用的旧的Invoker销毁
destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
} catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}
这个方法首先根据监听到的提供者url列表判断是否处于服务禁用状态，判断依据是：如果只有一个url，并且该url协议名称是empty，说明无提供者可用，将forbidden变量设为true,即禁止服务调用，
并做一下其他的相关设置以及销毁缓存中的Invoker。如果不是禁止状态，继续往下走。如果从注册中心获取到的url列表为空，那么检查本地缓存的url列表是否为空，如果缓存不为空就用缓存的列表。如果本地缓存也为空，说明无服务可用，直接返回。
如果如果从注册中心获取到的url列表不为空，说明有服务可用，这时就不会再去尝试本地缓存了（因为缓存已经过期了）,并且将本地缓存更新为新获取的url列表。
将可用的提供者url列表转化为Invoker列表。
将新创建的Invoker列表设置到routerChain中，这里呼应了前文提到的在doList方法中，从routerChain对象中取出缓存的Invoker列表。
将本地缓存的url->Invoker map更新为新创建的。
最后销毁缓存中不再使用的Invoker
RegistryDirectory.toInvokers
/**
* Turn urls into invokers, and if url has been refer, will not re-reference.
*
* @param urls 从注册中心拉取的服务提供者信息
* @return invokers
*/
private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<>();
if (urls == null || urls.isEmpty()) {
return newUrlInvokerMap;
}
// 用于防止对相同的url重复创建Invoker
Set<String> keys = new HashSet<>();
String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
for (URL providerUrl : urls) {
// If protocol is configured at the reference side, only the matching protocol is selected
// 如果消费端配置了协议名称，那么只有符合条件的提供者url才会被使用
// 这段代码有待商榷 ，应该先把queryProtocols处理好，避免重复做同样的工作
if (queryProtocols != null && queryProtocols.length() > 0) {
boolean accept = false;
String[] acceptProtocols = queryProtocols.split(",");
for (String acceptProtocol : acceptProtocols) {
if (providerUrl.getProtocol().equals(acceptProtocol)) {
accept = true;
break;
}
}
if (!accept) {
continue;
}
}
// 如果协议名称是empty，那么忽略该条url
if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
continue;
}
// 如果当前classpath下找不到与提供者url中协议名称相对应的Protocol类，那么打印错误日志同时忽略该条url
if (!ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() +
" in notified url: " + providerUrl + " from registry " + getUrl().getAddress() +
" to consumer " + NetUtils.getLocalHost() + ", supported protocol: " +
ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
continue;
}
// 合并消费端设置的参数以及从注册中心，配置中心监听到的配置变更
URL url = mergeUrl(providerUrl);
// 以全路径作为该url的唯一标识
String key = url.toFullString(); // The parameter urls are sorted
if (keys.contains(key)) { // Repeated url
continue;
}
keys.add(key);
// Cache key is url that does not merge with consumer side parameters, regardless of how the consumer combines parameters, if the server url changes, then refer again
Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
// 如果之前已经创建过该url的Invoker对象，那么就不用再重复创建
Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
if (invoker == null) { // Not in the cache, refer again
try {
boolean enabled = true;
// 检查disabled和enabled参数的值
if (url.hasParameter(Constants.DISABLED_KEY)) {
enabled = !url.getParameter(Constants.DISABLED_KEY, false);
} else {
enabled = url.getParameter(Constants.ENABLED_KEY, true);
}
if (enabled) {
// 真正创建Invoker的地方，
// InvokerDelegate只是个简单的包装类，不需要多说
invoker = new InvokerDelegate<>(protocol.refer(serviceType, url), url, providerUrl);
}
} catch (Throwable t) {
logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
}
if (invoker != null) { // Put new invoker in cache
newUrlInvokerMap.put(key, invoker);
}
} else {
newUrlInvokerMap.put(key, invoker);
}
}
keys.clear();
return newUrlInvokerMap;
}
- 首先根据协议名称检查url是否可用。url的协议必须在本地配置的协议列表中（如果没有配置就不需要做此检查）；如果协议名称是empty则忽略这个url；如果当前classpath下找不到与提供者url中协议名称相对应的Protocol类，那么打印错误日志同时忽略该条url
- 合并消费端设置的参数以及从注册中心，配置中心监听到的配置变更
- 检查disabled，enabled参数的值，判断该url是否启用，如果disabled为true则跳过该url；如果没有disabled参数，检查enabled参数，如果enabled为false则跳过该url，enabled默认是true。
- 调用Protocol.refer方法创建Invoker对象。
这里需要说明一下，由于Directory不是通过SPI机制加载的，所以RegistryDirectory也不是通过ExtensionLoader加载的，所以也就不会受到ExtensionLoader的IOC影响。RegistryDirectory内部的protocol成员是在RegistryDirectory初始化之后通过调用setter方法设置进去的，是在RegistryProtocol.doRefer方法中完成的。而RegistryProtocol是通过ExtensionLoader机制加载的，会受到IOC影响，所以RegistryProtocol实例内部的protocol成员是通过ExtensionLoader的IOC机制自动注入的，是一个自适应的扩展类。
另外，InvokerDelegate只是个简单的包装类，不需要多说。
Invoker的创建最终还是通过protocol.refer方法，我们以最常用的dubbo协议为例进行分析。
DubboProtocol.refer
@Override
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
optimizeSerialization(url);
// create rpc invoker.
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
这个方法很简单，直接new了一个DubboInvoker。
DubboInvoker
看一下doInvoke方法，这个方法主要是处理了同步，异步，超时，单向调用等参数，并且对调用结果封装了异步调用，同步调用的逻辑。
真正执行远程调用的部分是靠ExchangeClient实现的，再往下就是调用参数的序列化，tcp连接创建，发送报文，获取响应报文，反序列化结果等的逻辑了，本文不再深入下去。
赞 (0)