dubbo源码分析系列——dubbo核心流程分析

扫码关注公众号:Java 技术驿站

发送:vip
将链接复制到本浏览器,永久解锁本站全部文章

【公众号:Java 技术驿站】 【加作者微信交流技术,拉技术群】

作者:杨武兵

出处:https://my.oschina.net/ywbrj042?tab=newest&catalogId=3594617


dubbo流程图

首先我们将引用dubbo官方文档和其它一些博客中发布的dubbo的流程图来对dubbo的流程有个概括的了解。

参考博客:http://shiyanjun.cn/archives/325.html

dubbo的总体架构图

服务发布及取消时序

服务引用及销毁时序

源代码调试流程追踪

接下来我们将通过调试源代码的方式来追踪核心流程,让我们更加细致和深入的了解dubbo的核心流程。

源代码调试环境

接口ProcessService定义

package com.alibaba.dubbo.test;

public interface ProcessService {

    public Object test(Object object);

}

接口实现类ProcessServiceImpl

package com.alibaba.dubbo.test;

public class ProcessServiceImpl implements ProcessService {

    public Object test(Object object) {
         //测试方法,直接返回参数值。
        return object;
    }

}

服务发布入口类ServerMain

package com.alibaba.dubbo.test;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class ServerMain {

    public static void main(String[] args) {
        ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("SpringApplicationContext.xml");
        ctx.start();
        synchronized (ServerMain.class) {
            try {
                ServerMain.class.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

服务发布配置文件SpringApplicationContext.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://code.alibabatech.com/schema/dubbo
        http://code.alibabatech.com/schema/dubbo/dubbo.xsd
        ">

    <bean id="processServiceImpl" class="com.alibaba.dubbo.test.ProcessServiceImpl" />

    <dubbo:registry address="zookeeper://127.0.0.1:2181"/>

    <!-- 服务应用配置 -->
    <dubbo:application name="dubbo_provider" />

    <!-- 服务提供者全局配置 -->
    <dubbo:protocol name="dubbo" port="20885"/>

    <!-- 服务提供者暴露服务配置 -->
    <dubbo:service id="processService" interface="com.alibaba.dubbo.test.ProcessService"
        ref="processServiceImpl"/>

</beans>

客户应用及调用主类ClientMain

package com.alibaba.dubbo.test;

import com.alibaba.dubbo.config.ApplicationConfig;
import com.alibaba.dubbo.config.ReferenceConfig;
import com.alibaba.dubbo.config.RegistryConfig;

public class ClientMain {

    public static void main(String[] args) {

        RegistryConfig registryConfig = new RegistryConfig("zookeeper://127.0.0.1:2181");

        ApplicationConfig application = new ApplicationConfig();
        application.setName("dubbo_consumer");

        ReferenceConfig<ProcessService> referenceConfig = new ReferenceConfig<ProcessService>();
        referenceConfig.setRegistry(registryConfig);
        referenceConfig.setApplication(application);

        referenceConfig.setInterface(ProcessService.class);

        ProcessService processService = referenceConfig.get();

        System.out.println(processService.test("hello, dubbo"));

    }

}

这里用了一个最简单的接口和方法,目的是将我们的关注点放在dubbo本身的实现及流程上,而无须关注业务逻辑上。

服务发布流程调试

服务发布首先进入的是dubbo-config-api模块的ServiceConfig类,进入该类的export方法。

从图中可以看出,我们在Spring容器中发布dubbo,在Spring容器启动的时候发布事件的时候,因为ServiceBean作为ApplicationListener的一个实现类,能够监听到容器应用的事件,在处理事件的时候会调用父类ServiceConfig的export方法,而该方法真正实现了服务的发布。

该方法源码如下:

 public synchronized void export() {
        if (provider != null) {
            if (export == null) {
                export = provider.getExport();
            }
            if (delay == null) {
                delay = provider.getDelay();
            }
        }
        if (export != null && ! export.booleanValue()) {
            return;
        }
        if (delay != null && delay > 0) {
            Thread thread = new Thread(new Runnable() {
                public void run() {
                    try {
                        Thread.sleep(delay);
                    } catch (Throwable e) {
                    }
                    doExport();
                }
            });
            thread.setDaemon(true);
            thread.setName("DelayExportServiceThread");
            thread.start();
        } else {
            doExport();
        }
    }

可以看出发布发布是支持延迟异步发布服务的,这样可以用于当我们发布的服务非常多,影响到应用启动的问题,前提是应用允许服务发布的延迟特性。

接下来就进入到内部方法doExport。

 protected synchronized void doExport() {
        if (unexported) {
            throw new IllegalStateException("Already unexported!");
        }
        if (exported) {
            return;
        }
        exported = true;
        if (interfaceName == null || interfaceName.length() == 0) {
            throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");
        }
        checkDefault();
        if (provider != null) {
            if (application == null) {
                application = provider.getApplication();
            }
            if (module == null) {
                module = provider.getModule();
            }
            if (registries == null) {
                registries = provider.getRegistries();
            }
            if (monitor == null) {
                monitor = provider.getMonitor();
            }
            if (protocols == null) {
                protocols = provider.getProtocols();
            }
        }
        if (module != null) {
            if (registries == null) {
                registries = module.getRegistries();
            }
            if (monitor == null) {
                monitor = module.getMonitor();
            }
        }
        if (application != null) {
            if (registries == null) {
                registries = application.getRegistries();
            }
            if (monitor == null) {
                monitor = application.getMonitor();
            }
        }
        if (ref instanceof GenericService) {
            interfaceClass = GenericService.class;
            if (StringUtils.isEmpty(generic)) {
                generic = Boolean.TRUE.toString();
            }
        } else {
            try {
                interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
                        .getContextClassLoader());
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            checkInterfaceAndMethods(interfaceClass, methods);
            checkRef();
            generic = Boolean.FALSE.toString();
        }
        if(local !=null){
            if(local=="true"){
                local=interfaceName+"Local";
            }
            Class<?> localClass;
            try {
                localClass = ClassHelper.forNameWithThreadContextClassLoader(local);
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            if(!interfaceClass.isAssignableFrom(localClass)){
                throw new IllegalStateException("The local implemention class " + localClass.getName() + " not implement interface " + interfaceName);
            }
        }
        if(stub !=null){
            if(stub=="true"){
                stub=interfaceName+"Stub";
            }
            Class<?> stubClass;
            try {
                stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub);
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            if(!interfaceClass.isAssignableFrom(stubClass)){
                throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + interfaceName);
            }
        }
        checkApplication();
        checkRegistry();
        checkProtocol();
        appendProperties(this);
        checkStubAndMock(interfaceClass);
        if (path == null || path.length() == 0) {
            path = interfaceName;
        }
        doExportUrls();
    }

我们可以看出该方法的实现的逻辑包含了根据配置的优先级将ProviderConfig,ModuleConfig,MonitorConfig,ApplicaitonConfig等一些配置信息进行组装和合并。还有一些逻辑是检查配置信息的合法性。最后又调用了doExportUrls方法。

 private void doExportUrls() {
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

该方法第一步是加载注册中心列表,第二部是将服务发布到多种协议的url上,并且携带注册中心列表的参数,从这里我们可以看出dubbo是支持同时将一个服务发布成为多种协议的,这个需求也是很正常的,客户端也需要支持多协议,根据不同的场景选择合适的协议。

进入方法loadRegistries看看实现逻辑。

 protected List<URL> loadRegistries(boolean provider) {
        checkRegistry();
        List<URL> registryList = new ArrayList<URL>();
        if (registries != null && registries.size() > 0) {
            for (RegistryConfig config : registries) {
                String address = config.getAddress();
                if (address == null || address.length() == 0) {
                    address = Constants.ANYHOST_VALUE;
                }
                String sysaddress = System.getProperty("dubbo.registry.address");
                if (sysaddress != null && sysaddress.length() > 0) {
                    address = sysaddress;
                }
                if (address != null && address.length() > 0
                        && ! RegistryConfig.NO_AVAILABLE.equalsIgnoreCase(address)) {
                    Map<String, String> map = new HashMap<String, String>();
                    appendParameters(map, application);
                    appendParameters(map, config);
                    map.put("path", RegistryService.class.getName());
                    map.put("dubbo", 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 (! map.containsKey("protocol")) {
                        if (ExtensionLoader.getExtensionLoader(RegistryFactory.class).hasExtension("remote")) {
                            map.put("protocol", "remote");
                        } else {
                            map.put("protocol", "dubbo");
                        }
                    }
                    List<URL> urls = UrlUtils.parseURLs(address, map);
                    for (URL url : urls) {
                        url = url.addParameter(Constants.REGISTRY_KEY, url.getProtocol());
                        url = url.setProtocol(Constants.REGISTRY_PROTOCOL);
                        if ((provider && url.getParameter(Constants.REGISTER_KEY, true))
                                || (! provider && url.getParameter(Constants.SUBSCRIBE_KEY, true))) {
                            registryList.add(url);
                        }
                    }
                }
            }
        }
        return registryList;
    }

该方法先检查了注册中心地址的合法性,然后将注册中心配置转化为URL列表,其中若地址为'N/A' 则表示无效注册中心地址,则会跳过。最后返回的URL列表的协议名称为registry,并且会在URL参数registry的值中保存原始的注册中心地址协议名称值。后面会使用到该值。

然后我们回到方法doExportUrls,下一步是将得到注册中心URL列表作为一个参数,另外一个参数是协议配置信息,调用方法doExportUrlsFor1Protocol继续实现发布服务逻辑。

 private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        String name = protocolConfig.getName();
        if (name == null || name.length() == 0) {
            name = "dubbo";
        }

        String host = protocolConfig.getHost();
        if (provider != null && (host == null || host.length() == 0)) {
            host = provider.getHost();
        }
        boolean anyhost = false;
        if (NetUtils.isInvalidLocalHost(host)) {
            anyhost = true;
            try {
                host = InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                logger.warn(e.getMessage(), e);
            }
            if (NetUtils.isInvalidLocalHost(host)) {
                if (registryURLs != null && registryURLs.size() > 0) {
                    for (URL registryURL : registryURLs) {
                        try {
                            Socket socket = new Socket();
                            try {
                                SocketAddress addr = new InetSocketAddress(registryURL.getHost(), registryURL.getPort());
                                socket.connect(addr, 1000);
                                host = socket.getLocalAddress().getHostAddress();
                                break;
                            } finally {
                                try {
                                    socket.close();
                                } catch (Throwable e) {}
                            }
                        } catch (Exception e) {
                            logger.warn(e.getMessage(), e);
                        }
                    }
                }
                if (NetUtils.isInvalidLocalHost(host)) {
                    host = NetUtils.getLocalHost();
                }
            }
        }

        Integer port = protocolConfig.getPort();
        if (provider != null && (port == null || port == 0)) {
            port = provider.getPort();
        }
        final int defaultPort = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(name).getDefaultPort();
        if (port == null || port == 0) {
            port = defaultPort;
        }
        if (port == null || port <= 0) {
            port = getRandomPort(name);
            if (port == null || port < 0) {
                port = NetUtils.getAvailablePort(defaultPort);
                putRandomPort(name, port);
            }
            logger.warn("Use random available port(" + port + ") for protocol " + name);
        }

        Map<String, String> map = new HashMap<String, String>();
        if (anyhost) {
            map.put(Constants.ANYHOST_KEY, "true");
        }
        map.put(Constants.SIDE_KEY, Constants.PROVIDER_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()));
        }
        appendParameters(map, application);
        appendParameters(map, module);
        appendParameters(map, provider, Constants.DEFAULT_KEY);
        appendParameters(map, protocolConfig);
        appendParameters(map, this);
        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");
                    }
                }
                List<ArgumentConfig> arguments = method.getArguments();
                if (arguments != null && arguments.size() > 0) {
                    for (ArgumentConfig argument : arguments) {
                        //类型自动转换.
                        if(argument.getType() != null && argument.getType().length() >0){
                            Method[] methods = interfaceClass.getMethods();
                            //遍历所有方法
                            if(methods != null && methods.length > 0){
                                for (int i = 0; i < methods.length; i++) {
                                    String methodName = methods[i].getName();
                                    //匹配方法名称,获取方法签名.
                                    if(methodName.equals(method.getName())){
                                        Class<?>[] argtypes = methods[i].getParameterTypes();
                                        //一个方法中单个callback
                                        if (argument.getIndex() != -1 ){
                                            if (argtypes[argument.getIndex()].getName().equals(argument.getType())){
                                                appendParameters(map, argument, method.getName() + "." + argument.getIndex());
                                            }else {
                                                throw new IllegalArgumentException("argument config error : the index attribute and type attirbute not match :index :"+argument.getIndex() + ", type:" + argument.getType());
                                            }
                                        } else {
                                            //一个方法中多个callback
                                            for (int j = 0 ;j<argtypes.length ;j++) {
                                                Class<?> argclazz = argtypes[j];
                                                if (argclazz.getName().equals(argument.getType())){
                                                    appendParameters(map, argument, method.getName() + "." + j);
                                                    if (argument.getIndex() != -1 && argument.getIndex() != j){
                                                        throw new IllegalArgumentException("argument config error : the index attribute and type attirbute not match :index :"+argument.getIndex() + ", type:" + argument.getType());
                                                    }
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }else if(argument.getIndex() != -1){
                            appendParameters(map, argument, method.getName() + "." + argument.getIndex());
                        }else {
                            throw new IllegalArgumentException("argument config must set index or type attribute.eg: <dubbo:argument index='0' .../> or <dubbo:argument type=xxx .../>");
                        }

                    }
                }
            } // end of methods for
        }

        if (ProtocolUtils.isGeneric(generic)) {
            map.put("generic", generic);
            map.put("methods", Constants.ANY_VALUE);
        } else {
            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)), ","));
            }
        }
        if (! ConfigUtils.isEmpty(token)) {
            if (ConfigUtils.isDefault(token)) {
                map.put("token", UUID.randomUUID().toString());
            } else {
                map.put("token", token);
            }
        }
        if ("injvm".equals(protocolConfig.getName())) {
            protocolConfig.setRegister(false);
            map.put("notify", "false");
        }
        // 导出服务
        String contextPath = protocolConfig.getContextpath();
        if ((contextPath == null || contextPath.length() == 0) && provider != null) {
            contextPath = provider.getContextpath();
        }
        URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

        if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                .hasExtension(url.getProtocol())) {
            url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                    .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
        }

        String scope = url.getParameter(Constants.SCOPE_KEY);
        //配置为none不暴露
        if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {

            //配置不是remote的情况下做本地暴露 (配置为remote,则表示只暴露远程服务)
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            //如果配置不是local则暴露为远程服务.(配置为local,则表示只暴露远程服务)
            if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
                if (logger.isInfoEnabled()) {
                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                }
                if (registryURLs != null && registryURLs.size() > 0
                        && url.getParameter("register", true)) {
                    for (URL registryURL : registryURLs) {
                        url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
                        URL monitorUrl = loadMonitor(registryURL);
                        if (monitorUrl != null) {
                            url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
                        }
                        if (logger.isInfoEnabled()) {
                            logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                        }
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));

                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                } else {
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);

                    Exporter<?> exporter = protocol.export(invoker);
                    exporters.add(exporter);
                }
            }
        }
        this.urls.add(url);
    }

该方法的逻辑是先根据服务配置、协议配置、发布服务的服务器信息、方法列表、dubbo版本等等信息组装成一个发布的URL对象。

若没有配置协议的host,则会自动生成一个host,这对于有多个网卡的系统会,比如本机上部署了虚拟机的情况下,生成的host可能不是开发者期望的地址,这种情况下需要开发者自己指定一个期望的host地址。

服务配置的scope是发布范围,配置为“none”表示不发布服务,则会停止发布操作;

若配置为非“remote”(包括null),则会调用exportLocal()方法继续发布服务;

若配置为非“local”(包含null),则会将服务发布到远程协议上,这里又2种情况:

第一种是有效的注册中心列表和无注册中心列表,如果有注册中心列表则会逐一将url的协议替换为regitrsy,而且dubbo支持多个注册中心,会注册到多个注册中心去,根据SPI会调用实现类RegistryProtocol的export方法发布服务;RegistryProtocol中主要的逻辑请参考另外一篇博文:http://my.oschina.net/ywbrj042/blog/690342

第二种是无有效的注册中心则会调用配置的协议类型对应的实现类,比如本例中使用的dubbo,则会调用实现类DubboProtocol的export方法方法服务。该协议的详细实现逻辑请参考博文:http://my.oschina.net/ywbrj042/blog/684718

exportLocal方法源码如下:

 private void exportLocal(URL url) {
        if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
            URL local = URL.valueOf(url.toFullString())
                    .setProtocol(Constants.LOCAL_PROTOCOL)
                    .setHost(NetUtils.LOCALHOST)
                    .setPort(0);

            // modified by lishen
            ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref));

            Exporter<?> exporter = protocol.export(
                    proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
            exporters.add(exporter);
            logger.info("Export dubbo service " + interfaceClass.getName() +" to local registry");
        }
    }

该方法的实现时将url发布在jvm中,在本地同一个jvm中调用服务,由于url的协议改为injvm,则我们根据SPI机制可以知道最终会调用实现类InjvmProtocol的export方法实现发布服务在injvm中。该协议的实现类非常简单,仅仅是用一个map记录了serviceKey和service bean的对应关系,则调用的时候直接拿到服务对应的bean调用目标方法即可。

服务引用流程调试

我们通过启动客户端应用远程服务的代码进入调试,进入ReferenceConfig.get()方法。

通过代码可以看出来,先检查状态,然后看是否已经有“ref”,如果已经说明已经引用过,则直接返回,否则会调用init()方法进行初始化。

我们进入该方法看看它的源码和流程。

private void init() {
        if (initialized) {
            return;
        }
        initialized = true;
        if (interfaceName == null || interfaceName.length() == 0) {
            throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
        }
        // 获取消费者全局配置
        checkDefault();
        appendProperties(this);
        if (getGeneric() == null && getConsumer() != null) {
            setGeneric(getConsumer().getGeneric());
        }
        if (ProtocolUtils.isGeneric(getGeneric())) {
            interfaceClass = GenericService.class;
        } else {
            try {
                interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
                        .getContextClassLoader());
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            checkInterfaceAndMethods(interfaceClass, methods);
        }
        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.");
                }
            }
        }
        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();
            }
        }
        if (module != null) {
            if (registries == null) {
                registries = module.getRegistries();
            }
            if (monitor == null) {
                monitor = module.getMonitor();
            }
        }
        if (application != null) {
            if (registries == null) {
                registries = application.getRegistries();
            }
            if (monitor == null) {
                monitor = application.getMonitor();
            }
        }
        checkApplication();
        checkStubAndMock(interfaceClass);
        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()));
        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);
        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);
        ref = createProxy(map);
    }

程序的流程是这样的:

1.先检查状态,是否已经销毁过。

2.配置各种参数,将默认参数值、generic值、cosumerConfig参数值进行合并。

3.dubbo支持通用远程服务GenericService接口。应该是无须发布服务,可以通过该接口远程调用动态方法。

4.加载接口类并检查。若接口类不存在,则会抛出异常。检查类和方法,主要是检查接口类的合法性,接口方法的合法性等。

5.dubbo支持通过resolveFile的方式直接制定调用URL,实现p2p直接调用,当注册中心出现故障的时候这样直接调用可以容错。

6.合并application、consumerConfig和module等参数配置。

7.生成合并之后的参数map对象。将引用需要的各项参数进行构造、合并。

8.调用方法ref = createProxy(map);创建代理的引用对象。

我们要进入createProxy方法继续跟踪。

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) {
                invoker = refprotocol.refer(interfaceClass, urls.get(0));
            } else {
                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);
    }

首先判断是否要做本地jvm应用,即直接调用本地jvm中对象的方法。直接制定了url地址不做本地应用,没有设置为inJvm协议不做本地引用。

如果做本地应用则会构造一个协议为injvm的URL,然后最终会调用实现类InjvmProtocol的refer方法获得调用本地jvm对象方法的代理对象。返回的是InjvmInvoker的对象。该方法会直接从map中找到目标对象,然后可以直接调用该对象的方法。

如果不做本地应用则会构造一个registry协议的URL,然后会加载注册中心列表,这个与发布服务相似,不做详细描述了。这里有3种情况。

如果没有注册中心列表。因为又没有配置url,则无法找到服务发布,则会抛出异常信息。

如果有一个注册中心url,则会直接通过该url的注册中心获得带来。

如果是多个则会循环每一个注册中心,则会获得多个invokers列表,最后将多个invoker列表使用cluster.join(new StaticDirectory(u, invokers));聚合成一个集群。

每个注册中心都会调用refprotocol.refer(interfaceClass, url)应用远程服务,由于url是registry协议,则最终会调用RegistryProtocol实现类的refer方法。详细说明参考http://my.oschina.net/ywbrj042/blog/690342

服务调用流程调试

得到服务的代理对象后,我们要调用该代理对象的方法了,如图所示。

我们看到得到的是一个动态代理类,对象名称为:com.alibaba.dubbo.common.bytecode.proxy0@69fb6037的动态代理对象。F5进入到代理对象的方法,发现进入了InvokerInvocationHandler.invoke方法开始执行代理方法。

如果是在Object类中定义的方法则直接调用invoker的该方法;

如果是toString,hashCode和equals等特殊方法,则也直接调用invoker对象的该方法。

否则其它方法则调用return invoker.invoke(new RpcInvocation(method, args)).recreate();执行。正常情况下都不是那些特殊方法,一般都是自定义接口的自定义方法,走这条路径。

我们进入了实现类MockClusterInvoker的invoke方法,该类事一个支持模拟特性和集群特性的Invoker实现类。

先检查方法参数是否开启模拟,如果开启模拟则进入执行模拟调用方法的分之;

否则进入调用FailoverClusterInvoker实现类的invoke方法,因为我们没有配置集群策略,模拟就是自动故障转移模式的集群。

然后进入抽象类AbstractClusterInvoker的invoke方法,实际类型是FailoverClusterInvoker;如下图所示:

最后会调用执行的代码是:return doInvoke(invocation, invokers, loadbalance);

其中三个参数的值如下:

invocation:参数调用信息,封装了调用的类、方法,参数类型及值等信息。

invokers:从directory中获取到的可用的invoker列表,实际值是size=1,是一个被多个过滤器、装饰器封装的一个invoker对象。

loadbalance:是类型为RandomLoadBalance的负载均衡器,这是默认的负载均衡策略。

然后进入到FailoverClusterInvoker类的doInvoke方法,如下图。

执行的是自动故障转移的集群执行器,先会获得重试次数,然后从列表中选择一个invoker对象,使用了负载均衡器,会排出掉已经执行过的invoker,因为我们这个例子中自由一个invoker对象,因此会一直重试改对象,当某个一个invoker对象执行抛出了非业务的RpcException异常后,则会重新调用下一个。

我们这里调用的是类名为com.alibaba.dubbo.registry.integration.RegistryDirectory$InvokerDelegete@5552768b的对象。进入该方法,最后进入的是类InvokerWrapper.invoke方法。

赞(1) 打赏
版权归原创作者所有,任何形式的转载请联系博主:daming_90:Java 技术驿站 » dubbo源码分析系列——dubbo核心流程分析

  • 暂无文章

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏