dubbo - NettyServer源码解析-阿里云开发者社区

开发者社区> 开发与运维> 正文

dubbo - NettyServer源码解析

简介: 开篇  这篇文章的主要目的是记录下Dubbo当中NettyServer针对请求的处理过程,主要是理顺连接建立的过程和调用处理的过程。  这个解析的核心在于理解类的继承关系(可以理解为设计模式当中的装饰者模式),层层递进理解调用关系。

开篇

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

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


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过程

Receive过程

说明:

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


Dubbo-Netty的Connect过程

Connect过程
说明:

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

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
开发与运维
使用钉钉扫一扫加入圈子
+ 订阅

集结各类场景实战经验,助你开发运维畅行无忧

其他文章