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

一、框架设计

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

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

以及暴露服务时序图：

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

二、原理和源码解析

2.1 标签解析

从文章《Dubbo原理和源码解析之标签解析》中我们知道，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

1public void onApplicationEvent(ApplicationEvent event) {
2    if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
3        if (isDelay() && ! isExported() && ! isUnexported()) {
4            if (logger.isInfoEnabled()) {
5                logger.info("The service ready on spring started. service: " + getInterface());
6            }
7            export();
8        }
9    }
10}

2.2 延迟暴露

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

ServiceConfig.java

1public synchronized void export() {
2    if (provider != null) {
3        if (export == null) {
4            export = provider.getExport();
5        }
6        if (delay == null) {
7            delay = provider.getDelay();
8        }
9    }
10    if (export != null && ! export.booleanValue()) {
11        return;
12    }
13    if (delay != null && delay > 0) {
14        Thread thread = new Thread(new Runnable() {
15            public void run() {
16                try {
17                    Thread.sleep(delay);
18                } catch (Throwable e) {
19                }
20                doExport();
21            }
22        });
23        thread.setDaemon(true);
24        thread.setName("DelayExportServiceThread");
25        thread.start();
26    } else {
27        doExport();
28    }
29}

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

2.3 参数检查

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

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

ServiceConfig.java

1protected synchronized void doExport() {
2    if (unexported) {
3        throw new IllegalStateException("Already unexported!");
4    }
5    if (exported) {
6        return;
7    }
8    exported = true;
9    if (interfaceName == null || interfaceName.length() == 0) {
10        throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");
11    }
12    checkDefault();
13    //省略
14    if (ref instanceof GenericService) {
15        interfaceClass = GenericService.class;
16        generic = true;
17    } else {
18        try {
19            interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
20                    .getContextClassLoader());
21        } catch (ClassNotFoundException e) {
22            throw new IllegalStateException(e.getMessage(), e);
23        }
24        checkInterfaceAndMethods(interfaceClass, methods);
25        checkRef();
26        generic = false;
27    }
28    if(local !=null){
29        if(local=="true"){
30            local=interfaceName+"Local";
31        }
32        Class<?> localClass;
33        try {
34            localClass = ClassHelper.forNameWithThreadContextClassLoader(local);
35        } catch (ClassNotFoundException e) {
36            throw new IllegalStateException(e.getMessage(), e);
37        }
38        if(!interfaceClass.isAssignableFrom(localClass)){
39            throw new IllegalStateException("The local implemention class " + localClass.getName() + " not implement interface " + interfaceName);
40        }
41    }
42    if(stub !=null){
43        if(stub=="true"){
44            stub=interfaceName+"Stub";
45        }
46        Class<?> stubClass;
47        try {
48            stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub);
49        } catch (ClassNotFoundException e) {
50            throw new IllegalStateException(e.getMessage(), e);
51        }
52        if(!interfaceClass.isAssignableFrom(stubClass)){
53            throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + interfaceName);
54        }
55    }
56    //此处省略：检查并设置相关参数
57    doExportUrls();
58}

2.4 多协议、多注册中心

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

ServiceConfig.java

1private void doExportUrls() {
2    List<URL> registryURLs = loadRegistries(true);
3    for (ProtocolConfig protocolConfig : protocols) {
4        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
5    }
6}

2.5 组装URL

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

ServiceConfig.java

1private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
2    String name = protocolConfig.getName();
3    if (name == null || name.length() == 0) {
4        name = "dubbo";
5    }
6
7    //处理host
8
9    //处理port
10
11    Map<String, String> map = new HashMap<String, String>();
12    //设置参数到map
13
14    // 导出服务
15    String contextPath = protocolConfig.getContextpath();
16    if ((contextPath == null || contextPath.length() == 0) && provider != null) {
17        contextPath = provider.getContextpath();
18    }
19    URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
20
21    if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
22            .hasExtension(url.getProtocol())) {
23        url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
24                .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
25    }
26
27    //此处省略：服务暴露（详见 2.6 和 2.7）
28    
29    this.urls.add(url);
30}

2.6 本地暴露

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

ServiceConfig.java

1//public static final String LOCAL_PROTOCOL = "injvm";
2//public static final String LOCALHOST = "127.0.0.1";
3
4private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
5    //......
6    String scope = url.getParameter(Constants.SCOPE_KEY);
7    //配置为none不暴露
8    if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
9        //配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务)
10        if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
11            exportLocal(url);
12        }
13        //......
14    }
15    //......
16}
17
18private void exportLocal(URL url) {
19    if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
20        URL local = URL.valueOf(url.toFullString())
21                .setProtocol(Constants.LOCAL_PROTOCOL)
22                .setHost(NetUtils.LOCALHOST)
23                .setPort(0);
24        Exporter<?> exporter = protocol.export(
25                proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
26        exporters.add(exporter);
27        logger.info("Export dubbo service " + interfaceClass.getName() +" to local registry");
28    }
29}

1. 暴露服务的时候，会通过代理创建 Invoker；

2. 本地暴露时使用 injvm 协议，injvm 协议是一个伪协议，它不开启端口，不能被远程调用，只在 JVM 内直接关联，但执行 Dubbo 的 Filter 链。

2.7 远程暴露

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

ServiceConfig.java

1private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
2    String scope = url.getParameter(Constants.SCOPE_KEY);
3    //配置为none不暴露
4    if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
5        //......
6        //如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露远程服务)
7        if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
8            if (logger.isInfoEnabled()) {
9                logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
10            }
11            if (registryURLs != null && registryURLs.size() > 0
12                    && url.getParameter("register", true)) {
13                for (URL registryURL : registryURLs) {
14                    url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
15                    URL monitorUrl = loadMonitor(registryURL);
16                    if (monitorUrl != null) {
17                        url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
18                    }
19                    if (logger.isInfoEnabled()) {
20                        logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
21                    }
22                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
23    
24                    Exporter<?> exporter = protocol.export(invoker);
25                    exporters.add(exporter);
26                }
27            } else {
28                Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
29    
30                Exporter<?> exporter = protocol.export(invoker);
31                exporters.add(exporter);
32            }
33        }
34    }
35}

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

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

2.8 暴露服务

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

DubboProtocol.java

1public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
2    URL url = invoker.getUrl();
3    
4    // export service.
5    String key = serviceKey(url);
6    DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
7    exporterMap.put(key, exporter);
8    
9    //export an stub service for dispaching event
10    Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
11    Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
12    if (isStubSupportEvent && !isCallbackservice){
13        String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
14        if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
15            if (logger.isWarnEnabled()){
16                logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
17                        "], has set stubproxy support event ,but no stub methods founded."));
18            }
19        } else {
20            stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
21        }
22    }
23
24    openServer(url);
25    
26    return exporter;
27}

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

DubboProtocol.java

1private void openServer(URL url) {
2    // find server.
3    String key = url.getAddress();
4    //client 也可以暴露一个只有server可以调用的服务。
5    boolean isServer = url.getParameter(Constants.IS_SERVER_KEY,true);
6    if (isServer) {
7        ExchangeServer server = serverMap.get(key);
8        if (server == null) {
9            serverMap.put(key, createServer(url));
10        } else {
11            //server支持reset,配合override功能使用
12            server.reset(url);
13        }
14    }
15}
16
17private ExchangeServer createServer(URL url) {
18    //默认开启server关闭时发送readonly事件
19    url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
20    //默认开启heartbeat
21    url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
22    String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
23
24    if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
25        throw new RpcException("Unsupported server type: " + str + ", url: " + url);
26
27    url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
28    ExchangeServer server;
29    try {
30        server = Exchangers.bind(url, requestHandler);
31    } catch (RemotingException e) {
32        throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
33    }
34    str = url.getParameter(Constants.CLIENT_KEY);
35    if (str != null && str.length() > 0) {
36        Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
37        if (!supportedTypes.contains(str)) {
38            throw new RpcException("Unsupported client type: " + str);
39        }
40    }
41    return server;
42}

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

HeaderExchager.java

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

Transporters.java

1public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
2    if (url == null) {
3        throw new IllegalArgumentException("url == null");
4    }
5    if (handlers == null || handlers.length == 0) {
6        throw new IllegalArgumentException("handlers == null");
7    }
8    ChannelHandler handler;
9    if (handlers.length == 1) {
10        handler = handlers[0];
11    } else {
12        handler = new ChannelHandlerDispatcher(handlers);
13    }
14    return getTransporter().bind(url, handler);
15}

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

NettyTransporter.java

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

NettyServer.java

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

AbstractServer.java

1public abstract class AbstractServer extends AbstractEndpoint implements Server {
2    public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
3        super(url, handler);
4        localAddress = getUrl().toInetSocketAddress();
5        String host = url.getParameter(Constants.ANYHOST_KEY, false) 
6                        || NetUtils.isInvalidLocalHost(getUrl().getHost()) 
7                        ? NetUtils.ANYHOST : getUrl().getHost();
8        bindAddress = new InetSocketAddress(host, getUrl().getPort());
9        this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
10        this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
11        try {
12            doOpen();
13            if (logger.isInfoEnabled()) {
14                logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
15            }
16        } catch (Throwable t) {
17            throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() 
18                                        + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
19        }
20        if (handler instanceof WrappedChannelHandler ){
21            executor = ((WrappedChannelHandler)handler).getExecutor();
22        }
23    }
24}

NettyServer.java

1public class NettyServer extends AbstractServer implements Server {
2    protected void doOpen() throws Throwable {
3        NettyHelper.setNettyLoggerFactory();
4        ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
5        ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
6        ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
7        bootstrap = new ServerBootstrap(channelFactory);
8        
9        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
10        channels = nettyHandler.getChannels();
11        // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
12        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
13            public ChannelPipeline getPipeline() {
14                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
15                ChannelPipeline pipeline = Channels.pipeline();
16                /*int idleTimeout = getIdleTimeout();
17                if (idleTimeout > 10000) {
18                    pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
19                }*/
20                pipeline.addLast("decoder", adapter.getDecoder());
21                pipeline.addLast("encoder", adapter.getEncoder());
22                pipeline.addLast("handler", nettyHandler);
23                return pipeline;
24            }
25        });
26        // bind
27        channel = bootstrap.bind(getBindAddress());
28    }
29}

2.9 服务注册

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

RegistryProtocol.java

1public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
2    //export invoker
3    final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
4    //registry provider
5    final Registry registry = getRegistry(originInvoker);
6    final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
7    //向Zookeeper注册节点
8    registry.register(registedProviderUrl);
9    // 订阅override数据
10    // FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。
11    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
12    final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
13    overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
14    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
15    //保证每次export都返回一个新的exporter实例
16    return new Exporter<T>() {
17        public Invoker<T> getInvoker() {
18            return exporter.getInvoker();
19        }
20        public void unexport() {
21            try {
22                exporter.unexport();
23            } catch (Throwable t) {
24                logger.warn(t.getMessage(), t);
25            }
26            try {
27                registry.unregister(registedProviderUrl);
28            } catch (Throwable t) {
29                logger.warn(t.getMessage(), t);
30            }
31            try {
32                overrideListeners.remove(overrideSubscribeUrl);
33                registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
34            } catch (Throwable t) {
35                logger.warn(t.getMessage(), t);
36            }
37        }
38    };
39}
40
41private Registry getRegistry(final Invoker<?> originInvoker){
42    URL registryUrl = originInvoker.getUrl();
43    if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
44        String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
45        registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
46    }
47    return registryFactory.getRegistry(registryUrl);
48}

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

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

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

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