Dubbo消费端(consumer)源码分析(一)

半兽人 发表于: 2018-01-08   最后更新时间: 2018-01-12 16:33:55  
{{totalSubscript}} 订阅, 5,072 游览

Dubbo代码中,有两个很重要的对象就是Invoker和Exporter,Dubbo会根据用户配置的协议调用不同协议的Invoker,再通过ReferenceConfig将Invoker的引用关联到Reference的ref属性上提供给消费端调用。当用户调用一个Service接口的一个方法后由于Dubbo使用javassist动态代理,会调用Invoker的Invoke方法从而初始化一个RPC调用访问请求访问服务端的Service返回结果。下面我们就从Comsumer端开始逐步解析这个框架。
screenshot

Dubbo首先使用com.alibaba.dubbo.config.spring.schema.NamespaceHandler注册解析器,当spring解析xml配置文件时就会调用这些解析器生成对应的BeanDefinition交给spring管理:

public class DubboNamespaceHandler extends NamespaceHandlerSupport {  

    static {  
        Version.checkDuplicate(DubboNamespaceHandler.class);  
    }  

    public void init() {  
        //配置<dubbo:application>标签解析器  
        registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));  
        //配置<dubbo:module>标签解析器  
        registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));  
        //配置<dubbo:registry>标签解析器  
        registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));  
        //配置<dubbo:monitor>标签解析器  
        registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));  
        //配置<dubbo:provider>标签解析器  
        registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));  
        //配置<dubbo:consumer>标签解析器  
        registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));  
        //配置<dubbo:protocol>标签解析器  
        registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));  
        //配置<dubbo:service>标签解析器  
        registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));  
        //配置<dubbo:refenrence>标签解析器  
        registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));  
        //配置<dubbo:annotation>标签解析器  
        registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));  
    }  
}

Spring在初始化IOC容器时会利用这里注册的BeanDefinitionParser的parse方法获取对应的ReferenceBean的BeanDefinition实例,由于ReferenceBean实现了InitializingBean接口,在设置了bean的所有属性后会调用afterPropertiesSet方法:

public void afterPropertiesSet() throws Exception {  
    //如果Consumer还未注册  
    if (getConsumer() == null) {  
        //获取applicationContext这个IOC容器实例中的所有ConsumerConfig  
        Map<String, ConsumerConfig> consumerConfigMap = applicationContext == null ? null  : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConsumerConfig.class, false, false);  
        //如果IOC容器中存在这样的ConsumerConfig  
        if (consumerConfigMap != null && consumerConfigMap.size() > 0) {  
            ConsumerConfig consumerConfig = null;  
            //遍历这些ConsumerConfig  
            for (ConsumerConfig config : consumerConfigMap.values()) {  
                //如果用户没配置Consumer系统会生成一个默认Consumer,且它的isDefault返回ture  
                //这里是说要么是Consumer是默认的要么是用户配置的Consumer并且没设置isDefault属性  
                if (config.isDefault() == null || config.isDefault().booleanValue()) {  
                    //防止存在两个默认Consumer  
                    if (consumerConfig != null) {  
                        throw new IllegalStateException("Duplicate consumer configs: " + consumerConfig + " and " + config);  
                    }  
                    //获取默认Consumer  
                    consumerConfig = config;  
                }  
            }  
            if (consumerConfig != null) {  
                //设置默认Consumer  
                setConsumer(consumerConfig);  
            }  
        }  
    }  
    //如果reference未绑定application且(reference未绑定consumer或referenc绑定的consumer没绑定application  
    if (getApplication() == null  
            && (getConsumer() == null || getConsumer().getApplication() == null)) {  
        //获取IOC中所有application的实例  
        Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);  
        if (applicationConfigMap != null && applicationConfigMap.size() > 0) {  
            //如果IOC中存在application  
            ApplicationConfig applicationConfig = null;  
            //遍历这些application  
            for (ApplicationConfig config : applicationConfigMap.values()) {  
                //如果application是默认创建或者被指定成默认  
                if (config.isDefault() == null || config.isDefault().booleanValue()) {  
                    if (applicationConfig != null) {  
                        throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);  
                    }  
                    //获取application  
                    applicationConfig = config;  
                }  
            }  
            if (applicationConfig != null) {  
                //关联到reference  
                setApplication(applicationConfig);  
            }  
        }  
    }  
    //如果reference未绑定module且(reference未绑定consumer或referenc绑定的consumer没绑定module  
    if (getModule() == null  
            && (getConsumer() == null || getConsumer().getModule() == null)) {  
        //获取IOC中所有module的实例  
        Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);  
        if (moduleConfigMap != null && moduleConfigMap.size() > 0) {  
            ModuleConfig moduleConfig = null;  
          //遍历这些module  
            for (ModuleConfig config : moduleConfigMap.values()) {  
                //如果module是默认创建或者被指定成默认  
                if (config.isDefault() == null || config.isDefault().booleanValue()) {  
                    if (moduleConfig != null) {  
                        throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);  
                    }  
                  //获取module  
                    moduleConfig = config;  
                }  
            }  
            if (moduleConfig != null) {  
                //关联到reference  
                setModule(moduleConfig);  
            }  
        }  
    }  
    //如果reference未绑定注册中心(Register)且(reference未绑定consumer或referenc绑定的consumer没绑定注册中心(Register)  
    if ((getRegistries() == null || getRegistries().size() == 0)  
            && (getConsumer() == null || getConsumer().getRegistries() == null || getConsumer().getRegistries().size() == 0)  
            && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().size() == 0)) {  
        //获取IOC中所有的注册中心(Register)实例  
        Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);  
        if (registryConfigMap != null && registryConfigMap.size() > 0) {  
            List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();  
            //遍历这些registry  
            for (RegistryConfig config : registryConfigMap.values()) {  
                //如果registry是默认创建或者被指定成默认  
                if (config.isDefault() == null || config.isDefault().booleanValue()) {  
                    registryConfigs.add(config);  
                }  
            }  

            if (registryConfigs != null && registryConfigs.size() > 0) {  
                //关联到reference,此处可以看出一个consumer可以绑定多个registry(注册中心)  
                super.setRegistries(registryConfigs);  
            }  
        }  
    }  
  //如果reference未绑定监控中心(Monitor)且(reference未绑定consumer或reference绑定的consumer没绑定监控中心(Monitor)  
    if (getMonitor() == null  
            && (getConsumer() == null || getConsumer().getMonitor() == null)  
            && (getApplication() == null || getApplication().getMonitor() == null)) {  
        //获取IOC中所有的监控中心(Monitor)实例  
        Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);  
        if (monitorConfigMap != null && monitorConfigMap.size() > 0) {  
            MonitorConfig monitorConfig = null;  
            //遍历这些监控中心(Monitor)  
            for (MonitorConfig config : monitorConfigMap.values()) {  
                //如果monitor是默认创建或者被指定成默认  
                if (config.isDefault() == null || config.isDefault().booleanValue()) {  
                    if (monitorConfig != null) {  
                        throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);  
                    }  
                    monitorConfig = config;  
                }  
            }  
            if (monitorConfig != null) {  
                //关联到reference,一个consumer绑定到一个监控中心(monitor)  
                setMonitor(monitorConfig);  
            }  
        }  
    }  
    Boolean b = isInit();  
    if (b == null && getConsumer() != null) {  
        b = getConsumer().isInit();  
    }  
    if (b != null && b.booleanValue()) {  
        //如果consumer已经被关联则组装Reference  
        getObject();  
    }  
}  
}

这步其实是Reference确认生成Invoker所需要的组件是否已经准备好,都准备好后我们进入生成Invoker的部分。这里的getObject会调用父类ReferenceConfig的init方法完成组装:

private void init() {  
        //避免重复初始化  
        if (initialized) {  
            return;  
        }  
        //置为已经初始化  
        initialized = true;  
        //如果interfaceName不存在  
        if (interfaceName == null || interfaceName.length() == 0) {  
            throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");  
        }  
        // 获取消费者  
        checkDefault();  
        appendProperties(this);  
        //如果未使用泛接口并且consumer已经准备好的情况下,reference使用和consumer一样的泛接口  
        if (getGeneric() == null && getConsumer() != null) {  
            setGeneric(getConsumer().getGeneric());  
        }  
        //如果是泛接口那么interface的类型是GenericService  
        if (ProtocolUtils.isGeneric(getGeneric())) {  
            interfaceClass = GenericService.class;  
        } else {  
            //如果不是泛接口使用interfaceName指定的泛接口  
            try {  
                interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()  
                        .getContextClassLoader());  
            } catch (ClassNotFoundException e) {  
                throw new IllegalStateException(e.getMessage(), e);  
            }  
            //检查接口以及接口中的方法都是否配置齐全  
            checkInterfaceAndMethods(interfaceClass, methods);  
        }  
        //如果服务比较多可以指定dubbo-resolve.properties文件配置service(service集中配置文件)  
        String resolve = System.getProperty(interfaceName);  
        String resolveFile = null;  
        if (resolve == null || resolve.length() == 0) {  
            resolveFile = System.getProperty("dubbo.resolve.file");  
            if (resolveFile == null || resolveFile.length() == 0) {  
                File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");  
                if (userResolveFile.exists()) {  
                    resolveFile = userResolveFile.getAbsolutePath();  
                }  
            }  
            if (resolveFile != null && resolveFile.length() > 0) {  
                Properties properties = new Properties();  
                FileInputStream fis = null;  
                try {  
                    fis = new FileInputStream(new File(resolveFile));  
                    properties.load(fis);  
                } catch (IOException e) {  
                    throw new IllegalStateException("Unload " + resolveFile + ", cause: " + e.getMessage(), e);  
                } finally {  
                    try {  
                        if(null != fis) fis.close();  
                    } catch (IOException e) {  
                        logger.warn(e.getMessage(), e);  
                    }  
                }  
                resolve = properties.getProperty(interfaceName);  
            }  
        }  
        if (resolve != null && resolve.length() > 0) {  
            url = resolve;  
            if (logger.isWarnEnabled()) {  
                if (resolveFile != null && resolveFile.length() > 0) {  
                    logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service.");  
                } else {  
                    logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service.");  
                }  
            }  
        }  
        //如果application、module、registries、monitor未配置则使用consumer的  
        if (consumer != null) {  
            if (application == null) {  
                application = consumer.getApplication();  
            }  
            if (module == null) {  
                module = consumer.getModule();  
            }  
            if (registries == null) {  
                registries = consumer.getRegistries();  
            }  
            if (monitor == null) {  
                monitor = consumer.getMonitor();  
            }  
        }  
        //如果module已关联则关联module的registries和monitor  
        if (module != null) {  
            if (registries == null) {  
                registries = module.getRegistries();  
            }  
            if (monitor == null) {  
                monitor = module.getMonitor();  
            }  
        }  
      //如果application已关联则关联application的registries和monitor  
        if (application != null) {  
            if (registries == null) {  
                registries = application.getRegistries();  
            }  
            if (monitor == null) {  
                monitor = application.getMonitor();  
            }  
        }  
        //检查application  
        checkApplication();  
        //检查远端和本地服务接口真实存在(是否可load)  
        checkStubAndMock(interfaceClass);  
        Map<String, String> map = new HashMap<String, String>();  
        //配置dubbo的端属性(是consumer还是provider)、版本属性、创建时间、进程号  
        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()));  
        if (ConfigUtils.getPid() > 0) {  
            map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));  
        }  
        if (! isGeneric()) {  
            String revision = Version.getVersion(interfaceClass, version);  
            if (revision != null && revision.length() > 0) {  
                map.put("revision", revision);  
            }  

            String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();  
            if(methods.length == 0) {  
                logger.warn("NO method found in service interface " + interfaceClass.getName());  
                map.put("methods", Constants.ANY_VALUE);  
            }  
            else {  
                map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));  
            }  
        }  
        map.put(Constants.INTERFACE_KEY, interfaceName);  
        //调用application、module、consumer的get方法将属性设置到map中  
        appendParameters(map, application);  
        appendParameters(map, module);  
        appendParameters(map, consumer, Constants.DEFAULT_KEY);  
        appendParameters(map, this);  
        String prifix = StringUtils.getServiceKey(map);  
        if (methods != null && methods.size() > 0) {  
            for (MethodConfig method : methods) {  
                appendParameters(map, method, method.getName());  
                String retryKey = method.getName() + ".retry";  
                if (map.containsKey(retryKey)) {  
                    String retryValue = map.remove(retryKey);  
                    if ("false".equals(retryValue)) {  
                        map.put(method.getName() + ".retries", "0");  
                    }  
                }  
                appendAttributes(attributes, method, prifix + "." + method.getName());  
                checkAndConvertImplicitConfig(method, map, attributes);  
            }  
        }  
        //attributes通过系统context进行存储.  
        StaticContext.getSystemContext().putAll(attributes);  
        //在map装载了application、module、consumer、reference的所有属性信息后创建代理  
        ref = createProxy(map);  
    }
private T createProxy(Map<String, String> map) {  
        URL tmpUrl = new URL("temp", "localhost", 0, map);  
        final boolean isJvmRefer;  
        if (isInjvm() == null) {  
            if (url != null && url.length() > 0) { //指定URL的情况下,不做本地引用  
                isJvmRefer = false;  
            } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {  
                //默认情况下如果本地有服务暴露,则引用本地服务.  
                isJvmRefer = true;  
            } else {  
                isJvmRefer = false;  
            }  
        } else {  
            isJvmRefer = isInjvm().booleanValue();  
        }  

        if (isJvmRefer) {  
            URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);  
            invoker = refprotocol.refer(interfaceClass, url);  
            if (logger.isInfoEnabled()) {  
                logger.info("Using injvm service " + interfaceClass.getName());  
            }  
        } else {  
            if (url != null && url.length() > 0) { // 用户指定URL,指定的URL可能是对点对直连地址,也可能是注册中心URL  
                String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);  
                if (us != null && us.length > 0) {  
                    for (String u : us) {  
                        URL url = URL.valueOf(u);  
                        if (url.getPath() == null || url.getPath().length() == 0) {  
                            url = url.setPath(interfaceName);  
                        }  
                        if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {  
                            urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));  
                        } else {  
                            urls.add(ClusterUtils.mergeUrl(url, map));  
                        }  
                    }  
                }  
            } else { // 通过注册中心配置拼装URL  
                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)));  
                    }  
                }  
                if (urls == null || urls.size() == 0) {  
                    throw new IllegalStateException("No such any registry to reference " + interfaceName  + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");  
                }  
            }  

            if (urls.size() == 1) {  
                //此处举例说明如果是Dubbo协议则调用DubboProtocol的refer方法生成invoker,当用户调用service接口实际调用的是invoker的invoke方法  
                invoker = refprotocol.refer(interfaceClass, urls.get(0));  
            } else {  
                //多个service生成多个invoker  
                List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();  
                URL registryURL = null;  
                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));  
                }  
            }  
        }  

        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 " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());  
        }  
        if (logger.isInfoEnabled()) {  
            logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());  
        }  
        // 创建服务代理  
        return (T) proxyFactory.getProxy(invoker);  
    }

至此Reference在关联了所有application、module、consumer、registry、monitor、service、protocol后调用对应Protocol类的refer方法生成InvokerProxy。当用户调用service时dubbo会通过InvokerProxy调用Invoker的invoke的方法向服务端发起请求。客户端就这样完成了自己的初始化。

更新于 2018-01-12
在线,4小时前登录

查看dubbo更多相关的文章或提一个关于dubbo的问题,也可以与我们一起分享文章