2021SC@SDUSC
dubbo容错机制
集群容错源码包含四个部分,分别是服务目录 Directory、服务路由 Router、集群 Cluster 和负载均衡 LoadBalance。
服务目录概念
服务目录中存储了一些和服务提供者有关的信息,通过服务目录,服务消费者可获取到服务提供者的信息,比如 ip、端口、服务协议等。通过这些信息,服务消费者就可通过 Netty 等客户端进行远程调用。在一个服务集群中,服务提供者数量并不是一成不变的,如果集群中新增了一台机器,相应地在服务目录中就要新增一条服务提供者记录。或者,如果服务提供者的配置修改了,服务目录中的记录也要做相应的更新。
服务目录在获取注册中心的服务配置信息后,会为每条配置信息生成一个 Invoker 对象,并把这个 Invoker 对象存储起来,这个 Invoker 是服务目录最终持有的对象,是一个具有远程调用功能的对象。
继承体系
服务目录目前内置的实现有两个,分别为 StaticDirectory 和 RegistryDirectory,它们均是 AbstractDirectory 的子类。AbstractDirectory 实现了 Directory 接口,这个接口包含了一个重要的方法定义,即 list(Invocation),用于列举 Invoker。
AbstractDirectory 源码
public List> list(Invocation invocation) throws RpcException { if (destroyed) { throw new RpcException("Directory already destroyed..."); } // 调用 doList 方法列举 Invoker,doList 是模板方法,由子类实现 List > invokers = doList(invocation); // 获取路由 Router 列表 List localRouters = this.routers; if (localRouters != null && !localRouters.isEmpty()) { for (Router router : localRouters) { try { // 获取 runtime 参数,并根据参数决定是否进行路由 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: ..."); } } } return invokers; } // 模板方法,由子类实现 protected abstract List > doList(Invocation invocation) throws RpcException;
上面是AbstractDirectory 的 list 方法源码,这个方法封装了 Invoker 的列举过程:
- 调用 doList 获取 Invoker 列表
- 根据 Router 的 getUrl 返回值为空与否,以及 runtime 参数决定是否进行服务路由。
以上步骤中,doList 是模板方法,需由子类实现。Router 的 runtime 参数这里简单说明一下,这个参数决定了是否在每次调用服务时都执行路由规则。如果 runtime 为 true,那么每次调用服务前,都需要进行服务路由。
RegistryDirectory源码
RegistryDirectory 是一种动态服务目录,实现了 NotifyListener 接口。当注册中心服务配置发生变化后,RegistryDirectory 可收到与当前服务相关的变化。收到变更通知后,RegistryDirectory 可根据配置变更信息刷新 Invoker 列表。RegistryDirectory 中有几个比较重要的逻辑,第一是 Invoker 的列举逻辑,第二是接收服务配置变更的逻辑,第三是 Invoker 列表的刷新逻辑。
Invoke相关public List> doList(Invocation invocation) { if (forbidden) { // 服务提供者关闭或禁用了服务,此时抛出 No provider 异常 throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "No provider available from registry ..."); } List > invokers = null; // 获取 Invoker 本地缓存 Map >> 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 >> iterator = localMethodInvokerMap.values().iterator(); if (iterator.hasNext()) { invokers = iterator.next(); } } } // 返回 Invoker 列表 return invokers == null ? new ArrayList
>(0) : invokers; }
以上代码进行多次尝试,以期从 localMethodInvokerMap 中获取到 Invoker 列表。一般情况下,普通的调用可通过方法名获取到对应的 Invoker 列表,泛化调用可通过 * 获取到 Invoker 列表。localMethodInvokerMap 源自 RegistryDirectory 类的成员变量 methodInvokerMap。
接收服务变更public synchronized void notify(Listurls) { // 定义三个集合,分别用于存放服务提供者 url,路由 url,配置器 url List invokerUrls = new ArrayList (); List routerUrls = new ArrayList (); List configuratorUrls = new ArrayList (); 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 routers = toRouters(routerUrls); if (routers != null) { setRouters(routers); } } List 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 列表。
刷新 Invoker 列表private void refreshInvoker(ListinvokerUrls) { // invokerUrls 仅有一个元素,且 url 协议头为 empty,此时表示禁用所有服务 if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) { // 设置 forbidden 为 true this.forbidden = true; this.methodInvokerMap = null; // 销毁所有 Invoker destroyAllInvokers(); } else { this.forbidden = false; Map > oldUrlInvokerMap = this.urlInvokerMap; if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) { // 添加缓存 url 到 invokerUrls 中 invokerUrls.addAll(this.cachedInvokerUrls); } else { this.cachedInvokerUrls = new HashSet (); // 缓存 invokerUrls this.cachedInvokerUrls.addAll(invokerUrls); } if (invokerUrls.isEmpty()) { return; } // 将 url 转成 Invoker Map > newUrlInvokerMap = toInvokers(invokerUrls); // 将 newUrlInvokerMap 转成方法名到 Invoker 列表的映射 Map >> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 转换出错,直接打印异常,并返回 if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0) { logger.error(new IllegalStateException("urls to invokers error ...")); return; } // 合并多个组的 Invoker this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap; this.urlInvokerMap = newUrlInvokerMap; try { // 销毁无用 Invoker destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); } catch (Exception e) { logger.warn("destroyUnusedInvokers error. ", e); } } }
refreshInvoker 方法首先会根据入参 invokerUrls 的数量和协议头判断是否禁用所有的服务,如果禁用,则将 forbidden 设为 true,并销毁所有的 Invoker。若不禁用,则将 url 转成 Invoker,得到



