Dubbo学习之ServiceBean

相关阅读

Dubbo学习之PostProcessor
Dubbo学习之DubboReference
Dubbo学习之ReferenceBean
Dubbo学习之DubboService

简介

本文基于Spring Boot 2.6.6，dubbo-spring-boot-starter 3.0.6环境。

由上文Dubbo学习之DubboService可知，标注了注解DubboService的Bean最终会被注册为ServiceBean；本文主要分析ServiceBean的实例化过程，以及Dubbo Service的创建过程；

Demo

核心依赖：


    org.apache.dubbo
    dubbo-spring-boot-starter
    3.0.6



    org.apache.dubbo
    dubbo-registry-zookeeper
    3.0.6

生产者核心配置参数：

server:
  port: 8080
dubbo:
  application:
    id: dubbo-provider
    name: dubbo-provider
  registry:
    address: zookeeper://127.0.0.1:2181
  protocol:
    port: 20880
  scan:
    base-packages: demo.dubbo.producer.config

生产者Dubbo Service配置代码：

package demo.dubbo.producer.config;

@Configuration
public class ServiceConfig {

    @Bean
    @DubboService
    public DemoService demoServiceImpl() {
        return new DemoServiceImpl();
    }
}

ServiceBean实例化初始化

ServiceBean实现了InitializingBean接口，故在被创建后进行初始化时，会调用其afterPropertiesSet方法，代码如下：

public void afterPropertiesSet() throws Exception {
    if (StringUtils.isEmpty(getPath())) {
        if (StringUtils.isNotEmpty(getInterface())) {
            setPath(getInterface());
        }
    }
    
    ModuleModel moduleModel = DubboBeanUtils.getModuleModel(applicationContext);
    // 添加Service
    moduleModel.getConfigManager().addService(this);
    // 设置状态为PENDING
    moduleModel.getDeployer().setPending();
}

在afterPropertiesSet方法中将自身添加到ModuleConfigManager中，代码如下：

public void addService(ServiceConfigBase serviceConfig) {
    addConfig(serviceConfig);
}

public final  T addConfig(AbstractConfig config) {
    if (config == null) {
        return null;
    }
    // 此处会忽视MethodConfig
    if (!isSupportConfigType(config.getClass())) {
        throw new IllegalArgumentException("Unsupported config type: " + config);
    }

    if (config.getScopeModel() != scopeModel) {
        config.setScopeModel(scopeModel);
    }

    Map configsMap = configsCache.computeIfAbsent(getTagName(config.getClass()), type -> new ConcurrentHashMap<>());

    if (!(config instanceof ReferenceConfigBase || config instanceof ServiceConfigBase)) {
        for (AbstractConfig value : configsMap.values()) {
            if (value.equals(config)) {
                return (T) value;
            }
        }
    }

    synchronized (configsMap) {
        return (T) addIfAbsent(config, configsMap);
    }
}

private  C addIfAbsent(C config, Map configsMap)
    throws IllegalStateException {

    if (config == null || configsMap == null) {
        return config;
    }

    Optional prevConfig = findDuplicatedConfig(configsMap, config);
    if (prevConfig.isPresent()) {
        // 如果已存在，则直接返回
        return prevConfig.get();
    }

    String key = config.getId();
    if (key == null) {
        do {
            // 构造Key直至configsMap中不存在该Key
            key = generateConfigId(config);
        } while (configsMap.containsKey(key));
    }

    // 校验是否已存在Config
    C existedConfig = configsMap.get(key);
    if (existedConfig != null && !isEquals(existedConfig, config)) {
        String type = config.getClass().getSimpleName();
        logger.warn(String.format("Duplicate %s found, there already has one default %s or more than two %ss have the same id, " +
                "you can try to give each %s a different id, override previous config with later config. id: %s, prev: %s, later: %s",
            type, type, type, type, key, existedConfig, config));
    }

    // 保存当前Config
    configsMap.put(key, config);
    return config;
}

核心便是构建ReferenceConfig（用于后续创建Dubbo Reference），该动作由ReferenceCreator.build()完成，代码如下：

public final ReferenceConfig build() throws Exception {
    ReferenceConfig configBean = new ReferenceConfig();

    // 配置ReferenceConfig
    configureBean(configBean);

    if (logger.isInfoEnabled()) {
        logger.info("The configBean[type:" + configBean.getClass().getSimpleName() + "] has been built.");
    }

    return configBean;
}

protected void configureBean(ReferenceConfig configBean) throws Exception {
    // 填充属性
    populateBean(configBean);

    // deprecate application reference
    //configureApplicationConfig(configBean);

    // 配置Monitor
    configureMonitorConfig(configBean);
    // 配置Module
    configureModuleConfig(configBean);
    // 配置Consumer
    configureConsumerConfig(configBean);
}

至此，ServiceBean的实例化流程就结束了，ServiceBean继承自ServiceConfig，后续会被用于创建Dubbo Reference；

Dubbo Service创建

当AbstractApplicationContext完成fresh后，就会发布事件ContextRefreshedEvent，Dubbo中DubboDeployApplicationListener会关注该事件，并做相关处理，代码如下：

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

private void onContextRefreshedEvent(ContextRefreshedEvent event) {
    ModuleDeployer deployer = moduleModel.getDeployer();
    Assert.notNull(deployer, "Module deployer is null");
    // 启动Module
    Future future = deployer.start();

    if (!deployer.isBackground()) {
        try {
            // 同步等待Module启动完成
            future.get();
        } catch (InterruptedException e) {
            logger.warn("Interrupted while waiting for dubbo module start: " + e.getMessage());
        } catch (Exception e) {
            logger.warn("An error occurred while waiting for dubbo module start: " + e.getMessage(), e);
        }
    }
}

当Dubbo监听到ContextRefreshedEvent，便启动ModuleModel，代码如下：

public synchronized Future start() throws IllegalStateException {
    if (isStopping() || isStopped() || isFailed()) {
        throw new IllegalStateException(getIdentifier() + " is stopping or stopped, can not start again");
    }

    try {
        if (isStarting() || isStarted()) {
            // 正在启动或启动完成，则直接返回
            return startFuture;
        }

        // 设置STARTING状态
        onModuleStarting();

        applicationDeployer.initialize();
        initialize();

        // 暴露Service
        // 即处理@DubboService产生的ServiceBean
        exportServices();

    // 先启动内部ModuleModel若两者不一致的话
    if (moduleModel != moduleModel.getApplicationModel().getInternalModule()) {
        applicationDeployer.prepareInternalModule();
    }

        // 引用Service
        // 生产者端无处理
        referServices();

        if (asyncExportingFutures.isEmpty() && asyncReferringFutures.isEmpty()) {
            // 直接设置STARTED状态
            onModuleStarted();
        } else {
            executorRepository.getSharedExecutor().submit(() -> {
                try {
                    // 等待暴露Service完成
                    waitExportFinish();
                    // 等待引用Service完成
                    waitReferFinish();
                } catch (Throwable e) {
                    logger.warn("wait for export/refer services occurred an exception", e);
                } finally {
                    // 最后设置STARTED状态
                    onModuleStarted();
                }
            });
        }
    } catch (Throwable e) {
        onModuleFailed(getIdentifier() + " start failed: " + e, e);
        throw e;
    }
    return startFuture;
}

private void exportServices() {
    // 遍历所有的ServiceConfig进行暴露
    for (ServiceConfigBase sc : configManager.getServices()) {
        exportServiceInternal(sc);
    }
}

private void exportServiceInternal(ServiceConfigBase sc) {
    ServiceConfig serviceConfig = (ServiceConfig) sc;
    if (!serviceConfig.isRefreshed()) {
        // 还没刷新则刷新ServiceConfig的属性若需要的话
        serviceConfig.refresh();
    }
    if (sc.isExported()) {
        // 已暴露则直接退出
        return;
    }
    if (exportAsync || sc.shouldExportAsync()) {
        // 异步方式
        ExecutorService executor = executorRepository.getServiceExportExecutor();
        CompletableFuture future = CompletableFuture.runAsync(() -> {
            try {
                if (!sc.isExported()) {
                    sc.export();
                    exportedServices.add(sc);
                }
            } catch (Throwable t) {
                logger.error(getIdentifier() + " export async catch error : " + t.getMessage(), t);
            }
        }, executor);

        asyncExportingFutures.add(future);
    } else {
        // 同步方式
        if (!sc.isExported()) {
            // 暴露服务
            sc.export();
            // 缓存已暴露的服务
            exportedServices.add(sc);
        }
    }
}

继续分析ServiceConfig.export()方法，代码如下：

public void export() {
    if (this.exported) {
        // 如果已暴露，则直接退出
        return;
    }

    // 确保Module启动
    getScopeModel().getDeployer().start();

    synchronized (this) {
        // DCL
        if (this.exported) {
            return;
        }

        if (!this.isRefreshed()) {
            this.refresh();
        }
        if (this.shouldExport()) {
            // 暴露前的准备动作
            // 设置serviceMetadata相关属性
            this.init();

            if (shouldDelay()) {
                // 延迟暴露
                doDelayExport();
            } else {
                // 执行暴露
                doExport();
            }
        }
    }
}

protected void doDelayExport() {
    // 延迟执行doExport
    getScopeModel().getDefaultExtension(ExecutorRepository.class).getServiceExportExecutor()
        .schedule(() -> {
            try {
                doExport();
            } catch (Exception e) {
                logger.error("Failed to export service config: " + interfaceName, e);
            }
        }, getDelay(), TimeUnit.MILLISECONDS);
}

protected synchronized void doExport() {
    if (unexported) {
        // 如果执行过unexport，此时却进行export，则抛出异常
        throw new IllegalStateException("The service " + interfaceClass.getName() + " has already unexported!");
    }
    if (exported) {
        // 如果已暴露，则直接退出
        return;
    }

    if (StringUtils.isEmpty(path)) {
        path = interfaceName;
    }
    doExportUrls();
    exported();
}

private void doExportUrls() {
    ModuleServiceRepository repository = getScopeModel().getServiceRepository();
    // 注册Service
    ServiceDescriptor serviceDescriptor = repository.registerService(getInterfaceClass());
    providerModel = new ProviderModel(getUniqueServiceName(),
        ref,
        serviceDescriptor,
        this,
        getScopeModel(),
        serviceMetadata);

    // 注册Provider
    repository.registerProvider(providerModel);

    List registryURLs = ConfigValidationUtils.loadRegistries(this, true);

    for (ProtocolConfig protocolConfig : protocols) {
        String pathKey = URL.buildKey(getContextPath(protocolConfig)
                .map(p -> p + "/" + path)
                .orElse(path), group, version);
        // 如果用户指定了path，则将注册的Service映射到该path
        repository.registerService(pathKey, interfaceClass);
        // 向registryURLs暴露服务
        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
    }
}

doExportUrlsFor1Protocol方法中根据ProtocolConfig构建出服务的URL，然后通过exportUrl将服务暴露出去，exportUrl代码如下：

private void exportUrl(URL url, List registryURLs) {
    String scope = url.getParameter(SCOPE_KEY);
    // 未指定必须是NONE，则执行
    if (!SCOPE_NONE.equalsIgnoreCase(scope)) {

        // 未指定必须是REMOTE，则执行
        if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
            // 暴露到INJVM
            exportLocal(url);
        }

        // 未指定必须是LOCAL，则执行
        if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
            url = exportRemote(url, registryURLs);
            if (!isGeneric(generic) && !getScopeModel().isInternal()) {
                MetadataUtils.publishServiceDefinition(url, providerModel.getServiceModel(), getApplicationModel());
            }
        }
    }
    this.urls.add(url);
}

exportLocal和exportRemote中，构建出最终的URL后，执行doExportUrl，相关代码如下：

private void exportLocal(URL url) {
    ...
    doExportUrl(local, false);
    ...
}

private URL exportRemote(URL url, List registryURLs) {
    ...
    doExportUrl(registryURL.putAttribute(EXPORT_KEY, url), true);
    ...
}

private void doExportUrl(URL url, boolean withMetaData) {
    Invoker invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
    if (withMetaData) {
        // 需要ServiceConfig，则使用包装器
        invoker = new DelegateProviderMetaDataInvoker(invoker, this);
    }
    // 借助SPI机制，根据URL的协议名称调用对应Protocol方法
    Exporter exporter = protocolSPI.export(invoker);
    exporters.add(exporter);
}

exportRemote方法中URL的协议名称为：registry，则调用RegistryProtocol.export方法，代码如下：

public  Exporter export(final Invoker originInvoker) throws RpcException {
    URL registryUrl = getRegistryUrl(originInvoker);
    // url to export locally
    URL providerUrl = getProviderUrl(originInvoker);

    // Subscribe the override data
    // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call
    //  the same service. Because the subscribed is cached key with the name of the service, it causes the
    //  subscription information to cover.
    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
    final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
    Map overrideListeners = getProviderConfigurationListener(providerUrl).getOverrideListeners();
    overrideListeners.put(registryUrl, overrideSubscribeListener);

    providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
    // 暴露Invoker
    final ExporterChangeableWrapper exporter = doLocalExport(originInvoker, providerUrl);

    // url to registry
    final Registry registry = getRegistry(registryUrl);
    final URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);

    boolean register = providerUrl.getParameter(REGISTER_KEY, true) && registryUrl.getParameter(REGISTER_KEY, true);
    if (register) {
        register(registry, registeredProviderUrl);
    }

    registerStatedUrl(registryUrl, registeredProviderUrl, register);

    exporter.setRegisterUrl(registeredProviderUrl);
    exporter.setSubscribeUrl(overrideSubscribeUrl);

    if (!registry.isServiceDiscovery()) {
        // Deprecated! Subscribe to override rules in 2.6.x or before.
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
    }

    notifyExport(exporter);
    //Ensure that a new exporter instance is returned every time export
    return new DestroyableExporter<>(exporter);
}

private  ExporterChangeableWrapper doLocalExport(final Invoker originInvoker, URL providerUrl) {
    // 获取Service URL
    String key = getCacheKey(originInvoker);

    return (ExporterChangeableWrapper) bounds.computeIfAbsent(key, s -> {
        Invoker invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
        // 启动Service
        // 借助SPI机制，根据URL的协议名称调用对应Protocol方法
        return new ExporterChangeableWrapper<>((Exporter) protocol.export(invokerDelegate), originInvoker);
    });
}

本例采用Dubbo协议，故会调用DubboProtocol.export方法，代码如下：

public  Exporter export(Invoker invoker) throws RpcException {
    checkDestroyed();
    // 获取Service URL
    URL url = invoker.getUrl();

    String key = serviceKey(url);
    DubboExporter exporter = new DubboExporter(invoker, key, exporterMap);

    Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT);
    Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false);
    if (isStubSupportEvent && !isCallbackservice) {
        String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY);
        if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
            if (logger.isWarnEnabled()) {
                logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) +
                        "], has set stubproxy support event ,but no stub methods founded."));
            }

        }
    }

    // 启动服务
    openServer(url);
    optimizeSerialization(url);

    return exporter;
}

private void openServer(URL url) {
    checkDestroyed();
    String key = url.getAddress();
    boolean isServer = url.getParameter(IS_SERVER_KEY, true);
    if (isServer) {
        ProtocolServer server = serverMap.get(key);
        if (server == null) {
            synchronized (this) {
                server = serverMap.get(key);
                if (server == null) {
                    // 服务不存在，则创建服务
                    serverMap.put(key, createServer(url));
                }else {
                    // 服务存在，则重置服务
                    server.reset(url);
                }
            }
        } else {
            // 服务存在，则重置服务
            // 同一个Dubbo服务，可以存在多个Service URL，这些Service共用同一个地址：IP:PORT
            server.reset(url);
        }
    }
}

private ProtocolServer createServer(URL url) {
    url = URLBuilder.from(url)
            .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
            // 心跳默认60s
            .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
            .addParameter(CODEC_KEY, DubboCodec.NAME)
            .build();
    // 默认为netty
    String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER);

    if (StringUtils.isNotEmpty(str) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(str)) {
        throw new RpcException("Unsupported server type: " + str + ", url: " + url);
    }

    ExchangeServer server;
    try {
        // 根据Service URL和requestHandler构建ExchangeServer
        // 最终创建NettyServer，并封装为HeaderExchangeServer
        server = Exchangers.bind(url, requestHandler);
    } catch (RemotingException e) {
        throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
    }

    str = url.getParameter(CLIENT_KEY);
    if (StringUtils.isNotEmpty(str)) {
        Set supportedTypes = url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).getSupportedExtensions();
        if (!supportedTypes.contains(str)) {
            throw new RpcException("Unsupported client type: " + str);
        }
    }

    // 包装为DubboProtocolServer
    DubboProtocolServer protocolServer = new DubboProtocolServer(server);
    loadServerProperties(protocolServer);
    return protocolServer;
}

总结

接口调用流程如下：

标注了注解DubboService的Bean最终在Spring容器中是以ServiceBean（targetSource为DubboReferenceLazyInitTargetSource）形式存在；
ServiceBean实例化时，会将自身作为ServiceConfig加入ModuleConfigManager；
DubboDeployApplicationListener监听到ContextRefreshedEvent后，便会启动Module；
Module启动过程中，会暴露ModuleConfigManager中所有的ServiceConfig，从而完成Dubbo Service创建；

Dubbo学习之ServiceBean

云平台相关栏目本月热门文章