dubbo剖析：二服务引用

2018-04-01 本文已影响452人益文的圈

注：文章中使用的dubbo源码版本为2.5.4

零、文章目录

一、服务引用的目的
二、关键概念及关系
三、服务引用流程详解
四、整体流程图总结
五、后续系列文章预告

一、服务引用的目的

“服务消费者”向注册中心订阅“服务提供者”提供的服务地址，并生成服务接口的实际代理对象。

二、关键概念及关系

2.1 ReferenceBean & ReferenceConfig

服务引用Bean和引用配置类，服务引用信息的承载体。

2.2 RegistryProtocol

协议抽象接口Protocol的实现。其<T> Invoker<T> refer(Class<T> type, URL url)方法完成了服务引用的完整功能。

2.3 RegistryDirectory

同时具有目录服务Directory和监听器NotifyListener两个接口的功能，并组合了注册器Registry。

它的Invoker集合数据来源于zk注册中心；
其对外暴露的subscribe(URL url)方法完成了在注册中心对服务提供者地址变更的监听功能；
其同时实现了NotifyListener接口的notify(List<URL> urls)方法执行监听回调，监听服务提供者地址变更并创建对应的Invoker；

2.4 JavassistProxyFactory

基于javassist的动态代理工厂，其内部使用dubbo自己的动态代理机制Proxy完成动态代理类的创建工作。

2.5 关系图

服务引用相关接口关系图

三、服务引用流程详解

3.1 简洁流程图

引用配置及初始化
配置初始化
注册中心订阅 & Invoker生成
Invoker获取
生成代理对象
简洁流程图

3.2 服务引用入口

引用配置：

服务引用入口
服务引用方在工程中会有如上图中的Spring配置。
在容器启动的时候会解析schema元素<dubbo:reference/>转换成dubbo内部数据结构ReferenceBean。
ReferenceBean就是服务引用，它继承了dubbo的引用配置类ReferenceConfig，并实现了Spring的FactoryBean和InitializingBean接口。

对于服务引用获取的demoService，并不是想要ReferenceBean这个对象实例本身，而是想获取对远程调用的代理，能够通过这个代理服务调用远程dubbo服务；
dubbo就是通过factoryBean.getObject()来创建并返回基于DemoService接口的代理对象；

配置初始化：

    public Object getObject() throws Exception {
        return get();
    }

ReferenceBean的getObject方法执行其get()方法，最终会执行到ReferenceConfig的init()方法，在该方法中完成代理对象的生成过程。

    private void init() {
        /*...省略部分代码...*/
        //step1 根据配置生成Map
        Map<String, String> map = new HashMap<String, String>();
        Map<Object, Object> attributes = new HashMap<Object, Object>();
        map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
        map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
        map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));

        /*...省略部分代码...*/
        //step2 调用createProxy生成代理类ref
        ref = createProxy(map);
    }

如上，init()中主要干的事情有两件。一，根据配置生成map；二，调用createProxy生成代理对象ref。

下图展示了生成的map的信息：

map的信息结构
createProxy()方法主要做了如下三件事：1）加载配置中心拼装成urls；2）遍历urls，调用refProtocol创建远程的动态代理Invoker；3）调用proxyFactory创建服务代理。

    private T createProxy(Map<String, String> map) {
        /*...省略部分代码...*/
        //step1 加载配置的所有注册中心，拼装成urls
                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)));
                    }
                }
        /*...省略部分代码...*/
        //step2 遍历urls，调用refProtocol.refer创建远程的动态代理Invoker
                for (URL url : urls) {
                    invokers.add(refprotocol.refer(interfaceClass, url));
                    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                        registryURL = url; // 用了最后一个registry url
                    }
                }
                if (registryURL != null) { // 有 注册中心协议的URL
                    // 对有注册中心的Cluster 只用 AvailableCluster
                    URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
                    invoker = cluster.join(new StaticDirectory(u, invokers));
                } else { // 不是 注册中心的URL
                    invoker = cluster.join(new StaticDirectory(invokers));
                }
        /*...省略部分代码...*/
        // step3 调用proxyFactory创建服务代理
        return (T) proxyFactory.getProxy(invoker);

3.3 注册中心订阅及Invoker创建

创建动态代理Invoker：
refProtocol为dubbo通过ExtensionLoader动态注入的RegistryProtocol实例。通过其refer方法创建Invoker。
具体进行的工作分为如下几步：

从url的registryKey获取注册中心类型：zookeeper
从RegistryFactory获取注册器：Registry
构建RegistryDirectory，其同时也是回调监听器
向注册中心注册服务消费者
从注册中心订阅服务提供者（即引用的服务）

    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        //step1 从url的registryKey获取注册中心类型：zookeeper
        url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
        //step2 从RegistryFactory获取注册器
        Registry registry = registryFactory.getRegistry(url);
        //...省略部分代码...
        //doRefer
        return doRefer(cluster, registry, type, url);
    }

    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {

        //step3 构建RegistryDirectory,可以把它理解为注册资源,其中包含了消费者/服务/路由等相关信息
        //其同时也是回调监听器
        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)) {

            //step4 向注册中心注册服务消费者
            registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                    Constants.CHECK_KEY, String.valueOf(false)));
        }

        //step5 从注册中心订阅服务提供者（即引用的服务）
        directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
                Constants.PROVIDERS_CATEGORY
                        + "," + Constants.CONFIGURATORS_CATEGORY
                        + "," + Constants.ROUTERS_CATEGORY));
        //step6 
        return cluster.join(directory);
    }

订阅服务提供者并初始化监听处理：

RegistryDirectory

先按照上图对RegistryDirectory进行说明：

RegistryDirectory内包含了一个Registry实例，即其提供了subscribe方法用来实现对url的订阅；
RegistryDirectory同时实现了NotifyListener接口，即其本身具有回调监听器的功能；
以上结合，RegistryDirectory可以完成对服务引用url的主动订阅和监听，监听会触发其自身notify方法；

所以，执行到"step5 从注册中心订阅服务提供者"，调用RegistryDirectory的subscribe(URL url)方法时，完成了对引用url的订阅并同时出发监听流程：

根据类别将监听urls分类为invoker,router,configurator；
对于有变化的invokerUrls，调用DubboProtocol的refer方法生成DubboInvoker；
DubboInvoker即实际远程代理Invoker，其可完成请求远程dubbo服务并获取响应结果的功能；（具体DubboProtocol的refer方法执行过程不展开，后续有专门章节讲解）
最终刷新urlInvokerMap和methodInvokerMap，将对应的invokerUrl和dubboInvoker存入map中供后续使用；

    public synchronized void notify(List<URL> urls) {
        //step1 根据类别将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());
            }
        }
        //...省略部分代码...
        //providers
        refreshInvoker(invokerUrls);
    }

    private void refreshInvoker(List<URL> invokerUrls) {
           //...省略部分代码...
           Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// 将URL列表转成Invoker列表
           //...省略部分代码...
           //step3 最终刷新urlInvokerMap和methodInvokerMap
           this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
           this.urlInvokerMap = newUrlInvokerMap;
    }

    private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
           //...省略部分代码...
           //step2 protocol.refer根据url创建远程代理Invoker
         invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
    }

3.4 Invoker获取及集群

Invoker生成并返回：
执行到return cluster.join(directory)，即通过FailFastCluster获取并返回远程代理Invoker：

public class FailfastCluster implements Cluster {
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        return new FailfastClusterInvoker<T>(directory);
    }
}

其实就是new了一个FailfastClusterInvoker并返回，构造方法参数为3.3步骤中的RegistryDirectory。

FailfastClusterInvoker干了神马：
当invoke(Invocation invocation)方法被调用时:

根据参数invocation从RegistryDirectory中获取对应远程DubboInvocation；
动态创建负载均衡算法；
根据负载均衡算法从Invoker列表中选出一个invoker执行；

    public Result invoke(final Invocation invocation) throws RpcException {
        LoadBalance loadbalance;
        //step1 根据服务引用的url从RegistryDirectory中获取对应远程DubboInvocation
        List<Invoker<T>> invokers = list(invocation);
        if (invokers != null && invokers.size() > 0) {
            //step2 动态创建负载均衡算法
            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);
        //step3 根据负载均衡算法从Invoker列表中选出一个invoker执行
        return doInvoke(invocation, invokers, loadbalance);
    }

    protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        List<Invoker<T>> invokers = directory.list(invocation);
        return invokers;
    }

3.5 代理对象的生成

最终回到ReferenceConfig中init的最后一步return (T) proxyFactory.getProxy(invoker);，通过JavassistProxyFactory.getProxy(invoker)将Invoker转换成代理对象ref：

先从invoker中解析出要创建的代理对象所实现的接口；
new一个InvokerInvocationHandler，它包含了invoker，实际也会执行invoker的invoke方法；
Proxy.getProxy.newInstance
使用dubbo的动态代理接口创建动态代理对象T ref；

public class JavassistProxyFactory extends AbstractProxyFactory {
    @SuppressWarnings("unchecked")
    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    }
}

执行完以上步骤后，终于生成了dubbo远程动态代理对象T ref，我们可以直接使用该对象来完成RPC调用了。

四、整体流程图总结

整体流程图

整体看下来，服务引用主要分为以下几大步骤：

1、引用配置及配置初始化
该部分以Spring配置及ReferenceBean为入口，主要在ReferenceConfig中进行。

ReferenceConfig依赖RegistryProtocol完成了 "服务引用者注册"、"服务提供者订阅"和"Invoker创建" 的工作；
ReferenceConfig依赖JavassistProxyFactory完成了 "代理对象生成" 的工作；

2、注册中心订阅 & Invoker生成与获取
该部分主要由RegistryDirectory和FailfastCluster实现。

通过ReferenceConfig调用RegistryDirectory的subscribe方法，触发了对服务提供者url的订阅及监听，在监听过程中RegistryDirectory借助DubboProtocol完成了Invoker的创建工作，并保存了服务引用url和Invoker的关系；
通过ReferenceConfig调用FailfastCluster的join方法，完成了对Invoker对象的获取；

3、生成代理对象
该部分主要由JavassistProxyFactory完成。

以ReferenceConfig调用JavassistProxyFactory的getProxy方法为入口，传入Invoker；
新创建了InvokerInvocationHandler，并使用dubbo自己的动态代理工具Proxy最终生成代理对象T ref；

五、服务发布引用简洁过程图

简洁过程图

黑线表示方法调用
虚线表示对象引用
红线表示请求调用过程

dubbo剖析：二服务引用

零、文章目录

一、服务引用的目的

二、关键概念及关系

2.1 ReferenceBean & ReferenceConfig

2.2 RegistryProtocol

2.3 RegistryDirectory

2.4 JavassistProxyFactory

2.5 关系图

三、服务引用流程详解

3.1 简洁流程图

3.2 服务引用入口

3.3 注册中心订阅及Invoker创建

3.4 Invoker获取及集群

3.5 代理对象的生成

四、整体流程图总结

五、服务发布引用简洁过程图

猜你喜欢

热点阅读

dubbo剖析：二 服务引用

零、文章目录

一、服务引用的目的

二、关键概念及关系

2.1 ReferenceBean & ReferenceConfig

2.2 RegistryProtocol

2.3 RegistryDirectory

2.4 JavassistProxyFactory

2.5 关系图

三、服务引用流程详解

3.1 简洁流程图

3.2 服务引用入口

3.3 注册中心订阅及Invoker创建

3.4 Invoker获取及集群

3.5 代理对象的生成

四、整体流程图总结

五、服务发布引用简洁过程图

猜你喜欢

热点阅读

dubbo剖析：二服务引用