dubbo路由机制代码分析1

这回说说,dubbo路由特性,dubbo的路由干的事,就是一个请求过来, dubbo依据配置的路由规则,计算出哪些提供者可以提供这次的请求服务。 所以,它的优先级是在集群容错策略和负载均衡策略之前的。 即先有路由规则遴选出符合条件的服务提供者 然后,再在这些服务提供者之中应用负载均衡,集群容错策略。 流程图大概是如下图:

可以通过代码验证上面的逻辑。 看AbstractClusterInvoker类invoke方法,服务调用的入口方法:

public Result invoke(final Invocation invocation) throws RpcException {

        checkWhetherDestroyed();

        LoadBalance loadbalance;
        //list方法
        //会调用directory的list方法。
        //其实是AbstractDirectory的list方法,这个方法里就是利用路由规则(如果有),从所有
        //提供者中,遴选出符合规则的提供者s.
        //接下里才是集群容错和负载均衡。
        //可以发现每次调用这些过程都重新下走一遍这个逻辑
        List<Invoker<T>> invokers = list(invocation);
        if (invokers != null && invokers.size() > 0) {
            //从url通过key "loadbalance" 取不到值,就取默认random随机策略
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                    .getMethodParameter(invocation.getMethodName(), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
        } else {
            //取默认random随机策略
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
        }
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    }

list方法的具体实现,在AbstractDirectory类里

/***
     * 落地路由规则
     * @param invocation
     * @return
     * @throws RpcException
     */
    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);
        //本地路由规则,这个其实已通过setRouters方法设置好。什么时候设置的,稍后看看
        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
                        invokers = router.route(invokers, getConsumerUrl(), invocation);
                    }
                } catch (Throwable t) {
                    logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
                }
            }
        }
        return invokers;
    }

路由器在哪里设置的?还在AbstractDirectory类里setRouters方法

/***
     * 设置路由器
     * @param routers
     */
    protected void setRouters(List<Router> routers) {
        // copy list
        routers = routers == null ? new ArrayList<Router>() : new ArrayList<Router>(routers);
        //append url router
        //获取路由key
	//这里感觉是从consumer端配置router的路由规则,但是consumer没有router配置的地方???
	//官方文档给出<dubbo:protocol router="xxx" />配置方式,但是并没有实现。
        String routerkey = url.getParameter(Constants.ROUTER_KEY);
        if (routerkey != null && routerkey.length() > 0) {
           //根据routerkey获取三种路由工厂的其中一种
            RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getExtension(routerkey);
           //再由路由工厂获取具体类型路由器
            routers.add(routerFactory.getRouter(url));
        }
        // append mock invoker selector
        // 追加mock invoker 路由器,
        routers.add(new MockInvokersSelector());
	//排序后MockInvokersSelector会放在最后
	//MockInvokersSelector路由器,是dubbo对mock调用支持的一部分,稍后看下源码
        Collections.sort(routers);
        this.routers = routers;
    }

setRouters在哪里调用的?在AbstractDirectory子类。 RegistryDirectory类,可以看到notify里有,notify是注册中心通知consumer回调的方法

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);
    }

   这样就保证配置可以从注册中心下发到调用方。    可以看到,这里有个 toRouters(routerUrls)方法把路由配置转换成路由器实例

   /**
     * 把路由配置转化成具体路由器
     * @param urls
     * @return null : no routers ,do nothing
     * else :routers list
     */
    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的protocol为路由类型
                    url = url.setProtocol(routerType);
                }
                try {
                    //根据routerType获取对应的路由器工厂,然后获取具体路由器。然后放入路由器集合
		    //dubbo默认实现有file,script,condition三种类型工厂对应三种路由类型
                    //这些都是在RouterFactory$Adpative类完成适配的,RouterFactory$Adpative类是dubbo spi机制动态编码,编译生成的
                    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;
    }

下一次,可以说说,三种路由的具体实现和配置方法。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏文武兼修ing——机器学习与IC设计

队列及其实现队列队列的实现

队列 队列即FIFO,一言以蔽之就是先进先出。比如入队列的顺序是1,2,3,4,那么出队列的顺序也是1,2,3,4 队列的实现 软件——GO语言实现 除了使用链...

42670
来自专栏逸鹏说道

分享一个自制的 .net线程池2

/// <summary> /// 该方法必须在 locked 下执行 /// </summary> ...

25050
来自专栏進无尽的文章

编码篇-数据管理者Model

      Model是数据管理者和持有者,是数据解析层剥离ViewConyroller的关键所在。同是也是cell滑动不卡(省去每次解析)的好方式。

10730
来自专栏Java技术栈

Java程序员注意:Tomcat Get请求的巨坑!

3.4K20
来自专栏崔庆才的专栏

腾讯云上 Winpcap 网络编程四之主机通信

由于腾讯云上提供了Windows系统,所以我们这次Winpcap编程选用腾讯云主机实验,让大家简要了解两台云主机的通信方法以及实践过程。

58300
来自专栏微信公众号:Java团长

从并发编程到分布式系统——如何处理海量数据(上)

在这里想写写自己在学习并发处理的学习思路,也会聊聊自己遇到的那些坑,以此为记,希望鞭策自己不断学习、永不放弃!

9810
来自专栏Java编程技术

常用开源框架中设计模式使用分析(全)

说起来设计模式,大家应该都耳熟能详,设计模式代表了软件设计的最佳实践,是经过不断总结提炼出来的代码设计经验的分类总结,这些模式或者可以简化代码,或者可以是代码逻...

17820
来自专栏博客园

Asp.Net Web API(四)

    如果Web API控制器抛出一个未捕捉的异常,会发生什么呢?在默认情况下,大多数异常都会转换为一个带有状态码500的内部服务器错误的HTTP响应。

20020
来自专栏MasiMaro 的技术博文

Vista 及后续版本的新线程池

在上一篇的博文中,说了下老版本的线程池,在Vista之后,微软重新设计了一套线程池机制,并引入一组新的线程池API,新版线程池相对于老版本的来说,它的可控性更高...

16630
来自专栏coolblog.xyz技术专栏

Dubbo 源码分析 - 集群容错之 Cluster

为了避免单点故障,现在的应用至少会部署在两台服务器上。对于一些负载比较高的服务,会部署更多台服务器。这样,同一环境下的服务提供者数量会大于1。对于服务消费者来说...

12820

扫码关注云+社区

领取腾讯云代金券