DubboProtocol 服务发布

开篇

 这篇文章的目的是尝试讲解Dubbo Protocol使用Netty Server作为服务端的服务发布过程，为了便于说明清楚，在网上找了一张Dubbo服务发布的流程图，这里我们关注Protocol到Server的部分逻辑，其他的分布其他文章。

DubboProtocol

public class DubboProtocol extends AbstractProtocol {

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

        // 生成exporter对象并保存至exporterMap
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);

        // 开始开启服务端监控
        openServer(url);
        optimizeSerialization(url);

        return exporter;
    }

    private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        //client can export a service which's only for server to invoke
        boolean isServer = url.getParameter(IS_SERVER_KEY, true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                synchronized (this) {
                    server = serverMap.get(key);
                    if (server == null) {
                        // 通过createServer创建Server对象
                        serverMap.put(key, createServer(url));
                    }
                }
            } else {
                // server supports reset, use together with override
                server.reset(url);
            }
        }
    }


    private ExchangeServer createServer(URL url) {

        // 省略非核心代码
        ExchangeServer server;
        try {
            // 服务发布的入口，requestHandler包含invokerMap的对象
            // server 是HeaderExchangeServer对象
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }

        return server;
    }

    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
        // 省略代码
    }

}

• DubboProtocol的export()方法生成DubboExporter对象并openServer()开启Server。
• openServer()方法内部执行createServer(url)创建Server对象。
• createServer()方法内部生成ExchangeServer对象。
• ExchangeServer对象通过Exchangers.bind(url, requestHandler)生成。
• requestHandler是ExchangeHandlerAdapter对象，内部包含invokerMap对象。
• server对象在这里我们可以看出来是ExchangeServer，通过Exchangers生成。

Exchangers

public class Exchangers {
    public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
        if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        }
        url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
        return getExchanger(url).bind(url, handler);
    }

    // String DEFAULT_EXCHANGER = "header";
    public static Exchanger getExchanger(URL url) {
        String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
        return getExchanger(type);
    }

    // header=org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger
    public static Exchanger getExchanger(String type) {
        return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
    }
}

• Exchangers的bind()方法内部执行getExchanger(url).bind(url, handler)。
• getExchanger(url).bind(url, handler)中getExchange()通过ExtensionLoader获取Exchanger扩展，这里是HeaderExchanger对象。
• 继续执行HeaderExchanger的bind()方法。
• 这里的bind()方法的参数是ExchangeHandlerAdapter的requestHandler。

public class HeaderExchanger implements Exchanger {

    public static final String NAME = "header";

    @Override
    public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeClient(Transporters.connect(url, 
               new DecodeHandler(new HeaderExchangeHandler(handler))), true);
    }

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

• 继续执行HeaderExchanger的bind()方法，生成HeaderExchangeServer对象。
• HeaderExchangeServer对象的参数通过Transporters.bind()生成的Server对象。
• bind()的参数是通过DecodeHandler => HeaderExchangeHandler => requestHandler包装生成的DecodeHandler对象。
• 继续观察Transporters.bind()方法。

Transporters

public class Transporters {

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


    public static Transporter getTransporter() {
        return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
    }
}

• Transporters.bind()方法内部执行getTransporter().bind(url, handler)。
• 调用通过getTransporter()方法返回Transporter$Adaptive。
• getTransporter().bind()执行Transporter$Adaptive的bind()方法。
• 继续观察Transporter$Adaptive的bind()方法

public class Transporter$Adaptive implements Transporter {
  
    public Server bind(URL uRL, ChannelHandler channelHandler)
        throws RemotingException {

        if (uRL == null) {
            throw new IllegalArgumentException("url == null");
        }

        URL uRL2 = uRL;
        String string = uRL2.getParameter("server",
                uRL2.getParameter("transporter", "netty"));

        Transporter transporter = (Transporter) ExtensionLoader
         .getExtensionLoader(Transporter.class)
         .getExtension(string);

        return transporter.bind(uRL, channelHandler);
    }
}


public class NettyTransporter implements Transporter {

    public static final String NAME = "netty3";

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

• Transporter$Adaptive的bind()方法内部执行获取Transporter扩展并执行bind()方法。
• ExtensionLoader.getExtensionLoader().getExtension()返回NettyTransporter对象。
• 这里的ChannelHandler为DecodeHandler对象。
• 继续执行NettyTransporter的bind()方法，内部创建NettyServer对象。

NettyServer

public class NettyServer extends AbstractServer implements Server {

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    // 保存连接的Channel信息
    private Map<String, Channel> channels; // <ip:port, channel>

    private ServerBootstrap bootstrap;

    private org.jboss.netty.channel.Channel channel;

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

    @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(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.setOption("child.tcpNoDelay", true);
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            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());
    }

• 创建NettyServer对象并通过doOpen()方法创建Netty Server对象。
• doOpen()方法内部创建ServerBootstrap对象并实现bind启动服务监听。
• NettyServer构造函数内部通过ChannelHandlers.wrap()包装ChannelHandler。

public class ChannelHandlers {

    // 关注核心代码逻辑
    protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
        return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                .getAdaptiveExtension().dispatch(handler, url)));
    }
}

public class Dispatcher$Adaptive implements Dispatcher {
    public ChannelHandler dispatch(ChannelHandler channelHandler, URL uRL) {
        if (uRL == null) {
            throw new IllegalArgumentException("url == null");
        }

        URL uRL2 = uRL;
        String string = uRL2.getParameter("dispatcher",
                uRL2.getParameter("dispather",
                    uRL2.getParameter("channel.handler", "all")));

        Dispatcher dispatcher = (Dispatcher) ExtensionLoader.getExtensionLoader(Dispatcher.class)
                                                            .getExtension(string);

        return dispatcher.dispatch(channelHandler, uRL);
    }
}



public class AllDispatcher implements Dispatcher {

    public static final String NAME = "all";

    @Override
    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new AllChannelHandler(handler, url);
    }
}

• wrapInternal()的handler参数通过MultiMessageHandler => HeartbeatHandler => AllChannelHandler => DecodeHandler层层封装。
• Dispatcher$Adaptive的ExtensionLoader.getExtensionLoader().getExtension()返回AllDispatcher对象。
• AllDispatcher的dispatch()方法返回AllChannelHandler对象。
• AllChannelHandler封装DecodeHandler对象。

public abstract class AbstractServer extends AbstractEndpoint implements Server {

    protected static final String SERVER_THREAD_POOL_NAME = "DubboServerHandler";
    private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
    ExecutorService executor;
    private InetSocketAddress localAddress;
    private InetSocketAddress bindAddress;
    // 最大连接数
    private int accepts;
    private int idleTimeout;

    public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
        super(url, handler);
        localAddress = getUrl().toInetSocketAddress();

        String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
        int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
        if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
            bindIp = ANYHOST_VALUE;
        }
        bindAddress = new InetSocketAddress(bindIp, bindPort);
        // 最大连接数从URL当中获取
        this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
        this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT);
        try {
            // 开启服务监听
            doOpen();
        } catch (Throwable t) {
        }
        //fixme replace this with better method
        DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
        executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort()));
    }
}

• NettyServer的父类AbstractServer包含了核心的最大连接数的参数。
• AbstractServer的调用抽象方法doOpen()调用子类NettyServer启动服务监听。

handler封装关系图

• 描述创建Server服务 过程中Handler的封装关系。