Dubbo系列二之Consumer启动

@SpringBootApplication
@EnableDubbo(scanBasePackages = {"org.apache.dubbo.springboot.demo.provider"})
public class ProviderApplication {
    public static void main(String[] args) throws Exception {
        SpringApplication.run(ProviderApplication.class, args);
        System.out.println("dubbo service started");
        new CountDownLatch(1).await();
    }
}

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
@Import(DubboConfigConfigurationRegistrar.class)
public @interface EnableDubboConfig {......}

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(DubboComponentScanRegistrar.class)
public @interface DubboComponentScan {......}

public class DubboConfigConfigurationRegistrar implements ImportBeanDefinitionRegistrar {
    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        // initialize dubbo beans
        DubboSpringInitializer.initialize(registry);BeanDefinitionRegistry
    }
}

public class DubboComponentScanRegistrar implements ImportBeanDefinitionRegistrar {
    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        // initialize dubbo beans
        DubboSpringInitializer.initialize(registry);
        Set<String> packagesToScan = getPackagesToScan(importingClassMetadata);
        registerServiceAnnotationPostProcessor(packagesToScan, registry);
    }
}

static void registerCommonBeans(BeanDefinitionRegistry registry) {

    registerInfrastructureBean(registry, ServicePackagesHolder.BEAN_NAME, ServicePackagesHolder.class);

    registerInfrastructureBean(registry, ReferenceBeanManager.BEAN_NAME, ReferenceBeanManager.class);

    // Since 2.5.7 Register @Reference Annotation Bean Processor as an infrastructure Bean
    registerInfrastructureBean(registry, ReferenceAnnotationBeanPostProcessor.BEAN_NAME,
        ReferenceAnnotationBeanPostProcessor.class);

    // TODO Whether DubboConfigAliasPostProcessor can be removed ?
    // Since 2.7.4 [Feature] https://github.com/apache/dubbo/issues/5093
    registerInfrastructureBean(registry, DubboConfigAliasPostProcessor.BEAN_NAME,
        DubboConfigAliasPostProcessor.class);

    // register ApplicationListeners
    registerInfrastructureBean(registry, DubboDeployApplicationListener.class.getName(), DubboDeployApplicationListener.class);
    registerInfrastructureBean(registry, DubboConfigApplicationListener.class.getName(), DubboConfigApplicationListener.class);

    // Since 2.7.6 Register DubboConfigDefaultPropertyValueBeanPostProcessor as an infrastructure Bean
    registerInfrastructureBean(registry, DubboConfigDefaultPropertyValueBeanPostProcessor.BEAN_NAME,
        DubboConfigDefaultPropertyValueBeanPostProcessor.class);

    // Dubbo config initializer
    registerInfrastructureBean(registry, DubboConfigBeanInitializer.BEAN_NAME, DubboConfigBeanInitializer.class);

    // register infra bean if not exists later
    registerInfrastructureBean(registry, DubboInfraBeanRegisterPostProcessor.BEAN_NAME, DubboInfraBeanRegisterPostProcessor.class);
}

public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
    String[] beanNames = beanFactory.getBeanDefinitionNames();
    for (String beanName : beanNames) {
        Class<?> beanType;
        if (beanFactory.isFactoryBean(beanName)) {
            BeanDefinition beanDefinition = beanFactory.getBeanDefinition(beanName);
            if (isReferenceBean(beanDefinition)) {
                continue;
            }
            if (isAnnotatedReferenceBean(beanDefinition)) {
                // 处理@DubboReference和@Bean一起使用的场景，如下
                // @Bean
                // @DubboReference(scope = "remote", version = "1.0.0", group = "group1")
                // public ReferenceBean<DemoService> demoService() {
                //     return new ReferenceBean<>();
                // }
                // 这种情况已经注册了该ReferenceBean，所以processReferenceAnnotatedBeanDefinition就
                // 不需要再重复注册，只需要做一些其他的信息设置的工作
                processReferenceAnnotatedBeanDefinition(beanName, (AnnotatedBeanDefinition) beanDefinition);
                continue;
            }
            String beanClassName = beanDefinition.getBeanClassName();
            beanType = ClassUtils.resolveClass(beanClassName, getClassLoader());
        } else {
            beanType = beanFactory.getType(beanName);
        }
        if (beanType != null) {
            // 下面处理通过字段注入的场景，如下
            //    @DubboReference(scope = "remote", version = "1.0.0", group = "group1")
            //    private DemoService demoService;
            // 这种场景需要找到带有@DubboReference注解的字段信息，在prepareInjection方法中注册一个
            // ReferenceBean   
            AnnotatedInjectionMetadata metadata = findInjectionMetadata(beanName, beanType, null);
            try {
                prepareInjection(metadata);
            } catch (BeansException e) {
                throw e;
            } catch (Exception e) {
                throw new IllegalStateException("Prepare dubbo reference injection element failed", e);
            }
        }
    }

    ......

    try {
        // 发布一个DubboConfigInitEvent事件
        applicationContext.publishEvent(new DubboConfigInitEvent(applicationContext));
    } catch (Exception e) {
        logger.warn(CONFIG_DUBBO_BEAN_INITIALIZER, "", "", "publish early application event failed, please upgrade spring version to 4.2.x or later: " + e);
    }
}

public class DubboConfigApplicationListener implements ApplicationListener<DubboConfigInitEvent>, ApplicationContextAware {

    private ApplicationContext applicationContext;

    private ModuleModel moduleModel;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.moduleModel = DubboBeanUtils.getModuleModel(applicationContext);
    }
    @Override
    public void onApplicationEvent(DubboConfigInitEvent event) {
        if (nullSafeEquals(applicationContext, event.getSource())) {
            // It's expected to be notified at org.springframework.context.support.AbstractApplicationContext.registerListeners(),
            // before loading non-lazy singleton beans. At this moment, all BeanFactoryPostProcessor have been processed,
            if (initialized.compareAndSet(false, true)) {
                initDubboConfigBeans();
            }
        }
    }
    private void initDubboConfigBeans() {
        // load DubboConfigBeanInitializer to init config beans
        if (applicationContext.containsBean(DubboConfigBeanInitializer.BEAN_NAME)) {
            applicationContext.getBean(DubboConfigBeanInitializer.BEAN_NAME, DubboConfigBeanInitializer.class);
        } else {
            logger.warn(CONFIG_DUBBO_BEAN_NOT_FOUND, "", "", "Bean '" + DubboConfigBeanInitializer.BEAN_NAME + "' was not found");
        }
        // All infrastructure config beans are loaded, initialize dubbo here
        moduleModel.getDeployer().prepare();
    }
}

private synchronized void initReferenceBean(ReferenceBean referenceBean) throws Exception {
    ......
    // 获取referenceKey 
    String referenceKey = getReferenceKeyByBeanName(referenceBean.getId());
    if (StringUtils.isEmpty(referenceKey)) {
        referenceKey = ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext);
    }
    ......
    ReferenceConfig referenceConfig = referenceConfigMap.get(referenceKey);
    if (referenceConfig == null) {
        
        // 放入缓存
        referenceConfigMap.put(referenceKey, referenceConfig);

        // 和provider一样，将referenceBean添加到ConfigManager中
        moduleModel.getConfigManager().addReference(referenceConfig);
    }
    ......
}

public void onApplicationEvent(ApplicationContextEvent event) {
    if (nullSafeEquals(applicationContext, event.getSource())) {
        if (event instanceof ContextRefreshedEvent) {
            onContextRefreshedEvent((ContextRefreshedEvent) event);
        } else if (event instanceof ContextClosedEvent) {
            onContextClosedEvent((ContextClosedEvent) event);
        }
    }
}

public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    ......
    this.moduleModel = DubboBeanUtils.getModuleModel(applicationContext);
    ......
}

private void onContextRefreshedEvent(ContextRefreshedEvent event) {
    // 通过ModuleModel获取发布器
    ModuleDeployer deployer = moduleModel.getDeployer();
    Assert.notNull(deployer, "Module deployer is null");
    // 开始服务发布的流程
    Future future = deployer.start();
    ......
}

private synchronized Future startSync() throws IllegalStateException {
    ......
    try {
        // new了一个CompletableFuture来以获取start的结果，并且发布一个start事件
        onModuleStarting();
        ......
        // 发布服务
        exportServices();

        // 服务引用
        referServices();
        ......

        // 下面都是根据不同的发布结果，发布不同的事件的处理
        if (asyncExportingFutures.isEmpty() && asyncReferringFutures.isEmpty()) {
            onModuleStarted();
        } else {
            frameworkExecutorRepository.getSharedExecutor().submit(() -> {
                try {
                    // wait for export finish
                    waitExportFinish();
                    // wait for refer finish
                    waitReferFinish();
                } catch (Throwable e) {
                    logger.warn(CONFIG_FAILED_WAIT_EXPORT_REFER, "", "", "wait for export/refer services occurred an exception", e);
                } finally {
                    onModuleStarted();
                }
            });
        }
    } catch (Throwable e) {
        onModuleFailed(getIdentifier() + " start failed: " + e, e);
        throw e;
    }
    return startFuture;
}

private void referServices() {
    configManager.getReferences().forEach(rc -> {
        try {
            ReferenceConfig<?> referenceConfig = (ReferenceConfig<?>) rc;
            if (!referenceConfig.isRefreshed()) {
                referenceConfig.refresh();
            }

            if (rc.shouldInit()) {
                if (referAsync || rc.shouldReferAsync()) {
                    ExecutorService executor = executorRepository.getServiceReferExecutor();
                    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                        try {
                            referenceCache.get(rc);
                        } catch (Throwable t) {
                            logger.error(CONFIG_FAILED_EXPORT_SERVICE, "", "", "Failed to async export service config: " + getIdentifier() + " , catch error : " + t.getMessage(), t);
                        }
                    }, executor);

                    asyncReferringFutures.add(future);
                } else {
                    referenceCache.get(rc);
                }
            }
        } catch (Throwable t) {
            logger.error(CONFIG_FAILED_REFERENCE_MODEL, "", "", "Model reference failed: " + getIdentifier() + " , catch error : " + t.getMessage(), t);
            referenceCache.destroy(rc);
            throw t;
        }
    });
}

public <T> T get(ReferenceConfigBase<T> rc) {
    String key = generator.generateKey(rc);
    Class<?> type = rc.getInterfaceClass();

    boolean singleton = rc.getSingleton() == null || rc.getSingleton();
    T proxy = null;
    // 先从缓存里面去取，其实就是根据key到referenceKeyMap中获取
    if (singleton) {
        proxy = get(key, (Class<T>) type);
    } else {
        logger.warn(CONFIG_API_WRONG_USE, "", "", "Using non-singleton ReferenceConfig and ReferenceCache at the same time may cause memory leak. " +
            "Call ReferenceConfig#get() directly for non-singleton ReferenceConfig instead of using ReferenceCache#get(ReferenceConfig)");
    }

    if (proxy == null) {
        // 缓存中没有，再调用rc.get方法获取
        List<ReferenceConfigBase<?>> referencesOfType = ConcurrentHashMapUtils.computeIfAbsent(referenceTypeMap, type, _t -> Collections.synchronizedList(new ArrayList<>()));
        referencesOfType.add(rc);
        List<ReferenceConfigBase<?>> referenceConfigList = ConcurrentHashMapUtils.computeIfAbsent(referenceKeyMap, key, _k -> Collections.synchronizedList(new ArrayList<>()));
        referenceConfigList.add(rc);
        proxy = rc.get();
    }

    return proxy;
}

public T get() {
    if (destroyed) {
        throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
    }

    if (ref == null) {
        // ensure start module, compatible with old api usage
        getScopeModel().getDeployer().start();

        init();
    }

    return ref;
}

protected synchronized void init() {
    if (initialized && ref != null) {
        return;
    }
    try {
        ......

        ref = createProxy(referenceParameters);

        ......
    } catch (Throwable t) {
        // 异常处理逻辑
        ......
    }
    initialized = true;
}

private T createProxy(Map<String, String> referenceParameters) {
    ......
    createInvoker();
    ......
    // 创建代理对象，默认使用javassist
    return (T) proxyFactory.getProxy(invoker, ProtocolUtils.isGeneric(generic));
}

private void createInvoker() {
    if (urls.size() == 1) {
        URL curUrl = urls.get(0);
        invoker = protocolSPI.refer(interfaceClass, curUrl);
        // registry url, mesh-enable and unloadClusterRelated is true, not need Cluster.
        if (!UrlUtils.isRegistry(curUrl) &&
                !curUrl.getParameter(UNLOAD_CLUSTER_RELATED, false)) {
            List<Invoker<?>> invokers = new ArrayList<>();
            invokers.add(invoker);
            invoker = Cluster.getCluster(getScopeModel(), Cluster.DEFAULT).join(new StaticDirectory(curUrl, invokers), true);
        }
    } else {
        List<Invoker<?>> invokers = new ArrayList<>();
        URL registryUrl = null;
        for (URL url : urls) {
            // For multi-registry scenarios, it is not checked whether each referInvoker is available.
            // Because this invoker may become available later.
            invokers.add(protocolSPI.refer(interfaceClass, url));

            if (UrlUtils.isRegistry(url)) {
                // use last registry url
                registryUrl = 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(registryUrl.getScopeModel(), cluster, false).join(new StaticDirectory(registryUrl, invokers), false);
        } else {
            // not a registry url, must be direct invoke.
            if (CollectionUtils.isEmpty(invokers)) {
                throw new IllegalArgumentException("invokers == null");
            }
            URL curUrl = invokers.get(0).getUrl();
            String cluster = curUrl.getParameter(CLUSTER_KEY, Cluster.DEFAULT);
            invoker = Cluster.getCluster(getScopeModel(), cluster).join(new StaticDirectory(curUrl, invokers), true);
        }
    }
}

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    url = getRegistryUrl(url);
    Registry registry = getRegistry(url);
    if (RegistryService.class.equals(type)) {
        return proxyFactory.getInvoker((T) registry, type, url);
    }

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

    Cluster cluster = Cluster.getCluster(url.getScopeModel(), qs.get(CLUSTER_KEY));
    return doRefer(cluster, registry, type, url, qs);
}

protected <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url, Map<String, String> parameters) {
    Map<String, Object> consumerAttribute = new HashMap<>(url.getAttributes());
    consumerAttribute.remove(REFER_KEY);
    String p = isEmpty(parameters.get(PROTOCOL_KEY)) ? CONSUMER : parameters.get(PROTOCOL_KEY);
    URL consumerUrl = new ServiceConfigURL(
        p,
        null,
        null,
        parameters.get(REGISTER_IP_KEY),
        0, getPath(parameters, type),
        parameters,
        consumerAttribute
    );
    url = url.putAttribute(CONSUMER_URL_KEY, consumerUrl);
    ClusterInvoker<T> migrationInvoker = getMigrationInvoker(this, cluster, registry, type, url, consumerUrl);
    return interceptInvoker(migrationInvoker, url, consumerUrl);
}

protected <T> Invoker<T> interceptInvoker(ClusterInvoker<T> invoker, URL url, URL consumerUrl) {
    List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
    if (CollectionUtils.isEmpty(listeners)) {
        return invoker;
    }

    for (RegistryProtocolListener listener : listeners) {
        listener.onRefer(this, invoker, consumerUrl, url);
    }
    return invoker;
}

public void onRefer(RegistryProtocol registryProtocol, ClusterInvoker<?> invoker, URL consumerUrl, URL registryURL) {
    MigrationRuleHandler<?> migrationRuleHandler = ConcurrentHashMapUtils.computeIfAbsent(handlers, (MigrationInvoker<?>) invoker, _key -> {
        ((MigrationInvoker<?>) invoker).setMigrationRuleListener(this);
        return new MigrationRuleHandler<>((MigrationInvoker<?>) invoker, consumerUrl);
    });

    migrationRuleHandler.doMigrate(rule);
}

private boolean refreshInvoker(MigrationStep step, Float threshold, MigrationRule newRule) {
    if (step == null || threshold == null) {
        throw new IllegalStateException("Step or threshold of migration rule cannot be null");
    }
    MigrationStep originStep = currentStep;

    if ((currentStep == null || currentStep != step) || !currentThreshold.equals(threshold)) {
        boolean success = true;
        switch (step) {
            case APPLICATION_FIRST:
                migrationInvoker.migrateToApplicationFirstInvoker(newRule);
                break;
            case FORCE_APPLICATION:
                success = migrationInvoker.migrateToForceApplicationInvoker(newRule);
                break;
            case FORCE_INTERFACE:
            default:
                success = migrationInvoker.migrateToForceInterfaceInvoker(newRule);
        }

        if (success) {
            setCurrentStepAndThreshold(step, threshold);
            logger.info("Succeed Migrated to " + step + " mode. Service Name: " + consumerURL.getDisplayServiceKey());
            report(step, originStep, "true");
        } else {
            // migrate failed, do not save new step and rule
            logger.warn(INTERNAL_ERROR, "unknown error in registry module", "", "Migrate to " + step + " mode failed. Probably not satisfy the threshold you set "
                + threshold + ". Please try re-publish configuration if you still after check.");
            report(step, originStep, "false");
        }

        return success;
    }
    // ignore if step is same with previous, will continue override rule for MigrationInvoker
    return true;

public void migrateToApplicationFirstInvoker(MigrationRule newRule) {
    CountDownLatch latch = new CountDownLatch(0);
    refreshInterfaceInvoker(latch);
    refreshServiceDiscoveryInvoker(latch);

    // directly calculate preferred invoker, will not wait until address notify
    // calculation will re-occurred when address notify later
    calcPreferredInvoker(newRule);
}

protected void refreshInterfaceInvoker(CountDownLatch latch) {
    clearListener(invoker);
    if (needRefresh(invoker)) {
        if (logger.isDebugEnabled()) {
            logger.debug("Re-subscribing interface addresses for interface " + type.getName());
        }

        if (invoker != null) {
            invoker.destroy();
        }
        invoker = registryProtocol.getInvoker(cluster, registry, type, url);
    }
    setListener(invoker, () -> {
        latch.countDown();
        if (reportService.hasReporter()) {
            reportService.reportConsumptionStatus(
                reportService.createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "interface"));
        }
        if (step == APPLICATION_FIRST) {
            calcPreferredInvoker(rule);
        }
    });
}

public <T> ClusterInvoker<T> getInvoker(Cluster cluster, Registry registry, Class<T> type, URL url) {
    DynamicDirectory<T> directory = new RegistryDirectory<>(type, url);
    return doCreateInvoker(directory, cluster, registry, type);
}

protected <T> ClusterInvoker<T> doCreateInvoker(DynamicDirectory<T> directory, Cluster cluster, Registry registry, Class<T> type) {
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    // all attributes of REFER_KEY
    Map<String, String> parameters = new HashMap<>(directory.getConsumerUrl().getParameters());
    URL urlToRegistry = new ServiceConfigURL(
        parameters.get(PROTOCOL_KEY) == null ? CONSUMER : parameters.get(PROTOCOL_KEY),
        parameters.remove(REGISTER_IP_KEY),
        0,
        getPath(parameters, type),
        parameters
    );
    urlToRegistry = urlToRegistry.setScopeModel(directory.getConsumerUrl().getScopeModel());
    urlToRegistry = urlToRegistry.setServiceModel(directory.getConsumerUrl().getServiceModel());
    if (directory.isShouldRegister()) {
        // 将消费者信息写入注册中心
        directory.setRegisteredConsumerUrl(urlToRegistry);
        registry.register(directory.getRegisteredConsumerUrl());
    }
    directory.buildRouterChain(urlToRegistry);
    // 生成所有提供者的FilterInvoker，此处的Filter是针对单个invoker的Filter（最底层的是DubboInvoker）
    directory.subscribe(toSubscribeUrl(urlToRegistry));
    // 生成集群级别的FilterInvoker，包含了大部分默认的Filter
    return (ClusterInvoker<T>) cluster.join(directory, true);
}

public void subscribe(URL url) {
    ApplicationModel applicationModel = url.getApplicationModel();
    MetricsEventBus.post(RegistryEvent.toSubscribeEvent(applicationModel),() ->
        {
            super.subscribe(url);
            return null;
        }
    );
    if (moduleModel.getModelEnvironment().getConfiguration().convert(Boolean.class, org.apache.dubbo.registry.Constants.ENABLE_CONFIGURATION_LISTEN, true)) {
        consumerConfigurationListener.addNotifyListener(this);
        referenceConfigurationListener = new ReferenceConfigurationListener(moduleModel, this, url);
    }
}

public void subscribe(URL url) {
    setSubscribeUrl(url);
    registry.subscribe(url, this);
}

public void subscribe(URL url, NotifyListener listener) {
    try {
        if (registry != null) {
            // 在这里生成所有提供者的FilterInvoker，并且监听提供者的的变化
            // 这里的registry是一个ZookeeperRegistry实例
            registry.subscribe(url, listener);
        }
    } finally {
        listenerEvent(serviceListener -> serviceListener.onSubscribe(url, registry));
    }
}

public void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        checkDestroyed();
        if (ANY_VALUE.equals(url.getServiceInterface())) {
            // 由于我本地跑的都是具体的interface，所以不会走到这个分支
        } else {
            CountDownLatch latch = new CountDownLatch(1);

            try {
                List<URL> urls = new ArrayList<>();

                /*
                    Iterate over the category value in URL.
                    With default settings, the path variable can be when url is a consumer URL:

                        /dubbo/[service name]/providers,
                        /dubbo/[service name]/configurators
                        /dubbo/[service name]/routers
                */
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = ConcurrentHashMapUtils.computeIfAbsent(zkListeners, url, k -> new ConcurrentHashMap<>());
                    ChildListener zkListener = ConcurrentHashMapUtils.computeIfAbsent(listeners, listener, k -> new RegistryChildListenerImpl(url, k, latch));

                    if (zkListener instanceof RegistryChildListenerImpl) {
                        ((RegistryChildListenerImpl) zkListener).setLatch(latch);
                    }

                    // create "directories".
                    zkClient.create(path, false, true);

                    // 添加提供者的监听器，并且获取到所有的提供者添加到urls中
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    if (children != null) {
                        // The invocation point that may cause 1-1.
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                // 生成提供者的Invoker
                notify(url, listener, urls);
            } finally {
                // tells the listener to run only after the sync notification of main thread finishes.
                latch.countDown();
            }
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

private class RegistryChildListenerImpl implements ChildListener {
    ......
    @Override
    public void childChanged(String path, List<String> children) {
        // Notify 'notifiers' one by one.
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.warn(REGISTRY_ZOOKEEPER_EXCEPTION, "", "", "Zookeeper children listener thread was interrupted unexpectedly, may cause race condition with the main thread.");
        }

        notifier.notify(path, children);
    }
}

protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    if (url == null) {
        throw new IllegalArgumentException("notify url == null");
    }
    if (listener == null) {
        throw new IllegalArgumentException("notify listener == null");
    }
    try {
        doNotify(url, listener, urls);
    } catch (Exception t) {
        // Record a failed registration request to a failed list
        logger.error(REGISTRY_FAILED_NOTIFY_EVENT, "", "", "Failed to notify addresses for subscribe " + url + ", cause: " + t.getMessage(), t);
    }
}

protected synchronized void refreshOverrideAndInvoker(List<URL> urls) {
    // mock zookeeper://xxx?mock=return null
    this.directoryUrl = overrideWithConfigurator(getOriginalConsumerUrl());
    refreshInvoker(urls);
}

public ClusterFilterInvoker(AbstractClusterInvoker<T> invoker) {
    List<FilterChainBuilder> builders = ScopeModelUtil.getApplicationModel(invoker.getUrl().getScopeModel()).getExtensionLoader(FilterChainBuilder.class).getActivateExtensions();
    if (CollectionUtils.isEmpty(builders)) {
        filterInvoker = invoker;
    } else {
        ClusterInvoker<T> tmpInvoker = invoker;
        for (FilterChainBuilder builder : builders) {
            tmpInvoker = builder.buildClusterInvokerChain(tmpInvoker, REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
        }
        filterInvoker = tmpInvoker;
    }
}

private Map<URL, Invoker<T>> toInvokers(Map<URL, Invoker<T>> oldUrlInvokerMap, List<URL> urls) {
    Map<URL, Invoker<T>> newUrlInvokerMap = new ConcurrentHashMap<>(urls == null ? 1 : (int) (urls.size() / 0.75f + 1));
    if (urls == null || urls.isEmpty()) {
        return newUrlInvokerMap;
    }
    String queryProtocols = this.queryMap.get(PROTOCOL_KEY);
    for (URL providerUrl : urls) {
        ......
        Invoker<T> invoker = oldUrlInvokerMap == null ? null : oldUrlInvokerMap.remove(url);
        if (invoker == null) { // Not in the cache, refer again
            try {
                ......
                if (enabled) {
                    // 核心代码就这一行
                    invoker = protocol.refer(serviceType, url);
                }
            } catch (Throwable t) {
                // 异常处理逻辑省略
                ......
            }
            if (invoker != null) { // Put new invoker in cache
                newUrlInvokerMap.put(url, invoker);
            }
        } else {
            newUrlInvokerMap.put(url, invoker);
        }
    }
    return newUrlInvokerMap;
}

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    checkDestroyed();
    return protocolBindingRefer(type, url);
}

@Override
public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
    checkDestroyed();
    optimizeSerialization(url);

    // create rpc invoker.
    DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
    invokers.add(invoker);

    return invoker;
}