微服务架构和实践Dubbo剖析程序员

dubbo剖析:三 网络通信之 -- Server实现

2018-04-22  本文已影响176人  益文的圈

引子:

一、入口流程

服务发布流程中,RegistryProtocol会调用到DubboProtocolexport()方法,用于完成网络服务的启动和监听。

1.1 入口代码

    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();

        //step1 export service.
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);

        //...省略部分代码...

        //step2 创建ExchangeServer
        openServer(url);

        return exporter;
    }
    private void openServer(URL url) {
        String key = url.getAddress();
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                //调用createServer
                serverMap.put(key, createServer(url));
            } else {
                server.reset(url);
            }
        }
    }
    private ExchangeServer createServer(URL url) {
        //...省略部分代码,参数解析之类的...
        ExchangeServer server;
        try {
            //关键代码,使用HeaderExchanger.bind创建HeaderExchangeServer
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        return server;
    }

1.2 流程图解

DubboProtocol.export()流程图

第一步:生成Exporter:

第二步:创建ExchangeServer:

二、server端网络层结构

server端网络层类图关系说明

2.1 网络传输层

2.2 信息交换层

2.3 网络通道Channel

网络通道接口定义

三、HeaderExchangeServer & NettyServer实现详解

Server实现层次结构图

3.1 网络层

AbstractPeer类(网络事件处理器和网络节点的通用实现):

AbstractEndPoint类(加入编解码功能):

AbstractServer类(网络服务端通用抽象,抽象出openclosesend的公共流程,并提供了doOpendoClose的实现扩展):

NettyServer类(网络服务端Netty实现类,实现了doOpendoClosegetChannels三个具体扩展):

    @Override
    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();
        // https://issues.jboss.org/browse/NETTY-365
        // https://issues.jboss.org/browse/NETTY-379
        // 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());
    }
    @Override
    protected void doClose() throws Throwable {
        try {
            if (channel != null) {
                // unbind.
                channel.close();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            Collection<com.alibaba.dubbo.remoting.Channel> channels = getChannels();
            if (channels != null && channels.size() > 0) {
                for (com.alibaba.dubbo.remoting.Channel channel : channels) {
                    try {
                        channel.close();
                    } catch (Throwable e) {
                        logger.warn(e.getMessage(), e);
                    }
                }
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            if (bootstrap != null) {
                // release external resource.
                bootstrap.releaseExternalResources();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            if (channels != null) {
                channels.clear();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
    }

3.2 交换层

HeaderExchangeServer类(交换层服务端,将网络层的Channel扩展为交换层的ExchangeChannel,并加入心跳检测功能):

上一篇 下一篇

猜你喜欢

热点阅读