转载

聊聊SpringCloudRegistryFactory

序

本文主要研究一下SpringCloudRegistryFactory

SpringCloudRegistryFactory

spring-cloud-alibaba-2.1.0.RELEASE/spring-cloud-alibaba-dubbo/src/main/java/com/alibaba/cloud/dubbo/registry/SpringCloudRegistryFactory.java

public class SpringCloudRegistryFactory implements RegistryFactory {

    public static String PROTOCOL = "spring-cloud";

    public static String ADDRESS = "localhost";

    private static String SERVICES_LOOKUP_SCHEDULER_THREAD_NAME_PREFIX = getProperty(
            "dubbo.services.lookup.scheduler.thread.name.prefix ",
            "dubbo-services-lookup-");

    private static ConfigurableApplicationContext applicationContext;

    private DiscoveryClient discoveryClient;

    private DubboServiceMetadataRepository dubboServiceMetadataRepository;

    private DubboMetadataServiceProxy dubboMetadataConfigServiceProxy;

    private JSONUtils jsonUtils;

    private volatile boolean initialized = false;

    public SpringCloudRegistryFactory() {
    }

    public static void setApplicationContext(
            ConfigurableApplicationContext applicationContext) {
        SpringCloudRegistryFactory.applicationContext = applicationContext;
    }

    protected void init() {
        if (initialized || applicationContext == null) {
            return;
        }
        this.discoveryClient = applicationContext.getBean(DiscoveryClient.class);
        this.dubboServiceMetadataRepository = applicationContext
                .getBean(DubboServiceMetadataRepository.class);
        this.dubboMetadataConfigServiceProxy = applicationContext
                .getBean(DubboMetadataServiceProxy.class);
        this.jsonUtils = applicationContext.getBean(JSONUtils.class);
    }

    @Override
    public Registry getRegistry(URL url) {
        init();
        return new SpringCloudRegistry(url, discoveryClient,
                dubboServiceMetadataRepository, dubboMetadataConfigServiceProxy,
                jsonUtils, applicationContext);
    }
}

SpringCloudRegistryFactory实现了RegistryFactory接口，其getRegistry方法首先调用init方法，然后创建并返回SpringCloudRegistry；init方法会从spring容器中获取DiscoveryClient、DubboServiceMetadataRepository、DubboMetadataServiceProxy、JSONUtils

SpringCloudRegistry

spring-cloud-alibaba-2.1.0.RELEASE/spring-cloud-alibaba-dubbo/src/main/java/com/alibaba/cloud/dubbo/registry/SpringCloudRegistry.java

public class SpringCloudRegistry extends AbstractSpringCloudRegistry {

    private final DubboServiceMetadataRepository dubboServiceMetadataRepository;

    public SpringCloudRegistry(URL url, DiscoveryClient discoveryClient,
            DubboServiceMetadataRepository dubboServiceMetadataRepository,
            DubboMetadataServiceProxy dubboMetadataConfigServiceProxy,
            JSONUtils jsonUtils, ConfigurableApplicationContext applicationContext) {
        super(url, discoveryClient, dubboServiceMetadataRepository,
                dubboMetadataConfigServiceProxy, jsonUtils, applicationContext);
        this.dubboServiceMetadataRepository = dubboServiceMetadataRepository;
    }

    @Override
    protected void doRegister0(URL url) {
        dubboServiceMetadataRepository.exportURL(url);
    }

    @Override
    protected void doUnregister0(URL url) {
        dubboServiceMetadataRepository.unexportURL(url);
    }
}

SpringCloudRegistry继承了AbstractSpringCloudRegistry，其doRegister0执行dubboServiceMetadataRepository.exportURL，doUnregister0执行dubboServiceMetadataRepository.unexportURL

AbstractSpringCloudRegistry

spring-cloud-alibaba-2.1.0.RELEASE/spring-cloud-alibaba-dubbo/src/main/java/com/alibaba/cloud/dubbo/registry/AbstractSpringCloudRegistry.java

public abstract class AbstractSpringCloudRegistry extends FailbackRegistry {

    /**
     * The parameter name of {@link #servicesLookupInterval}
     */
    public static final String SERVICES_LOOKUP_INTERVAL_PARAM_NAME = "dubbo.services.lookup.interval";

    protected static final String DUBBO_METADATA_SERVICE_CLASS_NAME = DubboMetadataService.class
            .getName();
    /**
     * Caches the IDs of {@link ApplicationListener}
     */
    private static final Set<String> registerListeners = new HashSet<>();

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * The interval in second of lookup service names(only for Dubbo-OPS)
     */
    private final long servicesLookupInterval;

    private final DiscoveryClient discoveryClient;

    private final DubboServiceMetadataRepository repository;

    private final DubboMetadataServiceProxy dubboMetadataConfigServiceProxy;

    private final JSONUtils jsonUtils;

    private final ConfigurableApplicationContext applicationContext;

    public AbstractSpringCloudRegistry(URL url, DiscoveryClient discoveryClient,
            DubboServiceMetadataRepository dubboServiceMetadataRepository,
            DubboMetadataServiceProxy dubboMetadataConfigServiceProxy,
            JSONUtils jsonUtils, ConfigurableApplicationContext applicationContext) {
        super(url);
        this.servicesLookupInterval = url
                .getParameter(SERVICES_LOOKUP_INTERVAL_PARAM_NAME, 60L);
        this.discoveryClient = discoveryClient;
        this.repository = dubboServiceMetadataRepository;
        this.dubboMetadataConfigServiceProxy = dubboMetadataConfigServiceProxy;
        this.jsonUtils = jsonUtils;
        this.applicationContext = applicationContext;
    }

    //......

    @Override
    public final void doSubscribe(URL url, NotifyListener listener) {

        if (isAdminURL(url)) {
            // TODO in future
        }
        else if (isDubboMetadataServiceURL(url)) { // for DubboMetadataService
            subscribeDubboMetadataServiceURLs(url, listener);
        }
        else { // for general Dubbo Services
            subscribeDubboServiceURLs(url, listener);
        }
    }

    protected void subscribeDubboServiceURLs(URL url, NotifyListener listener) {

        doSubscribeDubboServiceURLs(url, listener);

        registerServiceInstancesChangedEventListener(url, listener);
    }

    private void doSubscribeDubboServiceURLs(URL url, NotifyListener listener) {

        Set<String> subscribedServices = repository.getSubscribedServices();
        // Sync
        subscribedServices.forEach(service -> subscribeDubboServiceURL(url, listener,
                service, this::getServiceInstances));
    }

    private List<ServiceInstance> getServiceInstances(String serviceName) {
        return hasText(serviceName) ? doGetServiceInstances(serviceName) : emptyList();
    }

    private List<ServiceInstance> doGetServiceInstances(String serviceName) {
        List<ServiceInstance> serviceInstances = emptyList();
        try {
            serviceInstances = discoveryClient.getInstances(serviceName);
        }
        catch (Exception e) {
            if (logger.isErrorEnabled()) {
                logger.error(e.getMessage(), e);
            }
        }
        return serviceInstances;
    }

    protected void subscribeDubboServiceURL(URL url, NotifyListener listener,
            String serviceName,
            Function<String, Collection<ServiceInstance>> serviceInstancesFunction) {

        if (logger.isInfoEnabled()) {
            logger.info(
                    "The Dubbo Service URL[ID : {}] is being subscribed for service[name : {}]",
                    generateId(url), serviceName);
        }

        DubboMetadataService dubboMetadataService = dubboMetadataConfigServiceProxy
                .getProxy(serviceName);

        if (dubboMetadataService == null) { // If not found, try to initialize
            if (logger.isInfoEnabled()) {
                logger.info(
                        "The metadata of Dubbo service[key : {}] can't be found when the subscribed service[name : {}], "
                                + "and then try to initialize it",
                        url.getServiceKey(), serviceName);
            }
            repository.initializeMetadata(serviceName);
            dubboMetadataService = dubboMetadataConfigServiceProxy.getProxy(serviceName);
        }

        if (dubboMetadataService == null) { // It makes sure not-found, return immediately
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "The metadata of Dubbo service[key : {}] still can't be found, it could effect the further "
                                + "Dubbo service invocation",
                        url.getServiceKey());
            }
            return;
        }

        Collection<ServiceInstance> serviceInstances = serviceInstancesFunction
                .apply(serviceName);

        List<URL> allSubscribedURLs = new LinkedList<>();

        if (CollectionUtils.isEmpty(serviceInstances)) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "There is no instance from service[name : {}], and then Dubbo Service[key : {}] will not be "
                                + "available , please make sure the further impact",
                        serviceName, url.getServiceKey());
            }
            /**
             * URLs with {@link RegistryConstants#EMPTY_PROTOCOL}
             */
            allSubscribedURLs.addAll(emptyURLs(url));
        }
        else {
            List<URL> exportedURLs = getExportedURLs(dubboMetadataService, url);

            for (URL exportedURL : exportedURLs) {
                String protocol = exportedURL.getProtocol();
                List<URL> subscribedURLs = new LinkedList<>();
                serviceInstances.forEach(serviceInstance -> {
                    Integer port = repository.getDubboProtocolPort(serviceInstance,
                            protocol);
                    String host = serviceInstance.getHost();
                    if (port == null) {
                        if (logger.isWarnEnabled()) {
                            logger.warn(
                                    "The protocol[{}] port of Dubbo  service instance[host : {}] "
                                            + "can't be resolved",
                                    protocol, host);
                        }
                    }
                    else {
                        URL subscribedURL = new URL(protocol, host, port,
                                exportedURL.getParameters());
                        subscribedURLs.add(subscribedURL);
                    }
                });

                allSubscribedURLs.addAll(subscribedURLs);
            }
        }

        if (logger.isDebugEnabled()) {
            logger.debug("The subscribed URL[{}] will notify all URLs : {}", url,
                    allSubscribedURLs);
        }

        listener.notify(allSubscribedURLs);
    }

    //......

}

AbstractSpringCloudRegistry的doGetServiceInstances方法会使用discoveryClient.getInstances来获取ServiceInstance列表；subscribeDubboServiceURL方法会从ServiceInstance列表读取并组装subscribedURLs，然后使用listener.notify将这些subscribedURLs回传回去

小结

SpringCloudRegistryFactory实现了RegistryFactory接口，其getRegistry方法首先调用init方法，然后创建并返回SpringCloudRegistry；init方法会从spring容器中获取DiscoveryClient、DubboServiceMetadataRepository、DubboMetadataServiceProxy、JSONUtils
SpringCloudRegistry继承了AbstractSpringCloudRegistry，其doRegister0执行dubboServiceMetadataRepository.exportURL，doUnregister0执行dubboServiceMetadataRepository.unexportURL
AbstractSpringCloudRegistry的doGetServiceInstances方法会使用discoveryClient.getInstances来获取ServiceInstance列表；subscribeDubboServiceURL方法会从ServiceInstance列表读取并组装subscribedURLs，然后使用listener.notify将这些subscribedURLs回传回去