10Dubbo的服务引入

39次阅读

共计 10127 个字符,预计需要花费 26 分钟才能阅读完成。

1、Dubbo 服务引入简介

前两篇文章 Dubbo 的服务导出 1 之导出到本地、Dubbo 的服务导出 2 之导出到远程 详细分析了服务导出的过程,本篇文章我们趁热打铁,继续分析服务引用过程。在 Dubbo 中,我们可以通过两种方式引用远程服务。第一种是使用 服务直连的方式引用服务 ,第二种方式是 基于注册中心进行引用。服务直连的方式仅适合在调试或测试服务的场景下使用,不适合在线上环境使用。因此,本文我将重点分析通过注册中心引用服务的过程。从注册中心中获取服务配置只是服务引用过程中的一环,除此之外,服务消费者还需要经历 Invoker 创建、代理类创建等步骤。这些步骤,将在后续章节中一一进行分析。

2、服务引用原理

Dubbo 服务引用的时机有两个,第一个是在 Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时引用服务,第二个是在 ReferenceBean 对应的服务被注入到其他类中时引用。这两个引用服务的时机区别在于,第一个是饿汉式的,第二个是懒汉式的。 默认情况下,Dubbo 使用懒汉式引用服务 。如果需要使用饿汉式,可通过配置 <dubbo:reference> 的 init 属性开启。下面我们按照 Dubbo 默认配置进行分析,整个分析过程从 ReferenceBean 的 getObject 方法 开始。当我们的服务被注入到其他类中时,Spring 会第一时间调用 getObject 方法,并由该方法执行服务引用逻辑。按照惯例,在进行具体工作之前,需先进行配置检查与收集工作。接着根据收集到的信息决定服务用的方式,有三种,第一种是 引用本地 (JVM) 服务 ,第二是通过 直连方式引用远程服务 ,第三是通过 注册中心引用远程服务 。不管是哪种引用方式,最后都会得到一个 Invoker 实例。如果有多个注册中心,多个服务提供者,这个时候会得到一组 Invoker 实例,此时需要通过集群管理类 Cluster 将多个 Invoker 合并成一个实例。合并后的 Invoker 实例已经具备调用本地或远程服务的能力了,但并不能将此实例暴露给用户使用,这会对用户业务代码造成侵入。此时框架还需要通过代理工厂类 (ProxyFactory) 为服务接口生成代理类,并让代理类去调用 Invoker 逻辑。避免了 Dubbo 框架代码对业务代码的侵入,同时也让框架更容易使用。
以上就是服务引用的大致原理,下面我们深入到代码中,详细分析服务引用细节。

3、源码分析

3.1、配置检查与收集

// ReferenceBean 实现了 InitializingBean 接口, 因此在 Spring 容器初始化时会调用该方法, 这里也对应上述第一个引用时机
// 默认不会走这里的引用服务方法, 需要配置 init=true
@Override
public void afterPropertiesSet() throws Exception {
    // 删除一些代码
    
    // Dubbo 服务引用的时机有两个, 第一个是在 Spring 容器调用 ReferenceBean 的 afterPropertiesSet
    // 方法时引用服务, 第二个是在 ReferenceBean 对应的服务被注入到其他类中时引用. 这两个引用服务的时机
    // 区别在于, 第一个是饿汉式的, 第二个是懒汉式的. 默认情况下,Dubbo 使用懒汉式引用服务. 如果需要使用
    // 饿汉式, 可通过配置 <dubbo:reference> 的 init 属性开启.
    Boolean b = isInit();
    if (b == null && getConsumer() != null) {b = getConsumer().isInit();}
    if (b != null && b.booleanValue()) {getObject();
    }
}
/**
 * 整个分析过程从 ReferenceBean 的 getObject 方法开始. 当我们的服务被注入到其他类中时,
 * Spring 会第一时间调用 getObject 方法, 并由该方法执行服务引用逻辑
 */
@Override
public Object getObject() throws Exception {return get();
}

public synchronized T get() {if (destroyed) {throw new IllegalStateException("Already destroyed!");
    }
    // 检测 ref 是否为空, 为空则通过 init 方法创建
    if (ref == null) {
        // init 方法主要用于处理配置, 以及调用 createProxy 生成代理类
        init();}
    return ref;
}

init 方法比较长,为了排版,分成几段,并删除异常捕捉、日志记录等代码,核心代码是ref = createProxy(map)

private void init() {
    // 避免重复初始化
    if (initialized) {return;}
    initialized = true;
    // 检查接口合法性
    if (interfaceName == null || interfaceName.length() == 0) {throw new IllegalStateException("xxx");
    }
    // 获取 consumer 的全局配置
    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);
            } 
            // 获取与接口名对应的配置
            resolve = properties.getProperty(interfaceName);
        }
    }
    if (resolve != null && resolve.length() > 0) {
        // 将 resolve 赋值给 url, 用于点对点直连
        url = resolve;
    }
    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();
        }
    }
    
    // 检测 Application 合法性
    checkApplication();
    // 检测本地存根配置合法性
    checkStubAndMock(interfaceClass);

    // 添加 side、协议版本信息、时间戳和进程号等信息到 map 中,side 表示处于哪一侧, 目前是处于服务消费者侧
    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.getProtocolVersion());
    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);
        }

        // 获取接口方法列表, 并添加到 map 中
        String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
        if (methods.length == 0) {map.put("methods", Constants.ANY_VALUE);
        } else {map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
        }
    }
    map.put(Constants.INTERFACE_KEY, interfaceName);
    // 将 ApplicationConfig、ConsumerConfig、ReferenceConfig 等对象的字段信息添加到 map 中
    appendParameters(map, application);
    appendParameters(map, module);
    appendParameters(map, consumer, Constants.DEFAULT_KEY);
    appendParameters(map, this);
    // 形如 com.alibaba.dubbo.demo.DemoService
    String prefix = StringUtils.getServiceKey(map);
    if (methods != null && !methods.isEmpty()) {
        // 遍历 MethodConfig 列表
        for (MethodConfig method : methods) {appendParameters(map, method, method.getName());
            String retryKey = method.getName() + ".retry";
            // 检测 map 是否包含 methodName.retry
            if (map.containsKey(retryKey)) {String retryValue = map.remove(retryKey);
                if ("false".equals(retryValue)) {
                    // 添加重试次数配置 methodName.retries
                    map.put(method.getName() + ".retries", "0");
                }
            }
            // 添加 MethodConfig 中的“属性”字段到 attributes
            // 比如 onreturn、onthrow、oninvoke 等
            appendAttributes(attributes, method, prefix + "." + method.getName());
            checkAndConvertImplicitConfig(method, map, attributes);
        }
    }
    // 获取服务消费者 ip 地址
    String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
    if (hostToRegistry == null || hostToRegistry.length() == 0) {hostToRegistry = NetUtils.getLocalHost();
    } else if (isInvalidLocalHost(hostToRegistry)) {throw new IllegalArgumentException("");
    }
    map.put(Constants.REGISTER_IP_KEY, hostToRegistry);

    // 存储 attributes 到系统上下文中
    StaticContext.getSystemContext().putAll(attributes);

    // 创建代理类, 核心
    ref = createProxy(map);

    // 根据服务名,ReferenceConfig, 代理类构建 ConsumerModel,
    // 并将 ConsumerModel 存入到 ApplicationModel 中
    ConsumerModel consumerModel = 
                new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
    ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
}

3.2、创建 Invoker 及代理

上述 init()方法的核心代码是 ref = createProxy(map),下面分析这个方法。

3.2.1、总体分析

// 从字面意思上来看,createProxy 似乎只是用于创建代理对象的, 但实际上并非如此,
// 该方法还会调用其他方法构建以及合并 Invoker 实例
private T createProxy(Map<String, String> map) {URL tmpUrl = new URL("temp", "localhost", 0, map);
    final boolean isJvmRefer;
    if (isInjvm() == null) {
        // url 配置被指定, 则不做本地引用, 我理解该 url 是用来做直连的
        if (url != null && url.length() > 0) {isJvmRefer = false;}
        // 根据 url 的协议、scope 以及 injvm 等参数检测是否需要本地引用
        // 比如如果用户显式配置了 scope=local, 此时 isInjvmRefer 返回 true
        else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {isJvmRefer = true;} else {isJvmRefer = false;}
    } else {
        // 获取 injvm 配置值
        isJvmRefer = isInjvm().booleanValue();
    }
    // 本地引用
    if (isJvmRefer) {
        // 生成本地引用 URL, 协议为 injvm
        URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, 
                                                   interfaceClass.getName()).addParameters(map);
        // 调用 refer 方法构建 InjvmInvoker 实例
        invoker = refprotocol.refer(interfaceClass, url);
    }
    // 远程引用
    else {
        // url 不为空, 表明用户可能想进行点对点调用
        if (url != null && url.length() > 0) {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.isEmpty()) {for (URL u : us) {URL monitorUrl = loadMonitor(u);
                    if (monitorUrl != null) {map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                    }
                    // 添加 refer 参数到 url 中, 并将 url 添加到 urls 中
                    urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, 
                                                             StringUtils.toQueryString(map)));
                }
            }
            // 未配置注册中心, 抛出异常
            if (urls.isEmpty()) {// 抛异常}
        }
        // 单个注册中心或服务提供者(服务直连, 下同)
        if (urls.size() == 1) {
            // 1. 调用 RegistryProtocol 的 refer 构建 Invoker 实例, 普通情况下走这个逻辑
            invoker = refprotocol.refer(interfaceClass, urls.get(0));
        }
        // 多个注册中心或多个服务提供者, 或者两者混合
        else {List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null; 
            for (URL url : urls) {
                // 获取所有的 Invoker, 通过 refprotocol 调用 refer 构建 Invoker,refprotocol 会在运行时
                // 根据 url 协议头加载指定的 Protocol 实例, 并调用实例的 refer 方法
                invokers.add(refprotocol.refer(interfaceClass, url));
                if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {registryURL = url;}
            }
            // 如果注册中心链接不为空, 则将使用 AvailableCluster
            if (registryURL != null) {URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
                // 创建 StaticDirectory 实例, 并由 Cluster 对多个 Invoker 进行合并
                invoker = cluster.join(new StaticDirectory(u, invokers));
            } else {invoker = cluster.join(new StaticDirectory(invokers));
            }
        }
    }
    Boolean c = check;
    if (c == null && consumer != null) {c = consumer.isCheck();
    }
    if (c == null) {
        // default true
        c = true;
    }

    // invoker 可用性检查
    if (c && !invoker.isAvailable()) {// 抛异常}

    // 2. 生成代理类, 核心
    // Invoker 创建完毕后, 接下来要做的事情是为服务接口生成代理对象. 有了代理对象,
    // 即可进行远程调用. 代理对象生成的入口方法为 ProxyFactory 的 getProxy
    return (T) proxyFactory.getProxy(invoker);
}

3.2.2、创建 Invoker

// 下面分析创建 Invoker, 单个注册中心或服务提供者(服务直连, 下同)
if (urls.size() == 1) {
    // 1. 调用 RegistryProtocol 的 refer 构建 Invoker 实例, 普通情况下走这个逻辑
    invoker = refprotocol.refer(interfaceClass, urls.get(0));
}
  public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    // protocol 为 zookeeper, 调试加的
    String protocol = url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY);
    url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, 
                               Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
    // 获取注册中心实例
    Registry registry = registryFactory.getRegistry(url);
    // 这里的 type 是 ""
    if (RegistryService.class.equals(type)) {return proxyFactory.getInvoker((T) registry, type, url);
    }
    // 将 url 查询字符串转为 Map
    Map<String, String> qs = 
                      StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
    // 获取 group 配置
    String group = qs.get(Constants.GROUP_KEY);
    if (group != null && group.length() > 0) {if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
                || "*".equals(group)) {
            // 通过 SPI 加载 MergeableCluster 实例, 并调用 doRefer 继续执行服务引用逻辑
            return doRefer(getMergeableCluster(), registry, type, url);
        }
    }
    // 调用 doRefer 继续执行服务引用逻辑
    return doRefer(cluster, registry, type, url);
}

3.2.3、创建代理

正文完
 0