Dubbo基础篇服务引用

完整流程图

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

一句话概括流程

具体可以概括为以下五点

两种模式（饿汉式/懒汉式）
组装URL并向注册中心注册
获取服务提供者信息并根据协议（默认Dubbo协议）开始Invoker的创建流程
通过Cluster包装Invoker（默认FailoverCluster）
返回代理类

服务引用的两个时机

饿汉式：通过实现Spring的 IntitalizingBean 接口的 afterProperties 方法，容器通过调用 ReferenceBean 的 afterPropertiesSet 方法时引入服务 (ReferenceBean实现了IntitalizingBean接口)
Bean实例化完成后填充属性时引入服务，这里具体的Bean是Invoker的代理对象
懒汉式：只有当这个服务被注入到其他类中时启动引入流程

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

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();
    }
}

懒汉式：

@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

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);
}

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，从注册中心获取服务提供者信息

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返回给调用点

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为例

@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);
}

protected <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    return interceptInvoker(getInvoker(cluster, registry, type, url), url);
}

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);
}

代理类

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进行方法拦截处理，把看似本地调用的方法变为远程调用

最后编辑于：2022.08.02 22:26:52

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 213,558评论 6赞 492
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 91,002评论 3赞 387
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 159,036评论 0赞 349
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 57,024评论 1赞 285
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,144评论 6赞 385
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,255评论 1赞 292
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,295评论 3赞 412
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 38,068评论 0赞 268
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,478评论 1赞 305
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,789评论 2赞 327
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,965评论 1赞 341
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,649评论 4赞 336
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,267评论 3赞 318
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,982评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,223评论 1赞 267
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 46,800评论 2赞 365
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 43,847评论 2赞 351

Dubbo基础篇 服务引用