Dubbo基础篇服务引用
大黄 Lv4
  2022-08-02 21:40:14 2022-08-02 21:40    2.3k Words  12 Mins  

完整流程图

Dubbo基础篇 服务引用完整流程.png

一句话概括流程

具体可以概括为以下五点

  • 两种模式（饿汉式/懒汉式）

  • 组装URL并向注册中心注册

  • 获取服务提供者信息并根据协议（默认Dubbo协议）开始Invoker的创建流程

  • 通过Cluster包装Invoker（默认FailoverCluster）

  • 返回代理类

    服务引用的两个时机

  • 饿汉式：通过实现Spring的 IntitalizingBean 接口的 afterProperties 方法，容器通过调用 ReferenceBean 的 afterPropertiesSet 方法时引入服务 (ReferenceBean实现了IntitalizingBean接口)
    Bean实例化完成后填充属性时引入服务，这里具体的Bean是Invoker的代理对象

  • 懒汉式：只有当这个服务被注入到其他类中时启动引入流程

注释：我们创建出来的远程调用的代理对象，实际上是ReferenceBean的一个成员变量（相当于我们在代理对象上套了个壳，把这个壳作为一个bean注册到spring容器中）
饿汉式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void afterPropertiesSet() throws Exception {

    // Initializes Dubbo's Config Beans before @Reference bean autowiring
    prepareDubboConfigBeans();

    // lazy init by default.
    if (init == null) {
        init = false;
    }

    // eager init if necessary.
    if (shouldInit()) {
        getObject();
    }
}

懒汉式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Override
public Object getObject() {
    return get();
}

public synchronized T get() {
    if (destroyed) {
        throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
    }
    if (ref == null) {
        init();
    }
    return ref;
}

关于这点的解释。如果 dubbo:reference 的init属性开启（饿汉式），创建完ReferenceBean 的时候，在afterProperties方法里面调用对应的getObject方法，将对应的Invoker的Bean创建出来并注册到容器中。而懒汉式则是在其他的Bean去注入它的时候，才会去进行Invoker的创建流程
然后ContextRefreshedEvent事件那一步的处理会将我上面没有去实例化的那些Invoker也实例化一次（个人猜测是那些懒加载没有被其他Bean注入的Bean）
为什么要是用FactoryBean接口的方式去创建对应的Bean。因为FactoryBean本身的作用是为了能够进行复杂的Bean逻辑创建，而不通过Spring本身的三级缓存的方式创建Bean。而且此处的Invoker还涉及到动态代理之类的情况，FactoryBean就很适合这种创建Bean场景
同样的，例如SpringCloud的Feign之类的也是通过这样的方式注入Bean

入口关键类 ReferenceConfig

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public <T> T get(ReferenceConfigBase<T> referenceConfig) {
    String key = generator.generateKey(referenceConfig);
    Class<?> type = referenceConfig.getInterfaceClass();

    proxies.computeIfAbsent(type, _t -> new ConcurrentHashMap<>());

    ConcurrentMap<String, Object> proxiesOfType = proxies.get(type);
    proxiesOfType.computeIfAbsent(key, _k -> {
        // 触发的地方，最终调用到init
        Object proxy = referenceConfig.get();
        referredReferences.put(key, referenceConfig);
        return proxy;
    });

    return (T) proxiesOfType.get(key);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
public synchronized void init() {
    if (initialized) {
        return;
    }


    if (bootstrap == null) {
        bootstrap = DubboBootstrap.getInstance();
        bootstrap.initialize();
    }

    checkAndUpdateSubConfigs();

    checkStubAndLocal(interfaceClass);
    ConfigValidationUtils.checkMock(interfaceClass, this);

    Map<String, String> map = new HashMap<String, String>();
    map.put(SIDE_KEY, CONSUMER_SIDE);

    ReferenceConfigBase.appendRuntimeParameters(map);
    if (!ProtocolUtils.isGeneric(generic)) {
        String revision = Version.getVersion(interfaceClass, version);
        if (revision != null && revision.length() > 0) {
            map.put(REVISION_KEY, revision);
        }

        String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
        if (methods.length == 0) {
            logger.warn("No method found in service interface " + interfaceClass.getName());
            map.put(METHODS_KEY, ANY_VALUE);
        } else {
            map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), COMMA_SEPARATOR));
        }
    }
    map.put(INTERFACE_KEY, interfaceName);
    AbstractConfig.appendParameters(map, getMetrics());
    AbstractConfig.appendParameters(map, getApplication());
    AbstractConfig.appendParameters(map, getModule());
    // remove 'default.' prefix for configs from ConsumerConfig
    // appendParameters(map, consumer, Constants.DEFAULT_KEY);
    AbstractConfig.appendParameters(map, consumer);
    AbstractConfig.appendParameters(map, this);
    MetadataReportConfig metadataReportConfig = getMetadataReportConfig();
    if (metadataReportConfig != null && metadataReportConfig.isValid()) {
        map.putIfAbsent(METADATA_KEY, REMOTE_METADATA_STORAGE_TYPE);
    }
    Map<String, AsyncMethodInfo> attributes = null;
    if (CollectionUtils.isNotEmpty(getMethods())) {
        attributes = new HashMap<>();
        for (MethodConfig methodConfig : getMethods()) {
            AbstractConfig.appendParameters(map, methodConfig, methodConfig.getName());
            String retryKey = methodConfig.getName() + ".retry";
            if (map.containsKey(retryKey)) {
                String retryValue = map.remove(retryKey);
                if ("false".equals(retryValue)) {
                    map.put(methodConfig.getName() + ".retries", "0");
                }
            }
            AsyncMethodInfo asyncMethodInfo = AbstractConfig.convertMethodConfig2AsyncInfo(methodConfig);
            if (asyncMethodInfo != null) {
//                    consumerModel.getMethodModel(methodConfig.getName()).addAttribute(ASYNC_KEY, asyncMethodInfo);
                attributes.put(methodConfig.getName(), asyncMethodInfo);
            }
        }
    }

    String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);
    if (StringUtils.isEmpty(hostToRegistry)) {
        hostToRegistry = NetUtils.getLocalHost();
    } else if (isInvalidLocalHost(hostToRegistry)) {
        throw new IllegalArgumentException("Specified invalid registry ip from property:" + DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
    }
    map.put(REGISTER_IP_KEY, hostToRegistry);

    serviceMetadata.getAttachments().putAll(map);

    // 创建代理 即生成invoker（代理中包括invoker）
    ref = createProxy(map);

    serviceMetadata.setTarget(ref);
    serviceMetadata.addAttribute(PROXY_CLASS_REF, ref);
    ConsumerModel consumerModel = repository.lookupReferredService(serviceMetadata.getServiceKey());
    consumerModel.setProxyObject(ref);
    consumerModel.init(attributes);

    initialized = true;

    checkInvokerAvailable();

    // dispatch a ReferenceConfigInitializedEvent since 2.7.4
    dispatch(new ReferenceConfigInitializedEvent(this, invoker));
}

创建代理类

主要分为以下几点

  • injvm：走jvm内部调用（自己服务中的dubbo接口调用）
  • 配置了URL：点对点
  • 未配置URL，从注册中心获取服务提供者信息
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    private T createProxy(Map<String, String> map) {
        if (shouldJvmRefer(map)) {
            // 本地调用的情况
            URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
            invoker = REF_PROTOCOL.refer(interfaceClass, url);
            if (logger.isInfoEnabled()) {
                logger.info("Using injvm service " + interfaceClass.getName());
            }
        } else {
            urls.clear();
            // 配置的url的情况 点对点/注册中心地址
            if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
                String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
                if (us != null && us.length > 0) {
                    for (String u : us) {
                        URL url = URL.valueOf(u);
                        if (StringUtils.isEmpty(url.getPath())) {
                            url = url.setPath(interfaceName);
                        }
                        if (UrlUtils.isRegistry(url)) {
                            // 如果是注册中心，将map转换为查询字符串，并作为refer参数的值添加到url中
                            urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                        } else {
                            // 点对点直连，合并url，移除一些服务提供者的配置
                            urls.add(ClusterUtils.mergeUrl(url, map));
                        }
                    }
                }
            } else { // assemble URL from register center's configuration
                // 未配置url的情况
                // if protocols not injvm checkRegistry
                if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
                    checkRegistry();
                    // 获取注册中心地址
                    List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
                    if (CollectionUtils.isNotEmpty(us)) {
                        for (URL u : us) {
                            URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
                            if (monitorUrl != null) {
                                map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                            }
                            urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map))); // registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=dubbo-demo-api-consumer&dubbo=2.0.2&pid=52327&refer=application=dubbo-demo-api-consumer&dubbo=2.0.2&generic=true&interface=org.apache.dubbo.demo.DemoService&pid=52327&register.ip=10.167.10.19&side=consumer&sticky=false&timestamp=1631087606024&registry=zookeeper&timestamp=1631087917015
                        }
                    }
                    if (urls.isEmpty()) {
                        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 = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
            } else {
                List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
                URL registryURL = null;
                for (URL url : urls) {
                    invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
                    if (UrlUtils.isRegistry(url)) {
                        registryURL = url; // use last registry url
                    }
                }
                if (registryURL != null) { // registry url is available
                    // for multi-subscription scenario, use 'zone-aware' policy by default
                    String cluster = registryURL.getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME);
                    // The invoker wrap sequence would be: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
                    invoker = Cluster.getCluster(cluster, false).join(new StaticDirectory(registryURL, invokers));
                } else { // not a registry url, must be direct invoke.
                    String cluster = CollectionUtils.isNotEmpty(invokers)
                            ? (invokers.get(0).getUrl() != null ? invokers.get(0).getUrl().getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME) : Cluster.DEFAULT)
                            : Cluster.DEFAULT;
                    invoker = Cluster.getCluster(cluster).join(new StaticDirectory(invokers));
                }
            }
        }

        if (logger.isInfoEnabled()) {
            logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
        }

        URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
        MetadataUtils.publishServiceDefinition(consumerURL);

        // create service proxy 创建代理服务
        return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
    }
    此处细讲一下通过注册中心创建Invoker的方式（多个服务提供者的情况）
    遍历注册中心，将注册中心所有的服务提供者信息（DynamicDirectory）汇集起来，创建出一个ClusterInvoker，并将ClusterInvoker的代理对象T返回给调用点
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
    URL registryURL = null;
    for (URL url : urls) {
    		// 引用服务
        invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
        if (UrlUtils.isRegistry(url)) {
            registryURL = url; // use last registry url
        }
    }
    if (registryURL != null) { // registry url is available
        // for multi-subscription scenario, use 'zone-aware' policy by default
        String cluster = registryURL.getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME);
        // The invoker wrap sequence would be: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
        invoker = Cluster.getCluster(cluster, false).join(new StaticDirectory(registryURL, invokers));
    } else { // not a registry url, must be direct invoke.
        String cluster = CollectionUtils.isNotEmpty(invokers)
                ? (invokers.get(0).getUrl() != null ? invokers.get(0).getUrl().getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME) : Cluster.DEFAULT)
                : Cluster.DEFAULT;
        invoker = Cluster.getCluster(cluster).join(new StaticDirectory(invokers));
    }

    引用

    具体协议的引用方法，以RegisterProtocol#refer为例
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    @Override
    @SuppressWarnings("unchecked")
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        url = getRegistryUrl(url);
    		// 获取注册中心
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }

        // group="a,b" or group="*"
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
        String group = qs.get(GROUP_KEY);
        if (group != null && group.length() > 0) {
            if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
                return doRefer(Cluster.getCluster(MergeableCluster.NAME), registry, type, url);
            }
        }

        Cluster cluster = Cluster.getCluster(qs.get(CLUSTER_KEY));
    		// 引用具体的服务
        return doRefer(cluster, registry, type, url);
    }
    1
    2
    3
    protected <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        return interceptInvoker(getInvoker(cluster, registry, type, url), url);
    }
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    protected <T> ClusterInvoker<T> getInvoker(Cluster cluster, Registry registry, Class<T> type, URL url) {
        // 封装成一个directory （invoker的集合）
        DynamicDirectory<T> directory = createDirectory(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY
        Map<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());
        URL urlToRegistry = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
        if (directory.isShouldRegister()) {
            directory.setRegisteredConsumerUrl(urlToRegistry);
            registry.register(directory.getRegisteredConsumerUrl());
        }
        // 建立路由规则链
        directory.buildRouterChain(urlToRegistry);
        // 订阅provider
        directory.subscribe(toSubscribeUrl(urlToRegistry));

        // 包装集群容错策略到invoker
        return (ClusterInvoker<T>) cluster.join(directory);
    }

    代理类

    1
    2
    3
    4
    5
    6
    return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));

    // 以Javassist动态代理为例
    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    }
    此处创建的代理类，会有一个成员变量，InvocationHandler
    当进行Dubbo远程调用的时候，就会被对应的InvokerInvocationHandler进行方法拦截处理，把看似本地调用的方法变为远程调用
  • Post title：Dubbo基础篇服务引用
  • Post author：大黄
  • Create time：2022-08-02 21:40:14
  • Post link：https://huangbangjing.cn/2022/08/02/Dubbo基础篇服务引用/
  • Copyright Notice：All articles in this blog are licensed under BY-NC-SA unless stating additionally.