微服务架构之Dubbo集群容错(5)

以 Invoker 为中心，从 Cluster, Directory, Router, LoadBalance，来解析各个接口。

Router

Router 负责从多个 Invoker 中按路由规则选出子集，比如读写分离，应用隔离等。下图是dubbo-admin的配置界面。

路由用于灰度发布等业务场景。根据官网描述，目前路由有3种：ConditionRouter, MockInvokersSelector, ScriptRouter。

ScriptRouter：脚本路由规则支持 JDK 脚本引擎的所有脚本，比如：javascript, jruby, groovy 等，通过 type=javascript 参数设置脚本类型，缺省为 javascript。

MockInvokersSelector：一般dubbo-admin没有配置时候会走这条路线，在微服务架构之Dubbo集群容错(2)有介绍，一般的调用没有设置mock，会在getNormalInvokers方法中的hasMockProviders判断中就返回了；若果设置mock后，就将设置为mock的invoker返回。

ConditionRouter：即条件路由，主要根据dubbo管理控制台配置的路由规则来过滤相关的invoker, 当我们对路由规则点击启用的时候, 就会触发RegistryDirectory类之前提到的notify方法。在RegistryDirectory的notify方法中有这样的几行代码

// routers
       if (routerUrls != null && routerUrls.size() > 0) {
           List<Router> routers = toRouters(routerUrls);
           if (routers != null) { // null - do nothing
               setRouters(routers);
           }
       }

//url转化为router
private List<Router> toRouters(List<URL> urls) {
        List<Router> routers = new ArrayList<Router>();
        if (urls == null || urls.size() < 1) {
            return routers;
        }
        if (urls != null && urls.size() > 0) {
            for (URL url : urls) {
                if (Constants.EMPTY_PROTOCOL.equals(url.getProtocol())) {
                    continue;
                }
                String routerType = url.getParameter(Constants.ROUTER_KEY);
                if (routerType != null && routerType.length() > 0) {
                    url = url.setProtocol(routerType);
                }
                try {
                  //ConditionRouterFactory 来获取
                    Router router = routerFactory.getRouter(url);
                    if (!routers.contains(router))
                        routers.add(router);
                } catch (Throwable t) {
                    logger.error("convert router url to router error, url: " + url, t);
                }
            }
        }
        return routers;
    }
protected void setRouters(List<Router> routers) {
        // copy list
        routers = routers == null ? new ArrayList<Router>() : new ArrayList<Router>(routers);
        // append url router
        String routerkey = url.getParameter(Constants.ROUTER_KEY);
        if (routerkey != null && routerkey.length() > 0) {
            RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getExtension(routerkey);
            routers.add(routerFactory.getRouter(url));
        }
        // append mock invoker selector
        routers.add(new MockInvokersSelector());
        Collections.sort(routers);
        this.routers = routers;
    }

public class ConditionRouterFactory implements RouterFactory {
    public static final String NAME = "condition";
    public Router getRouter(URL url) {
        return new ConditionRouter(url);
    }
}
public ConditionRouter(URL url) {
        this.url = url;
  //对应dubbo-admin界面的优先级，默认为0
        this.priority = url.getParameter(Constants.PRIORITY_KEY, 0);
  //当路由结果为空时，是否强制执行，如果不强制执行，路由结果为空的路由规则将自动失效，可不填，缺省为flase。
        this.force = url.getParameter(Constants.FORCE_KEY, false);
        try {
            String rule = url.getParameterAndDecoded(Constants.RULE_KEY);
            if (rule == null || rule.trim().length() == 0) {
                throw new IllegalArgumentException("Illegal route rule!");
            }
            rule = rule.replace("consumer.", "").replace("provider.", "");
            int i = rule.indexOf("=>");//切开
            String whenRule = i < 0 ? null : rule.substring(0, i).trim();
            String thenRule = i < 0 ? rule.trim() : rule.substring(i + 2).trim();
            Map<String, MatchPair> when = StringUtils.isBlank(whenRule) || "true".equals(whenRule) ? new HashMap<String, MatchPair>() : parseRule(whenRule);
            Map<String, MatchPair> then = StringUtils.isBlank(thenRule) || "false".equals(thenRule) ? null : parseRule(thenRule);
            // NOTE: It should be determined on the business level whether the `When condition` can be empty or not.
            this.whenCondition = when;
            this.thenCondition = then;
        } catch (ParseException e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

其中如何理解ConditionRouter处理流程呢？首先我们要知道基于条件表达式的规则

1	host = 10.20.153.10 => host = 10.20.153.11

规则：

“=>”之前的为消费者匹配条件，所有参数和消费者的URL进行对比，当消费者满足匹配条件时，对该消费者执行后面的过滤规则。
“=>”之后为提供者地址列表的过滤条件，所有参数和提供者的URL进行对比，消费者最终只拿到过滤后的地址列表。
如果匹配条件为空，表示对所有消费方应用，如：=> host != 10.20.153.11
如果过滤条件为空，表示禁止访问，如：host = 10.20.153.10 =>

白名单：host != 10.20.153.10,10.20.153.11 =>

黑名单：host = 10.20.153.10,10.20.153.11 =>

表达式：

参数支持：
服务调用信息，如：method, argument 等 (暂不支持参数路由)
URL本身的字段，如：protocol, host, port 等
以及URL上的所有参数，如：application, organization 等

条件支持：
等号”=”表示”匹配”，如：host = 10.20.153.10
不等号”!=”表示”不匹配”，如：host != 10.20.153.10

值支持：
以逗号”,”分隔多个值，如：host != 10.20.153.10,10.20.153.11
以星号“”结尾，表示通配，如：host != 10.20.
以美元符开头，表示引用消费者参数，如：host = $host

然后到了refreshInvoker(List invokerUrls) ，在这里面有toMethodInvokers方法，主要作用是将invokers列表转成与方法的映射关系，用到route方法

private List<Invoker<T>> route(List<Invoker<T>> invokers, String method) {
        Invocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
        List<Router> routers = getRouters();
        if (routers != null) {
            for (Router router : routers) {
                if (router.getUrl() != null) {
                    invokers = router.route(invokers, getConsumerUrl(), invocation);//这里是ConditionRouter实现的。
                }
            }
        }
        return invokers;
    }

//注意调用时候，这个url是 getConsumerUrl()，即消费者url
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation)throws RpcException {
        if (invokers == null || invokers.size() == 0) {
            return invokers;
        }
        try {
          //消费者与配置不匹配（配置为空，也意味着不匹配），就不用走后面提供者的过滤了
            if (!matchWhen(url, invocation)) {
                return invokers;
            }
            List<Invoker<T>> result = new ArrayList<Invoker<T>>();
          //而提供者没有配置，则是黑名单
            if (thenCondition == null) {
                logger.warn("The current consumer in the service blacklist. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey());
                return result;
            }
            for (Invoker<T> invoker : invokers) {
              //提供者过滤（消费者调用的url与提供者url匹配，就加入）
                if (matchThen(invoker.getUrl(), url)) {
                    result.add(invoker);
                }
            }
            if (result.size() > 0) {
                return result;
            } else if (force) {
                logger.warn("The route result is empty and force execute. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey() + ", router: " + url.getParameterAndDecoded(Constants.RULE_KEY));
                return result;
            }
        } catch (Throwable t) {
            logger.error("Failed to execute condition router rule: " + getUrl() + ", invokers: " + invokers + ", cause: " + t.getMessage(), t);
        }
  //其他情况就返回所有的，也就是说针对条件路由，当经过某条路由规则路由后，没有一个符合规则的Provider，那么此次路由失败，会直接返回路由本条规则前的所有Provider，也就是相当于没有经过该路由的结果。
        return invokers;
    }

参考

http://www.bubuko.com/infodetail-2316000.html