天天看点
返回

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

    本节将主要学习Dubbo是如何使用Netty来实现网络通讯的。

    从官网我们得知，Dubbo协议是使用单一长连接来进行网络传输，也就是说服务调用方持久与服务提供者建立一条连接，所有的服务调用调用信息通过。

    一条TCP连接进行传输，在网络层至少要考虑如下问题：

    1、服务端，客户端网络通讯模型（线程模型）

    2、传输（编码解码、序列化）。

    3、服务端转发策略等。

    Dubbo服务端的网络启动流程，在上篇中已给出序列图，本节还是以该图为切入点，引入本文的两个主人公：NettyServer、NettyClient。

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

    dubbo使用SPI机制，根据配置，可以支持如下框架实现网络通讯模型，例如netty3,netty4、mina、grizzly，本文重点分析基于Netty4的实现，包路径：dubbo-remoting-netty4。

    从上面的流程图，NettyTransport的职责就是调用new NettyServer的构造方法，从而构建NettyServer对象，在深入NettyServer对象构造过程之前，先来看一下NettyServer的类继承层次：

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

   NettyServer构造函数：

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

   代码@1：URL url：服务提供者URL；ChannelHandler网络事件处理器，

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

也就是当相应网络事件触发时，执行的事件处理器。

  • void connected(Channel channel) throws RemotingException

    连接事件，当收到客户端的连接事件时，执行该方法处理相关业务操作。

  • void disconnected(Channel channel) throws RemotingException：连接断开事件

  • void sent(Channel channel, Object message) throws RemotingException

    当可写事件触发时，服务端向客户端返回响应数据，就是通过该方法发送的。

  • void received(Channel channel, Object message) throws RemotingException

    当读事件触发时执行该方法，服务端在收到客户端的请求数据是，调用该方法执行解包等操作。

  • void caught(Channel channel, Throwable exception) throws RemotingException

    发生异常时，调用该方法。

   代码@2：调用ChannelHandlers.wrap对原生Handler进行包装，然后调用其父类的构造方法，首先，设置Dubbo服务端线程池中线程的名称，可以通过参数threadname来指定线程池中线程的前缀，默认为：DubboServerHandler + dubbo服务端IP与接口号。我比较好奇的是这里为什么需要对ChannelHandler进行包装呢？是增加了些什么逻辑呢？带着者问题，引出本节重点探讨的内容：事件派发机制。

    事件派发机制指的是网络事件（连接、读、写）等事件触发后，这些事件如何执行，是由IO线程还是派发到线程池中执行。Dubbo定义了如下5种事件派发机制：

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

   本文将详细分析各种事件的派发实现原理。

   ChannelHandlers#wrapInternal

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

   这里是典型的装饰模式，MultiMessageHandler，多消息处理Handler,HeartbeatHandler，心跳Handler,其主要功能是处理心跳返回与心跳请求，直接在IO线程中执行，每次收到信息，更新通道的读事件戳，每次发送数据时，记录通道的写事件戳。这里的核心关键是利用SPI自适配，返回合适的事件派发机制。Dispatcher的类层次结构如图所示：

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

   1、源码分析AllDispatcher实现原理

   线程派发机制：所有的消息都派发到线程池，包括请求、响应、连接事件、断开事件、心跳等。

public class AllDispatcher implements Dispatcher {
    public static final String NAME = "all";
    @Override
    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new AllChannelHandler(handler, url);
    }
}

   从中可以看出，事件派发类继承图分两个维度，Dispatcher（事件派发器）、与之对应的ChannelHandler,例如AllChannelHandler。

   1.1 WrappedChannelHandler

   接下来分析事件派发机制，重点关注ChannelHandler类的实现体系。

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

   纵观Dubbo ChannelHanler体系的设计，是经典的类装饰器模式，上述派发器主要解决的问题，是相关网络事件（连接、读（请求）、写（响应）、心跳请求、心跳响应）是在IO线程、还是在额外定义的线程池，故WrappedChannelHandler的主要职责是定义线程池相关的逻辑，具体是在IO线程上执行，还是在定义的线程池中执行，则由子类具体去定制，WrappedChannelHandler默认实现ChannelHandler的所有方法，各个方法的实现直接调用被装饰Handler的方法，见下图：

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

   接下来先重点关注一下WrappedChannelHandler的成员变量和构造方法的实现。

protected static final ExecutorService SHARED_EXECUTOR = Executors.newCachedThreadPool(new NamedThreadFactory("DubboSharedHandler", true));
protected final ExecutorService executor;
protected final ChannelHandler handler;
protected final URL url;

  • ExecutorService SHARED_EXECUTOR：共享线程池，默认线程池，如果

    ExecutorService executor为空，则使用SHARED_EXECUTOR

  • ExecutorService executor 定义的线程池
  • ChannelHandler handler：被装饰的ChannelHandler

  • URL url 服务提供者URL

       接下来关注一下其构造函数：

public WrappedChannelHandler(ChannelHandler handler, URL url) {
        this.handler = handler;
        this.url = url;
        executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);    // @1

        String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
        if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) {
            componentKey = Constants.CONSUMER_SIDE;
        }
        DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
        dataStore.put(componentKey, Integer.toString(url.getPort()), executor);  // @2
    }

   代码@1：构建线程池，这里基于SPI机制，用户可选择cached、eager、fixed、limited，将在本节下面详细介绍，这里只需要知道是构建了一个线程池。

   代码@2：将服务端都与线程池缓存起来，在服务端，线程池的缓存级别是 服务提供者协议（端口）：线程池。

   1.2 AllChannelHandler

事件派发机制：所有网络事件在线程池中执行，其实现机制肯定是重写ChannelHandler的所有网络事件方法，将调用其修饰的ChannelHanlder在线程池中执行。由于AllChannelHandler是第一个事件派发机制，故对其实现做一个详细描述。

   1.2.1 AllChannelHandler#connected

public void connected(Channel channel) throws RemotingException {
        ExecutorService cexecutor = getExecutorService();
        try {
            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t);
        }
    }

   连接事件，其主要实现是，首先先获取执行线程池，其获取逻辑是如果executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).

getAdaptiveExtension().getExecutor(url);获取不到线程池，则使用共享线程池。可以看出，连接事件的业务调用时异步执行，基于线程池。

注：调用时机，服务端收到客户端连接后，该方法会被调用。

   2.2.2 AllChannelHandler#disconnected

public void disconnected(Channel channel) throws RemotingException {
        ExecutorService cexecutor = getExecutorService();
        try {
            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("disconnect event", channel, getClass() + " error when process disconnected event .", t);
        }
    }

   其基本实现与connected相同，就是将具体的disconnected 事件所对应的业务扩展方法在线程池中执行。

   注：调用时机，服务端收到客户端断开连接后，该方法会被调用。

   2.2.3 AllChannelHandler#received

public void received(Channel channel, Object message) throws RemotingException {
        ExecutorService cexecutor = getExecutorService();
        try {
            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
        } catch (Throwable t) {
            //TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring
            //fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out
        	if(message instanceof Request && t instanceof RejectedExecutionException){
        		Request request = (Request)message;
        		if(request.isTwoWay()){
        			String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") threadpool is exhausted ,detail msg:" + t.getMessage();
        			Response response = new Response(request.getId(), request.getVersion());
        			response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
        			response.setErrorMessage(msg);
        			channel.send(response);
        			return;
        		}
        	}
            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
        }
    }

   调用时机：当服务端收到客户端发送的请求后，经过IO线程（Netty）会首先从二进制流中解码出一个个的请求，参数Object message，就是调用请求，然后在提交给线程池执行，执行完后，当业务处理完毕后，组装结果后，必然会在该线程中调用通道(Channel#write,flush)方法，向通道写入响应结果。

注：all事件派发机制，ChannelHandler#recive是在线程池中执行。

   2.2.4 AllChannelHandler#caught

public void caught(Channel channel, Throwable exception) throws RemotingException {
        ExecutorService cexecutor = getExecutorService();
        try {
            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception));
        } catch (Throwable t) {
            throw new ExecutionException("caught event", channel, getClass() + " error when process caught event .", t);
        }
    }

   当发生异常时，ChannelHandler#caught也在线程池中执行。

   令人颇感意外的是，AllChannelHandler并未重写WrappedChannelHandler的sent方法，也就是说ChannelHandler#sent事件回调方法，是在IO线程中执行。

WrappedChannelHandler#sent

public void sent(Channel channel, Object message) throws RemotingException {
        handler.sent(channel, message);
}

   这个和官方文档还是有一定出入的。

源码分析Dubbo网络通讯篇之NettyServer网络事件派发机制(Dispatch)

   1.3 ExecutionChannelHandler

   对应事件派件器：ExecutionDispatcher，其配置值：execution,从其源码的实现来看，与AllDispatcher实现基本类似，唯一的区别是，如果executor线程池为空时，并不会使用共享线程池，目前我还想不出什么情况下，线程池会初始化失败。

   1.4 DirectDispatcher

直接派发，也就是所有的事件全部在IO线程中执行，故其实现非常简单：

public class DirectDispatcher implements Dispatcher {
    public static final String NAME = "direct";
    @Override
    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return handler;
    }
}

   1.5 MessageOnlyDispatcher、MessageOnlyChannelHandler

   事件派发器：只有请求事件在线程池中执行，其他响应事件、心跳，连接，断开连接等事件在IO线程上执行，故其只需要重写recive方法即可：

@Override
    public void received(Channel channel, Object message) throws RemotingException {
        ExecutorService cexecutor = executor;
        if (cexecutor == null || cexecutor.isShutdown()) {
            cexecutor = SHARED_EXECUTOR;
        }
        try {
            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
        } catch (Throwable t) {
            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
        }
    }

   1.6 ConnectionOrderedDispatcher ConnectionOrderedChannelHandler

   事件派发器：连接、断开连接事件排队执行，并可通过connect.queue.capacity属性设置队列长度，请求事件、异常事件在线程池中执行。

   1.6.1 构造方法

public ConnectionOrderedChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
        String threadName = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
        connectionExecutor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(url.getPositiveParameter(Constants.CONNECT_QUEUE_CAPACITY, Integer.MAX_VALUE)),
                new NamedThreadFactory(threadName, true),
                new AbortPolicyWithReport(threadName, url)
        );  // FIXME There's no place to release connectionExecutor!
        queuewarninglimit = url.getParameter(Constants.CONNECT_QUEUE_WARNING_SIZE, Constants.DEFAULT_CONNECT_QUEUE_WARNING_SIZE);
    }

   重点关注一下connectionExecutor ，用来执行连接、断开事件的线程池，线程池中只有一个线程，并且队列可以选择时有界队列，通过connect.queue.capacity属性配置，超过的事件，则拒绝执行。

   1.6.2 ConnectionOrderedChannelHandler#connected

public void connected(Channel channel) throws RemotingException {
        try {
            checkQueueLength();
            connectionExecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t);
        }
    }

   检查队列长度，如果超过警告值，则输出警告信息，然后提交连接线程池中执行，disconnected事件类似。其他received、caught事件，则与AllDispatcher相同，就不在重复。

   总结：本文主要是分析阐述了Dubbo Dispatch机制，但与官方文档存在出入，先归纳如下：

Dispatch

   所有的sent事件方法、心跳请求全部在IO线程上执行。

   1、all : 除sent事件回调方法、心跳外，全部在线程池上执行。

   2、execution : 与all类似，唯一区就是all在线程池未指定时，可以使用共享线程池，这个差别等同于没有。

   3、 message : 只有请求事件在线程池中执行，其他在IO线程上执行。

   4、connection : 请求事件在线程池中执行，连接、断开连接事件排队执行（含一个线程的线程池）

   5、direct : 所有事件都在IO线程中执行。

欢迎加笔者微信号(dingwpmz)，加群探讨，笔者优质专栏目录：

1、源码分析RocketMQ专栏(40篇+)

2、源码分析Sentinel专栏(12篇+)

3、源码分析Dubbo专栏(28篇+)

4、源码分析Mybatis专栏

5、源码分析Netty专栏(18篇+)

6、源码分析JUC专栏

7、源码分析Elasticjob专栏

8、Elasticsearch专栏（20篇+）

9、源码分析MyCat专栏

Dubbo 源码分析Dubbo专栏 dispatch 事件派发机制 all、message、connection direct、execution
上一篇: Bash处理字符串系列函数（二）按分隔符拆分字符串将字符串改为小写将字符串改为大写
下一篇: python--从入门到实践--chapter 11 代码测试unittest

继续阅读