由Dubbo路由引发的一言不合就看源码

Posted 百里马

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了由Dubbo路由引发的一言不合就看源码相关的知识,希望对你有一定的参考价值。

由Router引入

由于业务上的需要,很早之前就想自己扩展Dubbo的路由功能。But,之前试了一下,发现好像并没有找到正确的打开方式,而且由于时间不是很充足,加上自己心浮气躁浅尝辄止,于是乎就放弃了这块的研究。

不过今日,在手头上一块比较忙的事情告一段落了之后,又开始漫漫的research之路。。

好了,下面先来说说之前被困在了什么地方。

首先吐槽一下,官方文档对于这一块的用法写得非常的不全面,只介绍了条件路由的语法,而没有告诉我们怎么样使其生效。当然最简单的一种做法,官方文档上也说了,装个dubbo-admin在其中设置就好了。但是如果是想自定义扩展的话呢?这个时候我们来看看Developer Guide是怎么写的:

没错,就是这里的 (3)扩展配置 误导了我,导致了之前的放弃。因为这里的扩展配置,完全就是错的。今天看了源码之后,发现完全就没有这么一个配置项,关于正确的路由规则的设置方法下面会讲。

好了,继续看,那我们就先看看条件路由的语法,然而,又发现一个官方文档的bug:

官方文档写的是runtime不填缺省为false,而Dubbo的代码中呢?

有点儿醉!接着看吧;既然看了代码,我们就从整体上来看看Dubbo的结构,这里从Consumer入手,来看看它是怎么找到合适的Provider来调用的。

源码分析

注意,下面我们会摒弃MockGeneric等非主流程,精简代码来分析。

核心类ReferenceConfig

好啦,从Consumer入手,我们看看到底发生了什么,首先,Dubbo会代理我们在Consumer中引用的接口,一般都用<dubbo:reference />来配置,具体的代理逻辑在ReferenceConfig中:

    private T createProxy(Map<String, String> map) 
        // 通过注册中心配置拼装URL,这里省略了inJvm、点对点直连等
        List<URL> us = loadRegistries(false);
        if (us != null && us.size() > 0) 
            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)));
            
        
        // 这里的url是注册中心的url
        if (urls == null || urls.size() == 0) 
            throw new IllegalStateException("No such any registry to reference");
        

        // 这里省略了多个注册中心的情况
        if (urls.size() == 1) 
            // 我们通过注册中心生成了invoker,它代表一个可执行体,可向它发起invoke调用
            invoker = refprotocol.refer(interfaceClass, urls.get(0));
            // 这里生成invoker还包含了两个额外的步骤 1.consumer注册到registry 2.consumer订阅该服务下的providers/routers/configurators
            // 另外这个invoker里包含了很多逻辑,比如cluster等
        

        Boolean c = check;
        if (c == null && consumer != null) 
            c = consumer.isCheck();
        
        if (c == null) 
            c = true; // default true
        
        if (c && !invoker.isAvailable()) 
            throw new IllegalStateException("Failed to check the status of the service");
        
        if (logger.isInfoEnabled()) 
            logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
        
        // 创建服务代理
        return (T) proxyFactory.getProxy(invoker);
    

默认proxyFactoryStubProxyFactoryWrapper。从名字也能看出来,这个类里实现了客户端Stub的逻辑。不过这次我们就先跳过这一段,以后再来看。最终代理的实现无非还是JDK的动态代理或是Javassist的代理,我们比较关心InvocationHandler——InvokerInvocationHandler

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable 
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getDeclaringClass() == Object.class) 
            return method.invoke(invoker, args);
        
        if ("toString".equals(methodName) && parameterTypes.length == 0) 
            return invoker.toString();
        
        if ("hashCode".equals(methodName) && parameterTypes.length == 0) 
            return invoker.hashCode();
        
        if ("equals".equals(methodName) && parameterTypes.length == 1) 
            return invoker.equals(args[0]);
        
        return invoker.invoke(new RpcInvocation(method, args)).recreate();
    

其中RpcInvocation是会话域,它持有调用过程中的变量,比如方法名,参数等。

然后我们关注一下其中的成员变量invoker,最终的调用逻辑都是由invoker去执行。我们来看看这个invoker究竟是什么:

可以看到这里的invoker有点像装饰者模式,一层层的添加功能。需要注意的是,上述截图中包含MergeableClusterInvoker是因为我在reference上配置了group=*,而一般的流程则只会走:MockClusterInvoker -> FailoverClusterInvoker:其中 FailoverClusterInvoker 继承了 AbstractClusterInvoker,这个抽象类实现了 routers 的过滤 以及 loadBalance的过滤,最终会选取一个唯一的Provider来进行调用:

    public Result invoke(final Invocation invocation) throws RpcException 

        checkWhetherDestroyed();

        LoadBalance loadbalance;

        // 这里的list方法也是根据directory的list方法进行过滤,也就是走routers进行过滤
        List<Invoker<T>> invokers = list(invocation);
        if (invokers != null && invokers.size() > 0) 
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                    .getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
         else 
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
        
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    

核心类RegistryDirectory

这里提到了routers就不得不提到一个类——RegistryDirectory:每个ReferenceBean都会对应一个RegistryDirectory,其中维护了几乎所有关于该服务接口的配置:比如configuratorsroutersproviders等。
这个类实现了NotifyListener接口,显然每次当这些配置项有变化时都需要通知这个类做refresh
其中每次刷新都会刷新这些配置项,并且还有一些缓存,比如:某个方法可调用的Provider的对应关系(methodInvokerMap),简单看看这个notify方法:

    public synchronized void notify(List<URL> urls) 
        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();
            String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            if (Constants.ROUTERS_CATEGORY.equals(category) 
                    || Constants.ROUTE_PROTOCOL.equals(protocol)) 
                routerUrls.add(url);
             else if (Constants.CONFIGURATORS_CATEGORY.equals(category) 
                    || Constants.OVERRIDE_PROTOCOL.equals(protocol)) 
                configuratorUrls.add(url);
             else if (Constants.PROVIDERS_CATEGORY.equals(category)) 
                invokerUrls.add(url);
             else 
                logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
            
        
        // configurators 
        if (configuratorUrls != null && configuratorUrls.size() >0 )
            this.configurators = toConfigurators(configuratorUrls);
        
        // routers
        if (routerUrls != null && routerUrls.size() >0 )
            List<Router> routers = toRouters(routerUrls);
            if(routers != null) // null - do nothing
                setRouters(routers);
            
        
        List<Configurator> localConfigurators = this.configurators; // local reference
        // 合并override参数
        this.overrideDirectoryUrl = directoryUrl;
        if (localConfigurators != null && localConfigurators.size() > 0) 
            for (Configurator configurator : localConfigurators) 
                this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
            
        
        // providers
        refreshInvoker(invokerUrls);
    

这里会去设置routersconfigurators等成员变量,然后在refereshInvoker方法中,重新构建各种缓存。

我们可以考虑一下该notify的触发点,肯定是当routersconfigurators等信息有变化时才会触发。而这些信息都是存储在我们注册中心当中的,不得不提到Dubbo中最常见的注册中心——ZooKeeperRegistry

Registry

关于注册中心,在上面的分析中也提到过一次,不知道大家有没有注意到。就是在ReferenceConfigcreateProxy方法中,有一步:

        // 这里省略了多个注册中心的情况
        if (urls.size() == 1) 
            // 我们通过注册中心生成了invoker,它代表一个可执行体,可向它发起invoke调用
            invoker = refprotocol.refer(interfaceClass, urls.get(0));
            // 这里生成invoker还包含了两个额外的步骤 1.consumer注册到registry 2.consumer订阅该服务下的providers/routers/configurators
        

这里我们可以深入再看看,这里我们用RegistryProtocol来做分析:

    // 这里两个参数一个是服务类型,另一个是注册中心的url
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException 
        url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
        // Registry是注册中心的抽象,比如ZooKeeperRegistry
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) 
            return proxyFactory.getInvoker((T) registry, type, url);
        

        // group="a,b" or group="*"
        // 这里关于group一个和group多个会有不同的调用分支
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
        String group = qs.get(Constants.GROUP_KEY);
        if (group != null && group.length() > 0 ) 
            if ( ( Constants.COMMA_SPLIT_PATTERN.split( group ) ).length > 1
                    || "*".equals( group ) ) 
                return doRefer( getMergeableCluster(), registry, type, url );
            
        
        return doRefer(cluster, registry, type, url);
    

doRefer方法中会生成Consumer的URL,并且将其注册到注册中心:

    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) 
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
        if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
                && url.getParameter(Constants.REGISTER_KEY, true)) 
            // 注册到注册中心
            registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                    Constants.CHECK_KEY, String.valueOf(false)));
        

        // 订阅该service下的 providers,configurators,routers
        directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
                Constants.PROVIDERS_CATEGORY 
                + "," + Constants.CONFIGURATORS_CATEGORY 
                + "," + Constants.ROUTERS_CATEGORY));
        return cluster.join(directory);
    

注册之后,就是消费者订阅,其中订阅了,一般都用ZK来做注册中心,利用ZK的watcher来做更新通知

   protected void doSubscribe(final URL url, final NotifyListener listener) 
        try 
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) 
                // 省略部分代码
             else 
                List<URL> urls = new ArrayList<URL>();
                // 这里toCategoriesPath(url)返回的是一个该服务下routers等zk路径
                for (String path : toCategoriesPath(url)) 
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                    if (listeners == null) 
                        zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                        listeners = zkListeners.get(url);
                    
                    ChildListener zkListener = listeners.get(listener);
                    if (zkListener == null) 
                        listeners.putIfAbsent(listener, new ChildListener() 
                            public void childChanged(String parentPath, List<String> currentChilds) 
                                // 这里会监听节点的变化
                                ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                            
                        );
                        zkListener = listeners.get(listener);
                    
                    zkClient.create(path, false);
                    // 这里会返回该zk路径下的所有数据,比如provider的话,就返回多个provider的URL
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) 
                        // 如果该path下的数据为空,那么返回empty://协议开头,暂时不知是何用意
                        // 注意:这里已经根据group等过滤过了
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    
                
                // 这里的listener也就是RegistryDirectory
                notify(url, listener, urls);
            
         catch (Throwable e) 
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        
    

订阅之后再执行一下notify,作为初始notify,之后每一次节点有变更,都会触发notify,而notify里的逻辑就是通知RegistryDirectory这个实例的内部配置的更新。其中第一个urlconsumerurl,而第三个参数是更新的URL列表,注意,下面的notify方法是Registry中的notify,其中参数listener才是RegistryDirectory

    protected void notify(URL url, NotifyListener listener, List<URL> urls) 
        if (url == null) 
            throw new IllegalArgumentException("notify url == null");
        
        if (listener == null) 
            throw new IllegalArgumentException("notify listener == null");
        
        if ((urls == null || urls.size() == 0) 
                && ! Constants.ANY_VALUE.equals(url.getServiceInterface())) 
            logger.warn("Ignore empty notify urls for subscribe url " + url);
            return;
        
        if (logger.isInfoEnabled()) 
            logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
        
        Map<String, List<URL>> result = new HashMap<String, List<URL>>();
        for (URL u : urls) 
            if (UrlUtils.isMatch(url, u)) 
                String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
                List<URL> categoryList = result.get(category);
                if (categoryList == null) 
                    categoryList = new ArrayList<URL>();
                    result.put(category, categoryList);
                
                categoryList.add(u);
            
        
        if (result.size() == 0) 
            return;
        
        Map<String, List<URL>> categoryNotified = notified.get(url);
        if (categoryNotified == null) 
            notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
            categoryNotified = notified.get(url);
        
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) 
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            saveProperties(url);
            listener.notify(categoryList);
        
    

这里的urllistener应该是一一对应的,url代表一个consumer端的reference,而listener则代表该referenceRegistryDirectory,然后urls则代表多个配置项,这些配置项会一个个应用到RegistryDirectory上去。这里的urls根据category可以分成三类:routersconfiguratorsproviders。并且每个category只会执行一次notify,之后每次这三个category的值有变更,都会进行通知,然后执行notify方法。

我们还可以来看看UrlUtils.isMatch(url, u)这个方法:

   public static boolean isMatch(URL consumerUrl, URL providerUrl) 
        String consumerInterface = consumerUrl.getServiceInterface();
        String providerInterface = providerUrl.getServiceInterface();
        if( ! (Constants.ANY_VALUE.equals(consumerInterface) || StringUtils.isEquals(consumerInterface, providerInterface)) ) return false;

        if (! isMatchCategory(providerUrl.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY), 
                consumerUrl.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY))) 
            return false;
        
        if (! providerUrl.getParameter(Constants.ENABLED_KEY, true) 
                && ! Constants.ANY_VALUE.equals(consumerUrl.getParameter(Constants.ENABLED_KEY))) 
            return false;
        

        String consumerGroup = consumerUrl.getParameter(Constants.GROUP_KEY);
        String consumerVersion = consumerUrl.getParameter(Constants.VERSION_KEY);
        String consumerClassifier = consumerUrl.getParameter(Constants.CLASSIFIER_KEY, Constants.ANY_VALUE);

        String providerGroup = providerUrl.getParameter(Constants.GROUP_KEY);
        String providerVersion = providerUrl.getParameter(Constants.VERSION_KEY);
        String providerClassifier = providerUrl.getParameter(Constants.CLASSIFIER_KEY, Constants.ANY_VALUE);
        return (Constants.ANY_VALUE.equals(consumerGroup) || StringUtils.isEquals(consumerGroup, providerGroup) || StringUtils.isContains(consumerGroup, providerGroup))
               && (Constants.ANY_VALUE.equals(consumerVersion) || StringUtils.isEquals(consumerVersion, providerVersion))
               && (consumerClassifier == null || Constants.ANY_VALUE.equals(consumerClassifier) || StringUtils.isEquals(consumerClassifier, providerClassifier));
    

上面这段代码实现了groupversion等之间的隔离逻辑。

OK,总算看完了一整个从Consumer调用远程服务的过程,具体的通信细节这里就先不展开讨论了(已经累得半死了)。

重回Routers

前面在分析源码时提到了RegistryDirectory,路由逻辑就是在这个类中完成的:

    public List<Invoker<T>> list(Invocation invocation) throws RpcException 
        if (destroyed)
            throw new RpcException("Directory already destroyed .url: "+ getUrl());
        
        List<Invoker<T>> invokers = doList(invocation);
        List<Router> localRouters = this.routers; // local reference
        if (localRouters != null && localRouters.size() > 0) 
            for (Router router: localRouters)
                try 
                    if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) 
                        invokers = router.route(invokers, getConsumerUrl(), invocation);
                    
                 catch (Throwable t) 
                    logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
                
            
        
        return invokers;
    

我们来细看一下上面的代码:
第一步——doList,这一步实际上是直接获取 方法和ProviderList的缓存关系
第二步——很关键,这里需要提到一个参数,就是这个runtime。每一条route都会有一个runtime这个属性,如果runtime配置为true,那么客户端会在每次调用服务端方法时都进行route一下,因为比如说根据参数来进行route,当然这对性能会有一定的影响,但是比如条件路由的话,也就正则表达式的判断,性能影响应该不是特别大。而runtime配置为false的话,就可以进行缓存,也就是在第一次调用时确定下来,后面只有当route规则进行变更了之后才会更新该缓存,也就是能直接在第一步中获得。这里又要吐槽一下官方文档了,童鞋们还记得前面的梗么。。。

再一次,继续吐槽,Router中有一个BUG,ConditionRouter中有两个核心方法,一个是matchWhen,另一个是MatchThen。也就是先判断Consumer是否满足matchWhen,若满足,则路由到满足matchThen条件的Provider

其中matchWhen方法:

    private boolean matchCondition(Map<String, MatchPair> condition, URL url, URL param) 
        Map<String, String> sample = url.toMap();
        for (Map.Entry<String, String> entry : sample.entrySet()) 
            String key = entry.getKey();
            MatchPair pair = condition.get(key);
            if (pair != null && ! pair.isMatch(entry.getValue(), param)) 
                return false;
            
        
        return true;
    

而这个url.toMap();方法,得到的是ConsumerparametersMap集合。我们可以通过注册中心观察到ConsumerURL上是没有method的参数的,类似如下:

consumer://192.168.1.88/com.alibaba.dubbo.examples.merge.api.MergeService?application=merge-consumer&category=providers,configurators,routers&dubbo=2.0.0&interface=com.alibaba.dubbo.examples.merge.api.MergeService&&pid=11020&side=consumer&timestamp=1466450710697

所以返回的map里是没有method这个key的,取而代之的是methods,而methods的value是所有的方法名,比如:methods=mergeResult,testRouter

而官网上给我们的条件路由的配置里面有如下图所示,我又醉了。。。。

另外还有两点需要注意
1. 如果有多条routers规则,那么会根据每一条routers来过滤出可调用的provider列表
2. 针对条件路由,当经过某条路由规则路由后,没有一个符合规则的Provider,那么此次路由失败,会直接返回路由本条规则前的所有Provider,也就是相当于没有经过该路由的结果。

配置Routers

边看边分析边写,虽然精简了不少,但是还是多而乱,不知道是否有交代清楚。最后我们以实战配置一条routers来结束这篇文章。

场景

假设我们有多个服务分了不同的group,然后需要根据某个参数调用不同的group的方法

设计

先自定义一个Router,依据原有的ConditionRouterConditionRouterFactory创建出我们的CustomRouterCoustomRouterFactory,当然核心逻辑还是在CustomRouterroute方法中,也很简单,只是根据请求参数路由一下而已:

    public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException 
        Object[] arguments = invocation.getArguments();
        if(arguments == null || arguments.length == 0)
            logger.error("method:" + invocation.getMethodName() + " do not have route param");
            return null;
        
        Object routeKey = arguments[0];
        List<Invoker<T>> result = new ArrayList<Invoker<T>>();
        for (Invoker<T> invoker : invokers) 
            String group = invoker.getUrl().getParameter(Constants.GROUP_KEY);
            if ((getGroup(routeKey)).equals(group)) 
                result.add(invoker);
                return result;
            
        
        result.add(invokers.get(0));
        return result;
    

然后再让Dubbo自动发现我们添加的类:

custom=com.alibaba.dubbo.rpc.cluster.router.custom.CustomRouterFactory

最后再配置路由规则:

        RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class)
                .getAdaptiveExtension();
        Registry registry = registryFactory.getRegistry(URL.valueOf("zookeeper://127.0.0.1:2181"));
        registry.register(URL.valueOf(
                "routers://0.0.0.0/com.alibaba.dubbo.examples.merge.api.MergeService?name=test&category=routers&router=custom&dynamic=false"));

这段配置我加在了Provider启动的时候。好了,最后模拟Consumer调用测试一下,OK

写在最后

总算写完了。已经好久没有写Blog了,接下来希望自己可以再重新坚持起来!加油

以上是关于由Dubbo路由引发的一言不合就看源码的主要内容,如果未能解决你的问题,请参考以下文章

由Dubbo路由引发的一言不合就看源码

一言不合就开始搞JDK源码

Dubbo路由功能实现灰度发布及源码分析

从Consumer分析Dubbo调用链

从Consumer分析Dubbo调用链

Dubbo源码