dubbo - NettyServer源码解析

开篇

 这篇文章的主要目的是记录下Dubbo当中NettyServer针对请求的处理过程，主要是理顺连接建立的过程和调用处理的过程。

 这个解析的核心在于理解类的继承关系（可以理解为设计模式当中的装饰者模式），层层递进理解调用关系。另外通过时序图能够帮助理解整个时序图的调用过程。

NettyServer

public abstract class AbstractPeer implements Endpoint, ChannelHandler {
    private final ChannelHandler handler;

    public AbstractPeer(URL url, ChannelHandler handler) {
        this.handler = handler;
    }

    public void connected(Channel ch) throws RemotingException {
        if (closed) {
            return;
        }
        handler.connected(ch);
    }

    public void disconnected(Channel ch) throws RemotingException {
        handler.disconnected(ch);
    }

    public void sent(Channel ch, Object msg) throws RemotingException {
        if (closed) {
            return;
        }
        handler.sent(ch, msg);
    }

    public void received(Channel ch, Object msg) throws RemotingException {
        if (closed) {
            return;
        }
        handler.received(ch, msg);
    }

    public void caught(Channel ch, Throwable ex) throws RemotingException {
        handler.caught(ch, ex);
    }
}

说明：

• Dubbo中NettyServer的继承关系图如上图所示，核心的变量在基类当中。
• 关注最顶层的基类AbstractPeer变量包含ChannelHandler对象。
• AbstractPeer的ChannelHandler变量是由NettyServer的构造函数中传入的。

public class NettyServer extends AbstractServer implements Server {

    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(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
        bootstrap = new ServerBootstrap(channelFactory);
        // NettyHandler传入的是NettyServer对象
        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        channels = nettyHandler.getChannels();

        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() {
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                ChannelPipeline pipeline = Channels.pipeline();
                pipeline.addLast("decoder", adapter.getDecoder());
                pipeline.addLast("encoder", adapter.getEncoder());
                // 核心关键nettyHandler
                pipeline.addLast("handler", nettyHandler);
                return pipeline;
            }
        });
        // bind
        channel = bootstrap.bind(getBindAddress());
    }
}


public class ChannelHandlers {

    private static ChannelHandlers INSTANCE = new ChannelHandlers();

    public static ChannelHandler wrap(ChannelHandler handler, URL url) {
        return ChannelHandlers.getInstance().wrapInternal(handler, url);
    }

    protected static ChannelHandlers getInstance() {
        return INSTANCE;
    }

    protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
        return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                .getAdaptiveExtension().dispatch(handler, url)));
    }
}


public class AllDispatcher implements Dispatcher {

    public static final String NAME = "all";

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

• 核心需要关注NettyServer的构造函数和doOpen()方法。
• NettyServer的构造函数传入的是DecodeHandler对象。
• NettyServer的构造函数通过ChannelHandlers.wrap()封装ChannelHandler对象。
• wrapInternal()方法按照MultiMessageHandler包含HeartbeatHandler，HeartbeatHandler包含AllChannelHandler的顺序进行封装，类似于装饰者模式。
• getAdaptiveExtension().dispatch()方法返回AllChannelHandler对象。
• NettyHandler的构造方法将NettyServer的this作为ChannelHandler对象传参。
• pipeline.addLast("handler", nettyHandler)将nettyHandler加到Netty的职责链中。
• NettyHandler作为处理的入口函数，NettyServer的this自身包括处理的逻辑。

NettyTransporter

public class HeaderExchanger implements Exchanger {

    public static final String NAME = "header";

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

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


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

说明：

• NettyTransporter的目的是为了打通和NettyServer的初始化过程形成串联。
• NettyTransporter的bind()方法传入的是DecodeHandler对象。
• DecodeHandler的构造函数参数是HeaderExchangeHandler对象。
• HeaderExchangeHandler的构造函数参数是DubboProtocol对象。
• NettyServer的构造函数的参数是DecodeHandler对象。

NettyHandler

public class NettyHandler extends SimpleChannelHandler {

    private final Map<String, Channel> channels = new ConcurrentHashMap<String, Channel>(); // <ip:port, channel>
    private final URL url;
    private final ChannelHandler handler;

    public NettyHandler(URL url, ChannelHandler handler) {
        this.url = url;
        this.handler = handler;
    }

    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            if (channel != null) {
                channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel);
            }
            handler.connected(channel);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            channels.remove(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()));
            handler.disconnected(channel);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.received(channel, e.getMessage());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        super.writeRequested(ctx, e);
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.sent(channel, e.getMessage());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.caught(channel, e.getCause());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

}

说明

• NettyHandler的channelConnected和channelDisconnected用于处理连接。
• NettyHandler的messageReceived和writeRequested用于处理消息。
• NettyHandler的exceptionCaught用于处理异常。
• NettyHandler构造函数的的参数handler是NettyServer对象。
• NettyHandler的操作通过调用NettyServer的方法connected/received完成。

Dubbo-Netty的Receive过程

说明：

• 1.NettyHandler
• 2.NettyServer
• 3.MultiMessageHandler
• 4.HeartbeatHandler
• 5.AllChannelHandler
• 6.ChannelEventRunnable
• 7.DecodeHandler
• 8.HeaderExchangeHandler
• 9.DubboProtocol.ExchangeHandlerAdapter

Dubbo-Netty的Connect过程

说明：

• 1.NettyHandler
• 2.NettyServer
• 3.MultiMessageHandler
• 4.HeartbeatHandler
• 5.AllChannelHandler
• 6.ChannelEventRunnable
• 7.DecodeHandler
• 8.HeaderExchangeHandler
• 9.DubboProtocol.ExchangeHandlerAdapter