前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >dubbo源码——服务提供者的服务暴露过程(二)

dubbo源码——服务提供者的服务暴露过程(二)

原创
作者头像
冰枫
发布2018-06-11 12:23:58
1.6K0
发布2018-06-11 12:23:58
举报
文章被收录于专栏:冰枫

上篇博客讲到了doExportUrls()方法

  • #1顾名思义加载注册中心。(后面详细讲解
  • #2获取并遍历所有协议,将服务根据不同协议暴露,并注册到每个注册中心上。
代码语言:javascript
复制
//ServiceConfig.java
//serviceConfig->doExportUrls()
@SuppressWarnings({"unchecked", "rawtypes"})
private void doExportUrls() {
    List<URL> registryURLs = loadRegistries(true);   #1
    for (ProtocolConfig protocolConfig : protocols) { #2
        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
    }
}
  • #1检查registry,如果registries为空,则从环境中加载registry参数。
  • #2如果registry标签的address属性为空,则默认为0.0.0.0
  • #3从环境加载address,参数配置-Ddubbo.registry.address=127.0.0.1:2181,此优先级最高。
  • #4将application,config填充到map
  • #5将path设置为RegistryService类名。example:registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService/xxxxxxx
  • #6dubbo版本号、时间戳、pid,protocol
  • #7

##1将address进行分割,和属性填充例如在一个标签中中配置了多个地址,example:<dubbo:registry

address="dubbo://127.0.0.1:2181;zookeeper://127.0.0.2:2181"/>,则会被分割为dubbo://127.0.0.1:2181 , zookeeper://127.0.0.12:2181来处理。

##2对url的参数进行一些填充,如protocol、username、password、port、host,path,最终生成的url:example: registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo2-service&dubbo=2.6.1&pid=1773&timestamp=1528679207767

  • #8设置url参数,example: registry=zookeeper
  • #9设置协议为registry
  • #10我们在调用loadRegistries()时传入的参数为true,代表为服务提供者,那么register不应该为false,如果传入false,那说明为消费者,则subscribe不应该为false。

最终的url形式为example: registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo2-service&dubbo=2.6.1&pid=1922&registry=zookeeper&timestamp=1528679861114

代码语言:javascript
复制
public static List<URL> parseURLs(String address, Map<String, String> defaults) {
    if (address == null || address.length() == 0) {
        return null;
    }
    String[] addresses = Constants.REGISTRY_SPLIT_PATTERN.split(address);  ##1
    if (addresses == null || addresses.length == 0) {
        return null; //here won't be empty
    }
    List<URL> registries = new ArrayList<URL>();
    for (String addr : addresses) {
        registries.add(parseURL(addr, defaults));     ##2
    }
    return registries;
}
代码语言:javascript
复制
public static URL parseURL(String address, Map<String, String> defaults) {
    if (address == null || address.length() == 0) {
        return null;
    }
    String url;
    if (address.indexOf("://") >= 0) {
        url = address;
    } else {
        String[] addresses = Constants.COMMA_SPLIT_PATTERN.split(address);
        url = addresses[0];
        if (addresses.length > 1) {
            StringBuilder backup = new StringBuilder();
            for (int i = 1; i < addresses.length; i++) {
                if (i > 1) {
                    backup.append(",");
                }
                backup.append(addresses[i]);
            }
            url += "?" + Constants.BACKUP_KEY + "=" + backup.toString();
        }
    }
    String defaultProtocol = defaults == null ? null : defaults.get("protocol");
    if (defaultProtocol == null || defaultProtocol.length() == 0) {
        defaultProtocol = "dubbo";
    }
    String defaultUsername = defaults == null ? null : defaults.get("username");
    String defaultPassword = defaults == null ? null : defaults.get("password");
    int defaultPort = StringUtils.parseInteger(defaults == null ? null : defaults.get("port"));
    String defaultPath = defaults == null ? null : defaults.get("path");
    Map<String, String> defaultParameters = defaults == null ? null : new HashMap<String, String>(defaults);
    if (defaultParameters != null) {
        defaultParameters.remove("protocol");
        defaultParameters.remove("username");
        defaultParameters.remove("password");
        defaultParameters.remove("host");
        defaultParameters.remove("port");
        defaultParameters.remove("path");
    }
    URL u = URL.valueOf(url);
    boolean changed = false;
    String protocol = u.getProtocol();
    String username = u.getUsername();
    String password = u.getPassword();
    String host = u.getHost();
    int port = u.getPort();
    String path = u.getPath();
    Map<String, String> parameters = new HashMap<String, String>(u.getParameters());
    if ((protocol == null || protocol.length() == 0) && defaultProtocol != null && defaultProtocol.length() > 0) {
        changed = true;
        protocol = defaultProtocol;
    }
    if ((username == null || username.length() == 0) && defaultUsername != null && defaultUsername.length() > 0) {
        changed = true;
        username = defaultUsername;
    }
    if ((password == null || password.length() == 0) && defaultPassword != null && defaultPassword.length() > 0) {
        changed = true;
        password = defaultPassword;
    }
    /*if (u.isAnyHost() || u.isLocalHost()) {
        changed = true;
        host = NetUtils.getLocalHost();
    }*/
    if (port <= 0) {
        if (defaultPort > 0) {
            changed = true;
            port = defaultPort;
        } else {
            changed = true;
            port = 9090;
        }
    }
    if (path == null || path.length() == 0) {
        if (defaultPath != null && defaultPath.length() > 0) {
            changed = true;
            path = defaultPath;
        }
    }
    if (defaultParameters != null && defaultParameters.size() > 0) {
        for (Map.Entry<String, String> entry : defaultParameters.entrySet()) {
            String key = entry.getKey();
            String defaultValue = entry.getValue();
            if (defaultValue != null && defaultValue.length() > 0) {
                String value = parameters.get(key);
                if (value == null || value.length() == 0) {
                    changed = true;
                    parameters.put(key, defaultValue);
                }
            }
        }
    }
    if (changed) {
        u = new URL(protocol, username, password, host, port, path, parameters);
    }
    return u;
}

代码语言:javascript
复制
//AbstractInterfaceConfig.java
//AbstractInterfaceConfig.loadRegistries()
    protected List<URL> loadRegistries(boolean provider) {
        checkRegistry();   #1
        List<URL> registryList = new ArrayList<URL>();
        if (registries != null && !registries.isEmpty()) {
            for (RegistryConfig config : registries) {
                String address = config.getAddress();
                if (address == null || address.length() == 0) {    #2
                    address = Constants.ANYHOST_VALUE;
                }
                String sysaddress = System.getProperty("dubbo.registry.address");  #3
                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);       #4
                    appendParameters(map, config);
                    map.put("path", RegistryService.class.getName());   #5
                    map.put("dubbo", Version.getVersion());  #6
                    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); #7
                    for (URL url : urls) {
                        url = url.addParameter(Constants.REGISTRY_KEY, url.getProtocol()); #8
                        url = url.setProtocol(Constants.REGISTRY_PROTOCOL);   #9
                        if ((provider && url.getParameter(Constants.REGISTER_KEY, true))  #10
                                || (!provider && url.getParameter(Constants.SUBSCRIBE_KEY, true))) {
                            registryList.add(url);
                        }
                    }
                }
            }
        }
        return registryList;
    }

ok,既然注册中心URL构造好了,那么就该遍历每个协议,进行服务暴露和注册了!doExportUrlsFor1Protocol()

emmm,dubbo的方法命名和Spring的好像啊,do....就是方法也太长了吧。。

  • #1协议name,默认为dubbo。支持的协议还有rmi、hessian、http、webservice、thrift、redis
  • #2设置side,dubbo,timestamp,pid,ApplicationConfig,ModuleConfig,ProviderCofnig,ProtocolConfig,ServiceConfig等参数
  • #3如果在<dubbo:service/>标签中配置的子标签属性放入map。
  • #4如果generic为true,那么将generic=true,method=*作为参数
  • #5否则为普通接口,将当前服务的版本reversion=xxxx作为参数。
  • #6获取接口中的方法名,用,分割for example: methods=test1,test2;如果不存在方法则使用通配符*。
  • #7如果配置文件配置token为true或者default,则使用UUID作为token,否则将自定义。
  • #8获取host

①从系统参数中中获取host example: -DDUBBO_IP_TO_BIND=192.168.1.1

String hostToBind = getValueFromConfig(protocolConfig, Constants.DUBBO_IP_TO_BIND);

②从配置文件中获取host example <dubbo:protocol host="192.168.1.1"/>

hostToBind = provider.getHost()

③从本机中获取host

hostToBind = InetAddress.getLocalHost().getHostAddress();

④获取注册中心的host

hostToBind = socket.getLocalAddress().getHostAddress();

⑤遍历本地网卡,返回第一个合理的IP

hostToBind = getLocalHost();

host有效性验证:

代码语言:javascript
复制
public static boolean isInvalidLocalHost(String host) {
    return host == null
            || host.length() == 0
            || host.equalsIgnoreCase("localhost")
            || host.equals("0.0.0.0")
            || (LOCAL_IP_PATTERN.matcher(host).matches());
}
  • #9 获取port

①从系统运行参数中获取 for example: -DDUBBO_PORT_TO_BIND=28080

String port = getValueFromConfig(protocolConfig, Constants.DUBBO_PORT_TO_BIND);

②从配置文件中获取 for example: <dubbo:protocol port="28080"/>

portToBind = protocolConfig.getPort();

③从provider配置中获取端口

portToBind = provider.getPort();

④获取协议默认端口

final int defaultPort = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(name).getDefaultPort();

⑤获取随机端口

portToBind = getRandomPort(name);

  • #10根据以上参数构造出协议对应的URL。
  • #11如果配置了scope为none,则不暴露服务。
  • #12如果scope配置为local,(默认为local),则在本地进行暴露,而不向注册中心进行注册服务。
  • #13如果scope配置为remote,则先在本地进行暴露,然后向注册中心注册服务。
代码语言:javascript
复制
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
    String name = protocolConfig.getName();    #1
    if (name == null || name.length() == 0) {
        name = "dubbo";
    }

    Map<String, String> map = new HashMap<String, String>();
    map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE); #2
    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.isEmpty()) {   #3
        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.isEmpty()) {
                for (ArgumentConfig argument : arguments) {
                    // convert argument type
                    if (argument.getType() != null && argument.getType().length() > 0) {
                        Method[] methods = interfaceClass.getMethods();
                        // visit all methods
                        if (methods != null && methods.length > 0) {
                            for (int i = 0; i < methods.length; i++) {
                                String methodName = methods[i].getName();
                                // target the method, and get its signature
                                if (methodName.equals(method.getName())) {
                                    Class<?>[] argtypes = methods[i].getParameterTypes();
                                    // one callback in the method
                                    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 attribute not match :index :" + argument.getIndex() + ", type:" + argument.getType());
                                        }
                                    } else {
                                        // multiple callbacks in the method
                                        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 attribute 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)) {    #4
        map.put("generic", generic);
        map.put("methods", Constants.ANY_VALUE);
    } else {    #5
        String revision = Version.getVersion(interfaceClass, version);
        if (revision != null && revision.length() > 0) {
            map.put("revision", revision);
        }

        String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames(); #6
        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)) {    #7
        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");
    }
    // export service
    String contextPath = protocolConfig.getContextpath();
    if ((contextPath == null || contextPath.length() == 0) && provider != null) {
        contextPath = provider.getContextpath();
    }

    String host = this.findConfigedHosts(protocolConfig, registryURLs, map); #8
    Integer port = this.findConfigedPorts(protocolConfig, name, map); #9
    URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
#10
    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);
    // don't export when none is configured
    if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {  #11

        // export to local if the config is not remote (export to remote only when config is remote)
        if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {  #12
            exportLocal(url);
        }
        // export to remote if the config is not local (export to local only when config is local)
        if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {  #13
            if (logger.isInfoEnabled()) {
                logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
            }
            if (registryURLs != null && !registryURLs.isEmpty()) {
                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()));
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

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

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

既然这样, 那么我们只需查看remote的代码即可。

  • #1增加dynamic参数,dynamic:服务是否动态注册,如果设为false,注册后将显示后disable状态,需人工启用,并且服务提供者停止时,也不会自动取消册,需人工禁用。 默认:true。
  • #2如果提供了监控中心,那么将会启用注册中心
  • #3使用ProxyFactory将实例包装成Invoker,dubbo默认使用JavaassistRpcProxyFactory
  • #4将invoker包装DelegateProviderMetaDataInvoker,包括invoker和ServiceConfig

代码语言:javascript
复制
if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {  
            if (logger.isInfoEnabled()) {
                logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
            }
            if (registryURLs != null && !registryURLs.isEmpty()) {
                for (URL registryURL : registryURLs) {
                    url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));  #1
                    URL monitorUrl = loadMonitor(registryURL); #2
                    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);
                    }
                    #3
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); #4

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

                Exporter<?> exporter = protocol.export(wrapperInvoker);
                exporters.add(exporter);
            }

export()方法是核心方法,根据不同协议暴露服务。下篇博客讲解

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
微服务引擎 TSE
微服务引擎(Tencent Cloud Service Engine)提供开箱即用的云上全场景微服务解决方案。支持开源增强的云原生注册配置中心(Zookeeper、Nacos 和 Apollo),北极星网格(腾讯自研并开源的 PolarisMesh)、云原生 API 网关(Kong)以及微服务应用托管的弹性微服务平台。微服务引擎完全兼容开源版本的使用方式,在功能、可用性和可运维性等多个方面进行增强。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档