本系列主要参考官网文档、芋道源码的源码解读和《深入理解Apache Dubbo与实战》一书。Dubbo版本为2.6.1。
文章内容顺序：
- 什么是Directory
- 1.1 Directory简介
- 1.2 集群容错的总体工作流程
在doRefer中调用了new RegistryDirectory()
Directory结构及代码分析
- 3.1 Directory的UML图
- 3.2 AbstractDirectory#list
- 3.3 RegistryDirectory类的简单介绍
- 3.4 RegistryDirectory实现订阅的代码
- 3.5 RegistryDirectory#doList 列举 Invoker
- 3.6 泛化调用的简单介绍
- Directory中Invoker的更新逻辑
- 4.1 AbstractRegistry#notify
- 4.2 RegistryDirectory#notify
- 4.3 RegistryDirectory#refreshInvoker
- 4.4 RegistryDirectory#toInvokers(invokerUrls)
- 4.5 RegistryDirectory#toMethodInvokers(newUrlInvokerMap)
- 4.6 服务分组的简单介绍
- 4.7 RegistryDirectory#destroyUnusedInvokers
- 4.8 Invoker 列表的刷新逻辑总结
Directory与ClusterInvoker的关系
StaticDirectory
1. 什么是Directory
1.1 Directory简介
在进行深入分析之前，我们先来了解一下服务目录(Directory)是什么。服务目录中存储了一些和服务提供者有关的信息，通过服务目录，服务消费者可获取到服务提供者的信息，比如 ip、端口、服务协议等。
通过这些信息，服务消费者就可通过 Netty 等客户端进行远程调用。在一个服务集群中，服务提供者数量并不是一成不变的，如果集群中新增了一台机器，相应地在服务目录中就要新增一条服务提供者记录。或者，如果服务提供者的配置修改了，服务目录中的记录也要做相应的更新。
实际上服务目录在获取注册中心的服务配置信息后，会为每条配置信息生成一个 Invoker 对象，并把这个 Invoker 对象存储起来，这个 Invoker 才是服务目录最终持有的对象。Invoker 有什么用呢？看名字就知道了，这是一个具有远程调用功能的对象。讲到这大家应该知道了什么是服务目录了，它可以看做是 Invoker 集合，且这个集合中的元素会随注册中心的变化而进行动态调整。
1.2 集群容错的总体工作流程
这里再提一遍在Dubbo集群容错——Cluster中提过的工作流程。
image.png
在这之前先说说Cluster的总体工作流程，
1.先生成Invoker对象，根据不同的Cluster实现生成不同类型的ClusterInvoker（这个就是服务引用阶段）
2.调用list从Directory中获取可用的服务列表（从这一步开始真正的调用流程）,接着使用Router接口根据路由规则过滤一部分服务后最终返回服务列表
3.调用select做负载均衡，通过不同的负载均衡策略选出一个服务作为最后的调用
4.调用invoke做RPC调用，对于调用出现异常、成功、失败等情况，每种容错机制会有不同的处理方式。
2. 在doRefer中调用了new RegistryDirectory()
那么这篇我们就从Directory来开始讲，在服务引用篇中，如果只配置了一个注册中心的话，会最终调用DoRefer方法，在里面new RegistryDirectory(),那么就来看看这个RegistryDirectory()的实现吧。
image.png
3. Directory结构及代码分析
3.1 Directory的UML图
Directory ，中文直译为目录，代表了多个 Invoker ，可以把它看成 List<Invoker> 。但与 List 不同的是，它的值可能是动态变化的，比如注册中心推送变更。
先来看看他的类图。
image.png
StaticDirectory ，静态 Directory 实现类，从命名上看出它是静态的 List<Invoker> 。(对于StaticDirectory的分析笔者实在力所不逮，只作简略描述)
RegistryDirectory ，基于注册中心的动态 Directory 实现类，从命名上看出它是动态的，会根据注册中心的推送变更 List<Invoker> 。
3.2 AbstractDirectory#list
AbstractDirectory 封装了 Invoker列举流程，具体的列举逻辑则由子类实现，这是典型的模板模式。所以，接下来我们先来看一下 AbstractDirectory 的源码。
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
// 调用 doList 方法列举 Invoker，doList 是模板方法，由子类实现
List<Invoker<T>> invokers = doList(invocation);
// 根据路由规则，筛选 Invoker 集合
List<Router> localRouters = this.routers; // local reference 本地引用，避免并发问题
if (localRouters != null && !localRouters.isEmpty()) {
for (Router router : localRouters) {
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
// 进行服务路由
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
return invokers;
}
protected abstract List<Invoker<T>> doList(Invocation invocation) throws RpcException;
上面就是 AbstractDirectory#list 方法源码，这个方法封装了 Invoker 的列举过程。如下：
- 调用
doList获取 Invoker 列表- 根据
Router#getUrl返回值为空与否，以及runtime参数决定是否进行服务路由
以上步骤中，doList是模板方法，需由子类实现。- 注意这里的
router，我们会在Dubbo集群容错——Router中详细介绍，这里只需要知道会通过某个规则筛选即可。
3.3 RegistryDirectory类的简单介绍
接下来看看他的子类RegistryDirectory的源码。
RegistryDirectory是一种动态服务目录，实现了NotifyListener接口，实现对变更的监听。
当注册中心服务配置发生变化后，RegistryDirectory可收到与当前服务相关的变化。收到变更通知后，RegistryDirectory可根据配置变更信息刷新 Invoker 列表。RegistryDirectory中有几个比较重要的逻辑，
第一是 Invoker 的列举逻辑，
第二是接收服务配置变更的逻辑，
第三是 Invoker 列表的刷新逻辑。
3.4 RegistryDirectory实现订阅的代码
在此之前先来简单看看RegistryDirectory的订阅逻辑，同样的，可以在执行服务引用的方法RegistryProtocol#doRefer中看到RegistryDirectory#subscribe
public void subscribe(URL url) {
// 设置消费者 URL
setConsumerUrl(url);
// 向注册中心，发起订阅
registry.subscribe(url, this);
}
通过
Registry实现类向注册中心发起订阅，以ZooKeeper为例，当订阅的节点发生变化，会通过这个url，在ZookeeperRegistry节点的zkListeners拿到对应的监听器实现，也就是RegistryDirectory，从而调用NotifyListener#notify方法来改变对应的invoker。如下面两图所示
image.png
image.png
3.5 RegistryDirectory#doList 列举 Invoker
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
// 服务提供者关闭或禁用了服务，此时抛出 No provider 异常
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
"No provider available from registry ...");
}
List<Invoker<T>> invokers = null;
// 获取 Invoker 本地缓存
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
// 获取方法名和参数列表
String methodName = RpcUtils.getMethodName(invocation);
Object[] args = RpcUtils.getArguments(invocation);
// 检测参数列表的第一个参数是否为 String 或 enum 类型
if (args != null && args.length > 0 && args[0] != null
&& (args[0] instanceof String || args[0].getClass().isEnum())) {
// 通过 方法名 + 第一个参数名称 查询 Invoker 列表，具体的使用场景暂时没想到
invokers = localMethodInvokerMap.get(methodName + "." + args[0]);
}
if (invokers == null) {
// 通过方法名获取 Invoker 列表
invokers = localMethodInvokerMap.get(methodName);
}
if (invokers == null) {
// 通过星号 * 获取 Invoker 列表
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
// 冗余逻辑，pull request #2861 移除了下面的 if 分支代码
if (invokers == null) {
Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
invokers = iterator.next();
}
}
}
// 返回 Invoker 列表
return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}
以上代码进行多次尝试，以期从localMethodInvokerMap 中获取到 Invoker 列表。一般情况下，普通的调用可通过方法名获取到对应的Invoker 列表，泛化调用可通过 * 获取到 Invoker 列表。
那么这methodInvokerMap又是何方神圣呢，它实际上是一个方法名对于Invoker的一个映射Map，相当于本地的缓存，接着往下看，在RegistryDirectory#refreshInvoker会详细介绍到
3.6 泛化调用的简单介绍：
服务消费者和服务提供者两端同路径下有同样的接口，只不过在服务提供者端会有该接口的具体实现，之所以在服务消费者有一个没有任何具体实现的接口，是因为在设计RPC之初，设计者的最高理念就是你去面向接口编程，你在进行远程调用的时候，并没有意识到你在进行远程调用，却也能拿到接口一样，相信你也感觉到了，服务消费者在调用服务的时候，与调用一个普通的接口是一样的。
泛化调用就是服务消费者端因为某种原因并没有该服务接口，这个原因有很多，比如是跨语言的，一个PHP工程师想调用某个java接口，他并不能按照你约定，去写一个个的接口，Dubbo并不是跨语言的RPC框架，但并不是不能解决这个问题，这个PHP程序员搭建了一个简单的java web项目，引入了dubbo的jar包，使用dubbo的泛化调用，然后利用web返回json，这样也能完成跨语言的调用。泛化调用的好处之一就是这个了。
好了，简而言之，泛化调用，最最直接的表现就是服务消费者不需要有任何接口的实现，就能完成服务的调用。
官网有关泛化调用的例子:泛化引用
4. Directory中Invoker的更新逻辑
上面我们知道了
RegistryDirectory是实现了NotifyListener的，他要实现这个接口的notify方法才能被AbstractRegistry#notify(URL url, NotifyListener, List<URL> urls)方法调用到，在RegistryDirectory#notify方法里实现缓存的更新。
4.1AbstractRegistry#notify
在了解RegistryDirectory#notify前，首先我们先看一下AbstractRegistry#notify(URL url, NotifyListener, List<URL> urls)
image.png
上图对应对应为
AbstractRegistry#notify(URL url, NotifyListener, List<URL> urls)方法，在注册中心( Registry )发现数据发生变化时，会通知对应的NotifyListener们。
因为RegistryDirectory作为一个NotifyListener，向注册中心( Registry )发起了订阅，所以此时会被通知。注意，是按照分类循环通知的，也就是说，一次只有一类 URL 。
4.2 RegistryDirectory#notify
RegistryDirectory 是一个动态服务目录，会随注册中心配置的变化进行动态调整：
public synchronized void notify(List<URL> urls) {
// 定义三个集合，分别用于存放服务提供者 url，路由 url，配置器 url
List<URL> invokerUrls = new ArrayList<URL>();
List<URL> routerUrls = new ArrayList<URL>();
List<URL> configuratorUrls = new ArrayList<URL>();
for (URL url : urls) {
String protocol = url.getProtocol();
// 获取 category 参数
String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
// 根据 category 参数将 url 分别放到不同的列表中
if (Constants.ROUTERS_CATEGORY.equals(category)
|| Constants.ROUTE_PROTOCOL.equals(protocol)) {
// 添加路由器 url
routerUrls.add(url);
} else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
|| Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
// 添加配置器 url
configuratorUrls.add(url);
} else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
// 添加服务提供者 url
invokerUrls.add(url);
} else {
// 忽略不支持的 category
logger.warn("Unsupported category ...");
}
}
if (configuratorUrls != null && !configuratorUrls.isEmpty()) {
// 将 url 转成 Configurator
this.configurators = toConfigurators(configuratorUrls);
}
if (routerUrls != null && !routerUrls.isEmpty()) {
// 将 url 转成 Router
List<Router> routers = toRouters(routerUrls);
if (routers != null) {
setRouters(routers);
}
}
List<Configurator> localConfigurators = this.configurators;
this.overrideDirectoryUrl = directoryUrl;
if (localConfigurators != null && !localConfigurators.isEmpty()) {
for (Configurator configurator : localConfigurators) {
// 配置 overrideDirectoryUrl
this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
}
}
// 刷新 Invoker 列表
refreshInvoker(invokerUrls);
}
如上，
notify方法首先是根据url的category参数对url进行分门别类存储，然后通过toRouters和toConfigurators两个方法将url列表转成Router和Configurator列表。最后调用refreshInvoker方法刷新Invoker列表。
那么就继续往下看看RegistryDirectory#refreshInvoker
4.3 RegistryDirectory#refreshInvoker
这边也会有我们刚才提到的存储了Invoker列表的methodInvokerMap
private void refreshInvoker(List<URL> invokerUrls) {
//invokerUrls 仅有一个元素，且 url 协议头为 empty，此时表示禁用所有服务
if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
&& Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
// 设置禁止访问
this.forbidden = true; // Forbid to access
// methodInvokerMap 置空
this.methodInvokerMap = null; // Set the method invoker map to null
// 销毁所有 Invoker 集合
destroyAllInvokers(); // Close all invokers
} else {
// 设置允许访问
this.forbidden = false; // Allow to access
// 引用老的 urlInvokerMap
Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
// 传入的 invokerUrls 为空，说明是路由规则或配置规则发生改变，参考notify中的三个集合
//此时 invokerUrls 是空的，直接使用 cachedInvokerUrls 。
if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
invokerUrls.addAll(this.cachedInvokerUrls);
// 传入的 invokerUrls 非空，更新 cachedInvokerUrls 。
} else {
this.cachedInvokerUrls = new HashSet<URL>();
this.cachedInvokerUrls.addAll(invokerUrls); //Cached invoker urls, convenient for comparison //缓存invokerUrls列表，便于交叉对比
}
// 忽略，若无 invokerUrls
if (invokerUrls.isEmpty()) {
return;
}
// 将传入的 invokerUrls ，转成新的 urlInvokerMap
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
// 将 newUrlInvokerMap 转成方法名到 Invoker 列表的映射
Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // Change method name to map Invoker Map
// 转换出错，直接打印异常，并返回
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :" + invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls.toString()));
return;
}
// 若服务引用多 group ，则按照 method + group 聚合 Invoker 集合
this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
this.urlInvokerMap = newUrlInvokerMap;
// 销毁不再使用的 Invoker 集合
try {
destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
} catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}
refreshInvoker方法首先会根据入参invokerUrls的数量和协议头判断是否禁用所有的服务，如果禁用，则将forbidden设为true，并销毁所有的Invoker。
若不禁用，则将url转成Invoker，得到<url, Invoker>的映射关系。然后进一步进行转换，得到<methodName, Invoker 列表>映射关系。之后进行多组Invoker合并操作，并将合并结果赋值给methodInvokerMap。methodInvokerMap变量在doList方法中会被用到，doList会对该变量进行读操作，在这里是写操作。当新的Invoker列表生成后，还要一个重要的工作要做，就是销毁无用的Invoker，避免服务消费者调用已下线的服务的服务。
注意这里有四个比较有意思的方法:
一个是将传入的
invokerUrls，转成新的urlInvokerMap的toInvokers(invokerUrls)
一个是将newUrlInvokerMap转成方法名到Invoker列表的映射toMethodInvokers(newUrlInvokerMap)
还有一个就是按照method + group聚合Invoker集合的方方法toMergeMethodInvokerMap(newMethodInvokerMap)
最后销毁不使用的Invoker的方法，destroyUnusedInvokers
4.4 RegistryDirectory#toInvokers(invokerUrls)
按照顺序一个个来看一下吧
private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
// 新的 `newUrlInvokerMap`
Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
// 若为空，直接返回
if (urls == null || urls.isEmpty()) {
return newUrlInvokerMap;
}
// 已初始化的服务器提供 URL 集合
Set<String> keys = new HashSet<String>();
// 获取服务消费端配置的协议
String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
// 循环服务提供者 URL 集合，转成 Invoker 集合
for (URL providerUrl : urls) {
// 如果 reference 端配置了 protocol ，则只选择匹配的 protocol
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) {
// 若服务消费者协议头不被消费者所支持，则忽略当前 providerUrl
continue;
}
}
// 忽略，若为 `empty://` 协议
if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
continue;
}
// 通过 SPI 检测服务端协议是否被消费端支持，不支持则抛出异常
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 url = mergeUrl(providerUrl);
// 忽略，若已经初始化
String key = url.toFullString(); // The parameter urls are sorted
if (keys.contains(key)) { // Repeated url
continue;
}
// 添加到 `keys` 中
keys.add(key);
// 将本地 Invoker 缓存赋值给 localUrlInvokerMap
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;
if (url.hasParameter(Constants.DISABLED_KEY)) {
// 获取 disable 配置，取反，然后赋值给 enable 变量
enabled = !url.getParameter(Constants.DISABLED_KEY, false);
} else {
// 获取 enable 配置，并赋值给 enable 变量
enabled = url.getParameter(Constants.ENABLED_KEY, true);
}
// 若开启，创建 Invoker 对象
if (enabled) {
// 注意，引用服务
invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
}
} catch (Throwable t) {
logger.error("Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
}
// 添加到 newUrlInvokerMap 中
if (invoker != null) { // Put new invoker in cache
newUrlInvokerMap.put(key, invoker);
}
} else { // 在缓存中，直接使用缓存的 Invoker 对象，添加到 newUrlInvokerMap 中
newUrlInvokerMap.put(key, invoker);
}
}
// 清空 keys
keys.clear();
return newUrlInvokerMap;
}
toInvokers方法一开始会对服务提供者 url 进行检测，若服务消费端的配置不支持服务端的协议，或服务端url协议头为empty时，toInvokers均会忽略服务提供方 url。
必要的检测做完后，紧接着是合并url，然后访问缓存，尝试获取与 url 对应的 invoker。如果缓存命中，直接将Invoker存入newUrlInvokerMap中即可。
如果未命中，则通过调用Protocol$Adaptive#refer(serviceType, url)方法，引用服务，创建服务提供者 Invoker 对象
4.5 RegistryDirectory#toMethodInvokers(newUrlInvokerMap)
private Map<String, List<Invoker<T>>> toMethodInvokers(Map<String, Invoker<T>> invokersMap) {
// 创建新的 `methodInvokerMap`，方法名 -> Invoker 列表
Map<String, List<Invoker<T>>> newMethodInvokerMap = new HashMap<String, List<Invoker<T>>>();
// 创建 Invoker 集合
List<Invoker<T>> invokersList = new ArrayList<Invoker<T>>();
// 按服务提供者 URL 所声明的 methods 分类，兼容注册中心执行路由过滤掉的 methods
if (invokersMap != null && invokersMap.size() > 0) {
// 循环每个服务提供者 Invoker
for (Invoker<T> invoker : invokersMap.values()) {
// 获取 methods 参数
String parameter = invoker.getUrl().getParameter(Constants.METHODS_KEY);
if (parameter != null && parameter.length() > 0) {
// 切分 methods 参数值，得到方法名数组
String[] methods = Constants.COMMA_SPLIT_PATTERN.split(parameter);
if (methods != null && methods.length > 0) {
// 循环每个方法，按照方法名为维度，聚合到 `methodInvokerMap` 中
for (String method : methods) {
// 方法名不为 *
if (method != null && method.length() > 0 && !Constants.ANY_VALUE.equals(method)) { // 当服务提供者的方法为 "*" ，代表泛化调用
// 根据方法名获取 Invoker 列表
List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
if (methodInvokers == null) {
methodInvokers = new ArrayList<Invoker<T>>();
newMethodInvokerMap.put(method, methodInvokers);
}
// 存储 Invoker 到列表中
methodInvokers.add(invoker);
}
}
}
}
// 添加到 `invokersList` 中
invokersList.add(invoker);
}
}
// // 进行服务级别路由，路由全 `invokersList` ，匹配合适的 Invoker 集合
List<Invoker<T>> newInvokersList = route(invokersList, null);
// 存储 <*, newInvokersList> 映射关系，添加 `newInvokersList` 到 `newMethodInvokerMap` 中，表示该服务提供者的全量 Invoker 集合
newMethodInvokerMap.put(Constants.ANY_VALUE, newInvokersList);
// 循环，基于每个方法路由，匹配合适的 Invoker 集合
if (serviceMethods != null && serviceMethods.length > 0) {
for (String method : serviceMethods) {
List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
if (methodInvokers == null || methodInvokers.isEmpty()) {
methodInvokers = newInvokersList;
}
// 进行方法级别路由
newMethodInvokerMap.put(method, route(methodInvokers, method));
}
}
// 循环排序每个方法的 Invoker 集合，并设置为不可变
for (String method : new HashSet<String>(newMethodInvokerMap.keySet())) {
List<Invoker<T>> methodInvokers = newMethodInvokerMap.get(method);
Collections.sort(methodInvokers, InvokerComparator.getComparator());
newMethodInvokerMap.put(method, Collections.unmodifiableList(methodInvokers));
}
return Collections.unmodifiableMap(newMethodInvokerMap);
}
上面方法主要做了三件事情， 第一是对入参进行遍历，然后从
Invoker的url成员变量中获取methods参数，并切分成数组。随后以方法名为键，Invoker列表为值，将映射关系存储到newMethodInvokerMap中。
第二是分别基于类和方法对Invoker列表进行路由操作。
第三是对Invoker列表进行排序，并转成不可变列表。关于toMethodInvokers方法就先分析到这，我们继续向下分析，这次要分析的多组服务的合并逻辑。
4.6 RegistryDirectory# toMergeMethodInvokerMap(methodMap)
private Map<String, List<Invoker<T>>> toMergeMethodInvokerMap(Map<String, List<Invoker<T>>> methodMap) {
Map<String, List<Invoker<T>>> result = new HashMap<String, List<Invoker<T>>>();
// 遍历入参
for (Map.Entry<String, List<Invoker<T>>> entry : methodMap.entrySet()) {
String method = entry.getKey();
List<Invoker<T>> invokers = entry.getValue();
// 按照 Group 聚合 Invoker 集合的结果。其中，KEY：group VALUE：Invoker 集合。
Map<String, List<Invoker<T>>> groupMap = new HashMap<String, List<Invoker<T>>>();
// 遍历 Invoker 列表
for (Invoker<T> invoker : invokers) {
String group = invoker.getUrl().getParameter(Constants.GROUP_KEY, "");
List<Invoker<T>> groupInvokers = groupMap.get(group);
if (groupInvokers == null) {
groupInvokers = new ArrayList<Invoker<T>>();
// 缓存 <group, List<Invoker>> 到 groupMap 中
groupMap.put(group, groupInvokers);
}
// 存储 invoker 到 groupInvokers
groupInvokers.add(invoker);
}
if (groupMap.size() == 1) {
// 如果 groupMap 中仅包含一组键值对，此时直接取出该键值对的值即可
result.put(method, groupMap.values().iterator().next());
// 大于 1，将每个 Group 的 Invoker 集合，创建成 Cluster Invoker 对象。
} else if (groupMap.size() > 1) {
List<Invoker<T>> groupInvokers = new ArrayList<Invoker<T>>();
for (List<Invoker<T>> groupList : groupMap.values()) {
// 通过集群类合并每个分组对应的 Invoker 列表
groupInvokers.add(cluster.join(new StaticDirectory<T>(groupList)));
}
// 缓存结果
result.put(method, groupInvokers);
// 大小为 0 ，使用原有值
} else {
result.put(method, invokers);
}
}
return result;
}
上面方法首先是生成
group到Invoker列表的映射关系表，若关系表中的映射关系数量大于1，表示有多组服务。此时通过集群类合并每组Invoker，并将合并结果存储到groupInvokers中。之后将方法名与groupInvokers存到到result中，并返回，整个逻辑结束。
4.6 服务分组的简单介绍
其中分组的概念指的是当一个接口有多种实现时，可以用 group 区分。
image.png
4.7 RegistryDirectory#destroyUnusedInvokers
private void destroyUnusedInvokers(Map<String, Invoker<T>> oldUrlInvokerMap, Map<String, Invoker<T>> newUrlInvokerMap) {
// 防御性编程，目前不存在这个情况
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) {
// 销毁所有服务提供者 Invoker
destroyAllInvokers();
return;
}
// 对比新老集合，计算需要销毁的 Invoker 集合
List<String> deleted = null;
if (oldUrlInvokerMap != null) {
// 获取新生成的 Invoker 列表
Collection<Invoker<T>> newInvokers = newUrlInvokerMap.values();
// 遍历老的 <url, Invoker> 映射表
for (Map.Entry<String, Invoker<T>> entry : oldUrlInvokerMap.entrySet()) {
// 检测 newInvokers 中是否包含老的 Invoker
if (!newInvokers.contains(entry.getValue())) {
if (deleted == null) {
deleted = new ArrayList<String>();
}
// 若不包含，则将老的 Invoker 对应的 url 存入 deleted 列表中
deleted.add(entry.getKey());
}
}
}
// 若有需要销毁的 Invoker ，则进行销毁
if (deleted != null) {
for (String url : deleted) {
if (url != null) {
// 移除出 `urlInvokerMap`
Invoker<T> invoker = oldUrlInvokerMap.remove(url);
if (invoker != null) {
try {
// 销毁 Invoker
invoker.destroy();
if (logger.isDebugEnabled()) {
logger.debug("destroy invoker[" + invoker.getUrl() + "] success. ");
}
} catch (Exception e) {
logger.warn("destroy invoker[" + invoker.getUrl() + "] failed. " + e.getMessage(), e);
}
}
}
}
}
}
destroyUnusedInvokers方法的主要逻辑是通过newUrlInvokerMap找出待删除Invoker对应的url，并将url存入到deleted列表中。然后再遍历deleted列表，并从oldUrlInvokerMap中移除相应的Invoker，销毁之。整个逻辑大致如此，不是很难理解。
4.8 Invoker 列表的刷新逻辑总结
到此关于 Invoker 列表的刷新逻辑就分析了，这里对整个过程进行简单总结。如下：
- 检测入参是否仅包含一个 url，且 url 协议头为 empty
- 若第一步检测结果为 true，表示禁用所有服务，此时销毁所有的 Invoker
- 若第一步检测结果为 false，此时将入参转为 Invoker 列表
- 对上一步逻辑生成的结果进行进一步处理，得到方法名到 Invoker 的映射关系表
合并多组 Invoker- 销毁无用 Invoker
5. Directory与ClusterInvoker的关系
执行体是ClusterInvoker 里面有Directory，RegistryDirectory中通过list方法可以取出来对应的接口方法所有的Invoker列表，这些Inovker只有provider的区别，实现上都是一样的，也就是说这些Invoker对应的是不同机器上的这个方法的实现。
6. StaticDirectory
StaticDirectory中存的是不同注册中心的invoker，默认用的是AvailableCluster，
public class StaticDirectory<T> extends AbstractDirectory<T> {
// Invoker 列表
private final List<Invoker<T>> invokers;
// 省略构造方法
@Override
public Class<T> getInterface() {
// 获取接口类
return invokers.get(0).getInterface();
}
// 检测服务目录是否可用
@Override
public boolean isAvailable() {
if (isDestroyed()) {
return false;
}
for (Invoker<T> invoker : invokers) {
if (invoker.isAvailable()) {
// 只要有一个 Invoker 是可用的，就认为当前目录是可用的
return true;
}
}
return false;
}
@Override
public void destroy() {
if (isDestroyed()) {
return;
}
// 调用父类销毁逻辑
super.destroy();
// 遍历 Invoker 列表，并执行相应的销毁逻辑
for (Invoker<T> invoker : invokers) {
invoker.destroy();
}
invokers.clear();
}
@Override
protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {
// 列举 Inovker，也就是直接返回 invokers 成员变量
return invokers;
}
}
StaticDirectory的代码非常简单，他的invokers是通过构造器传入的，不能动态更新。
而RegistryDirectory进一步实现了NotifyListener接口，当服务接口对应的注册信息发生变化时会回调notify(URL url, NotifyListener listener, List<URL> urls)方法，通知RegistryDirectory更新Invoker列表，才具有动态写的功能。
网友评论