dubbo-服务暴露过程之网络通信创建

上面分析了dubbo服务对配置的加载解析，接下来dubbo要完成开启网络通信，将服务注册到注册中心以供客户端发现以及通过网络通信完成远程调用。

spring ApplicationListener监听

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware {
    /**
     * Spring 容器完成事件触发器 ，服务监听ContextRefreshedEvent事件，ApplicationContext初始化完成或刷新触发
     * @param event
     */
    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());
                }
                //spring 容器加载完毕后，暴露当前Service服务
                export();
            }
        }
    }

根据ServiceBean可以看到我们配置的<dubbo:service/> 都会继承ApplicationListener并监听ContextRefreshedEvent事件，在spring容器加载完毕后调用export()进行服务暴露，那么我们跟踪下export源码看下具体实现

/**
     * 当前服务暴露方法，负责将当前服务暴露到远程注册中心上,此方法为线程安全
     */
    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;
        }
        //delay >0 表示延迟加载暴露，使用线程在delay后暴露服务
        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属性功能的实现，即延迟暴露，接下来调用了内部方法doExport进行继续暴露，doExport方法同样完成了对各种配置信息的加载，补全，以及对信息进行校验，完成后调用doExportUrls进行继续暴露

/**
     * 暴露服务，此方法供doExport调用，此方法负责将当前服务根据服务协议（ProtocolConfig）进行暴露
     */
    @SuppressWarnings({"unchecked", "rawtypes"})
    private void doExportUrls() {
        //加载服务配置的所有注册中心(RegistryConfig)URL
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            //依次根据服务协议将服务暴露到注册中心上
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

此方法与上面方法一样，同样完了了一些前置的条件，如注册中心URL的加载，并根据ProtocolConfig配置信息依次调用doExportUrlsFor1Protocol

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
      ......
      //方法比较长省略一些不太主要的部分，此部分主要完成了对URL创建的map的组装
        //根据参数信息生产暴露服务的URL
        URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

        //获取当前服务暴露的范围此参数可在配置中配置
        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);
            }
            //如果配置不是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);
                        }
                        //通过外部提供的实例(ref)获取这个实例的反射实例
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                        //此处的URL protocol 是registry  对应的处理protocol 为Protocol$Adpative 实际根据registry 查找出RegistryProtocol实例处理
                        // 通过protocol暴露服务，并将服务的包装类型返回 这个protocol实例是关键部分，很容易读不懂
                        //调用过程根据protocol值来加载实际的protocol
                        //此处的调用顺序是Protocol$Adpative.export() -> RegistryProtocol.export()->Protocol$Adpative.export() -> dubboProtocol.export()
                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                } 
            }
        }
        this.urls.add(url);
    }

此方法可以清晰的看到通过这句Exporter<?> exporter = protocol.export(invoker);进行暴露，protocol是一个自适应的扩展实例，具体在SPI机制中再详细说明，现在我只要知道调用此方法最终的调用顺序如下：

Protocol$Adpative.export() -> RegistryProtocol.export()->Protocol$Adpative.export() -> dubboProtocol.export()

限于目前我们分析的网络通信的创建 ，对RegistryProtocol.export()就不描述，我们直接看dubboProtocol.export的实现方法

/**
     * dubbo 协议暴露服务方法
     * @param invoker 服务的执行体
     * @param <T>
     * @return
     * @throws RpcException
     */
    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);
        //通过此方法来打开一个网络通信服务段
        openServer(url);
        return exporter;
    }

打开服务：openServer

private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        //client 也可以暴露一个只有server可以调用的服务。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
           // 首先在缓存中获取给定key信息交换层服务
            ExchangeServer server = serverMap.get(key);
            //当本地没有这个网络通信的服务创建新的
            if (server == null) {
              //创建一个底层通信服务key为IP:PORT,并将服务保存到缓存中
                serverMap.put(key, createServer(url));
            } else {
                //server支持reset,配合override功能使用
                server.reset(url);
            }
        }
    }

创建服务：createServer

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));
        //获取底层的通信组件名称，默认为netty,如果配置文件中配置了，根据配置文件
        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 {
            //创建一个信息交换层服务 默认获取的是HeaderExchanger
            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;
    }

上面方法我们看到实际HeaderExchanger.bind()，那么我们再看看bind到底做了什么

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
 //这里可以看到返回创建一个HeaderExchangeServer
 //并且通过Transporters这个创建一个网络传输服务
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }

最终调用NettyTransporter.bind()方法

/**
 * @author ding.lid
 */
public class NettyTransporter implements Transporter {

    public static final String NAME = "netty";

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

    public Client connect(URL url, ChannelHandler listener) throws RemotingException {
        return new NettyClient(url, listener);
    }

}

自此dubbo从配置加载以及到网络通信服务的建立就完成了，我们可以清晰的看到dubbo从上层我们配置到底层网络通信创建的调用链

ServiceBean.onApplicationEvent->export->doExport->doExportUrlsFor1Protocol->dubboProtocol.export->openServer->createServer->HeaderExchanger.bind->NettyTransporter.bind