Dubbo原理和源码解析之服务暴露

灰太狼 2022-12-27 14:05 116阅读 0赞

> # **github新增仓库 "dubbo-read"（[点此查看][Link 1]），集合所有《Dubbo原理和源码解析》系列文章，后续将继续补充该系列，同时将针对Dubbo所做的功能扩展也进行分享。不定期更新，欢迎Follow。** #

# 一、框架设计 #

在官方《[Dubbo 用户指南][Dubbo]》架构部分，给出了服务调用的整体架构和流程：

![53ba6f8ed285ea8e6037c42fa62aff46.png][]

另外，在官方《[Dubbo 开发指南][Dubbo 1]》框架设计部分，给出了整体设计：

![1d37a4f62971b60b989a5272f390ebc7.png][]

以及暴露服务时序图：

![f191375313c708a9731fa38ffc3af2ab.png][]

本文将根据以上几张图，分析服务暴露的实现原理，并进行详细的代码跟踪与解析。

# 二、原理和源码解析 #

## 2.1 标签解析 ##

从文章《[Dubbo原理和源码解析之标签解析][Dubbo 2]》中我们知道，<dubbo:service> 标签会被解析成 ServiceBean。

ServiceBean 实现了 InitializingBean，在类加载完成之后会调用 afterPropertiesSet() 方法。在 afterPropertiesSet() 方法中，依次解析以下标签信息：

*  <dubbo:provider>
 *  <dubbo:application>
 *  <dubbo:module>
 *  <dubbo:registry>
 *  <dubbo:monitor>
 *  <dubbo:protocol>

ServiceBean 还实现了 ApplicationListener，在 Spring 容器初始化的时候会调用 onApplicationEvent 方法。ServiceBean 重写了 onApplicationEvent 方法，实现了服务暴露的功能。

**ServiceBean.java**

public void onApplicationEvent(ApplicationEvent event) {
        if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
            if (isDelay() && ! isExported() && ! isUnexported()) {
                if (logger.isInfoEnabled()) {
                    logger.info("The service ready on spring started. service: " + getInterface());
                }
                export();
            }
        }
    }

## 2.2 延迟暴露 ##

ServiceBean 扩展了 ServiceConfig，调用 export() 方法时由 ServiceConfig 完成服务暴露的功能实现。

**ServiceConfig.java**

public synchronized void export() {
        if (provider != null) {
            if (export == null) {
                export = provider.getExport();
            }
            if (delay == null) {
                delay = provider.getDelay();
            }
        }
        if (export != null && ! export.booleanValue()) {
            return;
        }
        if (delay != null && delay > 0) {
            Thread thread = new Thread(new Runnable() {
                public void run() {
                    try {
                        Thread.sleep(delay);
                    } catch (Throwable e) {
                    }
                    doExport();
                }
            });
            thread.setDaemon(true);
            thread.setName("DelayExportServiceThread");
            thread.start();
        } else {
            doExport();
        }
    }

由上面代码可知，如果设置了 delay 参数，Dubbo 的处理方式是启动一个守护线程在 sleep 指定时间后再 doExport。

## 2.3 参数检查 ##

在 ServiceConfig 的 doExport() 方法中会进行参数检查和设置，包括：

*  泛化调用
 *  本地实现
 *  本地存根
 *  本地伪装
 *  配置（application、registry、protocol等）

**ServiceConfig.java**

protected synchronized void doExport() {
        if (unexported) {
            throw new IllegalStateException("Already unexported!");
        }
        if (exported) {
            return;
        }
        exported = true;
        if (interfaceName == null || interfaceName.length() == 0) {
            throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");
        }
        checkDefault();
        //省略
        if (ref instanceof GenericService) {
            interfaceClass = GenericService.class;
            generic = true;
        } else {
            try {
                interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
                        .getContextClassLoader());
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            checkInterfaceAndMethods(interfaceClass, methods);
            checkRef();
            generic = false;
        }
        if(local !=null){
            if(local=="true"){
                local=interfaceName+"Local";
            }
            Class<?> localClass;
            try {
                localClass = ClassHelper.forNameWithThreadContextClassLoader(local);
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            if(!interfaceClass.isAssignableFrom(localClass)){
                throw new IllegalStateException("The local implemention class " + localClass.getName() + " not implement interface " + interfaceName);
            }
        }
        if(stub !=null){
            if(stub=="true"){
                stub=interfaceName+"Stub";
            }
            Class<?> stubClass;
            try {
                stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub);
            } catch (ClassNotFoundException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            if(!interfaceClass.isAssignableFrom(stubClass)){
                throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + interfaceName);
            }
        }
        //此处省略：检查并设置相关参数
        doExportUrls();
    }

## 2.4 多协议、多注册中心 ##

在检查完参数之后，开始暴露服务。Dubbo 支持多协议和多注册中心：

**ServiceConfig.java**

private void doExportUrls() {
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

## 2.5 组装URL ##

针对每个协议、每个注册中心，开始组装 URL。

**ServiceConfig.java**

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        String name = protocolConfig.getName();
        if (name == null || name.length() == 0) {
            name = "dubbo";
        }
    
        //处理host
    
        //处理port
    
        Map<String, String> map = new HashMap<String, String>();
        //设置参数到map
    
        // 导出服务
        String contextPath = protocolConfig.getContextpath();
        if ((contextPath == null || contextPath.length() == 0) && provider != null) {
            contextPath = provider.getContextpath();
        }
        URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
    
        if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                .hasExtension(url.getProtocol())) {
            url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                    .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
        }
    
        //此处省略：服务暴露（详见 2.6 和 2.7）
        
        this.urls.add(url);
    }

## 2.6 本地暴露 ##

如果配置 scope=none， 则不会进行服务暴露；如果没有配置 scope 或者 scope=local，则会进行本地暴露。

**ServiceConfig.java**

//public static final String LOCAL_PROTOCOL = "injvm";
    //public static final String LOCALHOST = "127.0.0.1";
    
    private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        //......
        String scope = url.getParameter(Constants.SCOPE_KEY);
        //配置为none不暴露
        if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
            //配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务)
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            //......
        }
        //......
    }
    
    private void exportLocal(URL url) {
        if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
            URL local = URL.valueOf(url.toFullString())
                    .setProtocol(Constants.LOCAL_PROTOCOL)
                    .setHost(NetUtils.LOCALHOST)
                    .setPort(0);
            Exporter<?> exporter = protocol.export(
                    proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
            exporters.add(exporter);
            logger.info("Export dubbo service " + interfaceClass.getName() +" to local registry");
        }
    }

> 1. 暴露服务的时候，会通过代理创建 Invoker；
> 
> 2. 本地暴露时使用 injvm 协议，injvm 协议是一个伪协议，它不开启端口，不能被远程调用，只在 JVM 内直接关联，但执行 Dubbo 的 Filter 链。

## 2.7 远程暴露 ##

如果没有配置 scope 或者 scope=remote，则会进行远程暴露。

**ServiceConfig.java**

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        String scope = url.getParameter(Constants.SCOPE_KEY);
        //配置为none不暴露
        if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
            //......
            //如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露远程服务)
            if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
                if (logger.isInfoEnabled()) {
                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                }
                if (registryURLs != null && registryURLs.size() > 0
                        && url.getParameter("register", true)) {
                    for (URL registryURL : registryURLs) {
                        url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
                        URL monitorUrl = loadMonitor(registryURL);
                        if (monitorUrl != null) {
                            url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
                        }
                        if (logger.isInfoEnabled()) {
                            logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                        }
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
        
                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                } else {
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
        
                    Exporter<?> exporter = protocol.export(invoker);
                    exporters.add(exporter);
                }
            }
        }
    }

在服务暴露时，有两种情况：

*  不使用注册中心：直接暴露对应协议的服务，引用服务时只能通过直连方式引用
 *  使用注册中心：暴露对应协议的服务后，会将服务节点注册到注册中心，引用服务时可以通过注册中心动态获取服务提供者列表，也可以通过直连方式引用

## 2.8 暴露服务 ##

不使用注册中心时，直接调用对应协议（如 Dubbo 协议）的 export() 暴露服务。以 Dubbo 协议为例：

**DubboProtocol.java**

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();
        
        // export service.
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);
        
        //export an stub service for dispaching event
        Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
        Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
        if (isStubSupportEvent && !isCallbackservice){
            String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
            if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
                if (logger.isWarnEnabled()){
                    logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
                            "], has set stubproxy support event ,but no stub methods founded."));
                }
            } else {
                stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
            }
        }
    
        openServer(url);
        
        return exporter;
    }

调用 openServer() 方法创建并启动 Server：

**DubboProtocol.java**

private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        //client 也可以暴露一个只有server可以调用的服务。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY,true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                serverMap.put(key, createServer(url));
            } else {
                //server支持reset,配合override功能使用
                server.reset(url);
            }
        }
    }
    
    private ExchangeServer createServer(URL url) {
        //默认开启server关闭时发送readonly事件
        url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
        //默认开启heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
        String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
    
        if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);
    
        url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
        ExchangeServer server;
        try {
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        str = url.getParameter(Constants.CLIENT_KEY);
        if (str != null && str.length() > 0) {
            Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
            if (!supportedTypes.contains(str)) {
                throw new RpcException("Unsupported client type: " + str);
            }
        }
        return server;
    }

Exchanger （默认 HeaderExchanger）封装请求响应模式，同步转异步，以 Request、Response 为中心:

**HeaderExchager.java**

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }

**Transporters.java**

public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        if (handlers == null || handlers.length == 0) {
            throw new IllegalArgumentException("handlers == null");
        }
        ChannelHandler handler;
        if (handlers.length == 1) {
            handler = handlers[0];
        } else {
            handler = new ChannelHandlerDispatcher(handlers);
        }
        return getTransporter().bind(url, handler);
    }

底层传输默认使用 NettyTransporter，最终是创建 NettyServer：

**NettyTransporter.java**

public Server bind(URL url, ChannelHandler listener) throws RemotingException {
        return new NettyServer(url, listener);
    }

**NettyServer.java**

public class NettyServer extends AbstractServer implements Server {
        public NettyServer(URL url, ChannelHandler handler) throws RemotingException{
            super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
        }
    }

**AbstractServer.java**

public abstract class AbstractServer extends AbstractEndpoint implements Server {
        public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
            super(url, handler);
            localAddress = getUrl().toInetSocketAddress();
            String host = url.getParameter(Constants.ANYHOST_KEY, false) 
                            || NetUtils.isInvalidLocalHost(getUrl().getHost()) 
                            ? NetUtils.ANYHOST : getUrl().getHost();
            bindAddress = new InetSocketAddress(host, getUrl().getPort());
            this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
            this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
            try {
                doOpen();
                if (logger.isInfoEnabled()) {
                    logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
                }
            } catch (Throwable t) {
                throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() 
                                            + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
            }
            if (handler instanceof WrappedChannelHandler ){
                executor = ((WrappedChannelHandler)handler).getExecutor();
            }
        }
    }

**NettyServer.java**

public class NettyServer extends AbstractServer implements Server {
        protected void doOpen() throws Throwable {
            NettyHelper.setNettyLoggerFactory();
            ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
            ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
            ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
            bootstrap = new ServerBootstrap(channelFactory);
            
            final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
            channels = nettyHandler.getChannels();
            // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
            bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
                public ChannelPipeline getPipeline() {
                    NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
                    ChannelPipeline pipeline = Channels.pipeline();
                    /*int idleTimeout = getIdleTimeout();
                    if (idleTimeout > 10000) {
                        pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
                    }*/
                    pipeline.addLast("decoder", adapter.getDecoder());
                    pipeline.addLast("encoder", adapter.getEncoder());
                    pipeline.addLast("handler", nettyHandler);
                    return pipeline;
                }
            });
            // bind
            channel = bootstrap.bind(getBindAddress());
        }
    }

## 2.9 服务注册 ##

如果使用了注册中心，则在通过具体协议（如 Dubbo 协议）暴露服务之后（即在 2.8 基础之上）进入服务注册流程，将服务节点注册到注册中心。

**RegistryProtocol.java**

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        //registry provider
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
        //向Zookeeper注册节点
        registry.register(registedProviderUrl);
        // 订阅override数据
        // FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        //保证每次export都返回一个新的exporter实例
        return new Exporter<T>() {
            public Invoker<T> getInvoker() {
                return exporter.getInvoker();
            }
            public void unexport() {
                try {
                    exporter.unexport();
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    registry.unregister(registedProviderUrl);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
                try {
                    overrideListeners.remove(overrideSubscribeUrl);
                    registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        };
    }
    
    private Registry getRegistry(final Invoker<?> originInvoker){
        URL registryUrl = originInvoker.getUrl();
        if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
            String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
            registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
        }
        return registryFactory.getRegistry(registryUrl);
    }

getRegistry() 方法根据注册中心类型（默认 Zookeeper）获取注册中心客户端，由注册中心客户端实例来进行真正的服务注册。

注册中心客户端将节点注册到注册中心，同时订阅对应的 override 数据，实时监听服务的属性变动实现动态配置功能。

最终返回的 Exporter 实现了 unexport() 方法，这样在服务下线时清理相关资源。

至此，服务暴露流程结束。

[Link 1]: https://github.com/cyfonly/dubbo-read
[Dubbo]: http://dubbo.apache.org/books/dubbo-user-book/preface/architecture.html
[53ba6f8ed285ea8e6037c42fa62aff46.png]: /images/20221120/e12d3efb66584edf8bc7599f1db9a5fc.png
[Dubbo 1]: http://dubbo.apache.org/books/dubbo-dev-book/design.html
[1d37a4f62971b60b989a5272f390ebc7.png]: /images/20221120/43eea00bddee41d98b854726ef03c9e2.png
[f191375313c708a9731fa38ffc3af2ab.png]: /images/20221120/4207ced61f79429ca5ce025fc90d52bd.png
[Dubbo 2]: https://blog.csdn.net/chuixue24/article/details/111038616