Dubbo线程模型和调度策略

首先服务消费者通过代理对象 Proxy 发起远程调用，接着通过网络客户端 Client 将编码后的请求发送给服务提供方的网络层上，也就是 Server。Server 在收到请求后，首先要做的事情是对数据包进行解码。然后将解码后的请求发送至分发器 Dispatcher，再由分发器将请求派发到指定的线程池上，最后由线程池调用具体的服务。这就是一个远程调用请求的发送与接收过程。

那么在dubbo中请求是如何派发的？以及线程模型是什么样的那？

二、I/O线程和业务线程分离

• 如果事件处理的逻辑能迅速完成，并且不会发起新的 IO请求，比如只是在内存中记个标识，则直接在 IO线程上处理更快，因为减少了线程池调度。

• 但如果事件处理逻辑较慢，或者需要发起新的 IO 请求，比如需要查询数据库，则必须派发到线程池，否则 IO 线程阻塞，将导致不能接收其它请求。

• 如果用 IO 线程处理事件，又在事件处理过程中发起新的 IO 请求，比如在连接事件中发起登录请求，会报“可能引发死锁”异常，但不会真死锁。

所以在真实的业务场景中是需要将业务线程和I/O线程进行分离处理的。dubbo作为一个服务治理框架，底层的采用Netty作为网络通信的组件，在请求派发的时候支持不同的派发策略。

参考文章： www.cnblogs.com/my_life/art…

三、请求派发策略

连接建立

从官方描述来看，duboo支持五种派发策略，下面看下是如何实现的。以 Ntty4.x为 例：

1. NettyServer

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

2. ChannelHandlers#wrapInternal

   protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
        // 选择调度策略 默认是all
        return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                .getAdaptiveExtension().dispatch(handler, url))); 
    }
   复制代码

   在NettyServer的构造方法中通过 ChannelHandlers#wrap 方法设置 MultiMessageHandler ， HeartbeatHandler 并通过SPI扩展选择调度策略。
3. NettyServer#doOpen

protected void doOpen() throws Throwable {
      bootstrap = new ServerBootstrap();
      // 多线程模型
      // boss线程池，负责和消费者建立新的连接
      bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
      // worker线程池，负责连接的数据交换
      workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
              new DefaultThreadFactory("NettyServerWorker", true));

      final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
      channels = nettyServerHandler.getChannels();

      bootstrap.group(bossGroup, workerGroup)
              .channel(NioServerSocketChannel.class)
              .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE) // nagele 算法
              .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)// TIME_WAIT
              .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) //内存池
              .childHandler(new ChannelInitializer<NioSocketChannel>() {
                  @Override
                  protected void initChannel(NioSocketChannel ch) throws Exception {
                      NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                      ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                              .addLast("decoder", adapter.getDecoder()) //设置编解码器
                              .addLast("encoder", adapter.getEncoder())
                              .addLast("handler", nettyServerHandler);
                  }
              });
      // bind 端口
      ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
      channelFuture.syncUninterruptibly();
      channel = channelFuture.channel();

  }
复制代码

设置Netty的boss线程池数量为1，worker线程池(也就是I/O线程)为cpu核心数+1和向Netty中注测Handler用于消息的编解码和处理。

如果我们在一个JVM进程只暴露一个Dubbo服务端口，那么一个JVM进程只会有一个NettyServer实例，也会只有一个NettyHandler实例。并且设置了三个handler,用来处理编解码、连接的创建、消息读写等。在dubbo内部定义了一个 ChannelHandler 用来和Netty的 Channel 关联起来，通过上述的代码会发现 NettyServer 本身也是一个 ChannelHandler 。通过 NettyServer#doOpen 暴露服务端口后，客户端就能和服务端建立连接了，而提供者在初始化连接后会调用 NettyHandler#channelActive 方法来创建一个 NettyChannel

4. NettyChannel

public void channelActive(ChannelHandlerContext ctx) throws Exception {
      logger.debug("channelActive <" + NetUtils.toAddressString((InetSocketAddress) ctx.channel().remoteAddress()) + ">" + " channle <" + ctx.channel());
      //获取或者创建一个NettyChannel
      NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
      try {
          if (channel != null) {
              // <ip:port, channel>
              channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.channel().remoteAddress()), channel);
          }
          // 这里的 handler就是NettyServer
          handler.connected(channel);
      } finally {
          NettyChannel.removeChannelIfDisconnected(ctx.channel());
      }
  }
复制代码

与Netty和Dubbo都有自己的ChannelHandler一样，Netty和Dubbo也有着自己的Channel。该方法最后会调用 NettyServer#connected 方法来检查新添加channel后是否会超出提供者配置的accepts配置，如果超出，则直接打印错误日志并关闭该Channel，这样的话消费者端自然会收到连接中断的异常信息，详细可以见 AbstractServer#connected 方法。

5. AbstractServer#connected

public void connected(Channel ch) throws RemotingException {
     // If the server has entered the shutdown process, reject any new connection
     if (this.isClosing() || this.isClosed()) {
         logger.warn("Close new channel " + ch + ", cause: server is closing or has been closed. For example, receive a new connect request while in shutdown process.");
         ch.close();
         return;
     }

     Collection<Channel> channels = getChannels();
     //大于accepts的tcp连接直接关闭
     if (accepts > 0 && channels.size() > accepts) { 
         logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress() + " connections greater than max config " + accepts);
         ch.close();
         return;
     }
     super.connected(ch);
 }
复制代码

• 在dubbo中消费者和提供者默认只建立一个TCP长连接(详细代码请参考官网源码导读，服务引用一节)，为了增加消费者调用服务提供者的吞吐量，可以在消费方增加如下配置：

<dubbo:reference id="demoService" check="false" interface="org.apache.dubbo.demo.DemoService" connections="20"/>
复制代码

• 提供者可以使用accepts控制长连接的数量防止连接数量过多，配置如下：

<dubbo:protocol name="dubbo" port="20880" accepts="10"/>
复制代码

请求接收

当连接建立完成后，消费者就可以请求提供者的服务了，当请求到来，提供者这边会依次经过如下Handler的处理：

---> NettyServerHandler#channelRead ：接收请求消息。

---> AbstractPeer#received ：如果服务已经关闭，则返回，否则调用下一个Handler来处理。

---> MultiMessageHandler#received ：如果是批量请求，则依次对请求调用下一个Handler来处理。

---> HeartbeatHandler#received : 处理心跳消息。

---> AllChannelHandler#received ：该Dubbo的Handler非常重要，因为从这里是IO线程池和业务线程池的隔离。

---> DecodeHandler#received : 消息解码。

---> HeaderExchangeHandler#received ：消息处理。

---> DubboProtocol : 调用服务。

1. 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) {
        throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
       }
   }
复制代码

这里对execute进行了异常捕获，这是因为I/O线程池是无界的，但业务线程池可能是有界的，所以进行execute提交可能会遇到RejectedExecutionException异常 。

那么这里是如何获取到业务线程池的那？实际上 WrappedChannelHandler 是 xxxChannelHandlerd 的装饰类，根据dubbo spi可以知道，获取 AllChannelHandler 会首先实例化 WrappedChannelHandler 。

2. WrappedChannelHandler

public WrappedChannelHandler(ChannelHandler handler, URL url) {
        this.handler = handler;
        this.url = url;
        // 获取业务线程池
        executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);

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

复制代码

线程模型

1. FixedThreadPool

public class FixedThreadPool implements ThreadPool {

  @Override
  public Executor getExecutor(URL url) {
      // 线程池名称DubboServerHanler-server:port
      String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
      // 缺省线程数量200
      int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
      // 任务队列类型
      int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);

      return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
              queues == 0 ? new SynchronousQueue<Runnable>() :
                      (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                              : new LinkedBlockingQueue<Runnable>(queues)),
              new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
  }

}
复制代码

缺省情况下使用200个线程和 SynchronousQueue 这意味着如果如果线程池所有线程都在工作再有新任务会直接拒绝。

2. CachedThreadPool

public class CachedThreadPool implements ThreadPool {

    @Override
    public Executor getExecutor(URL url) {
        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
        // 核心线程数量 缺省为0
        int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
        // 最大线程数量 缺省为Integer.MAX_VALUE
        int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
        // queue 缺省为0
        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
        // 空闲线程存活时间
        int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
        return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS,
                queues == 0 ? new SynchronousQueue<Runnable>() :
                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }
}
复制代码

缓存线程池，可以看出如果提交任务的速度大于maxThreads将会不断创建线程，极端条件下将会耗尽CPU和内存资源。在突发大流量进入时不适合使用。

3. LimitedThreadPool

public class  LimitedThreadPool implements ThreadPool {

    @Override
    public Executor getExecutor(URL url) {
        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
        // 缺省核心线程数量为0
        int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
        // 缺省最大线程数量200
        int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
        // 任务队列缺省0
        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
        return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS,
                queues == 0 ? new SynchronousQueue<Runnable>() :
                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

}
复制代码

不配置的话和FixedThreadPool没有区别。

4. EagerThreadPool

public class EagerThreadPool implements ThreadPool {

   @Override
   public Executor getExecutor(URL url) {
       String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
       // 0
       int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
       // Integer.MAX_VALUE
       int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
       // 0
       int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
       // 60s
       int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);

       // init queue and executor
       // 初始任务队列为1
       TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues <= 0 ? 1 : queues);
       EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
               threads,
               alive,
               TimeUnit.MILLISECONDS,
               taskQueue,
               new NamedInternalThreadFactory(name, true),
               new AbortPolicyWithReport(name, url));
       taskQueue.setExecutor(executor);
       return executor;
   }
}
复制代码

EagerThreadPoolExecutor

public void execute(Runnable command) {
       if (command == null) {
           throw new NullPointerException();
       }
       // do not increment in method beforeExecute!
       //已提交任务数量
       submittedTaskCount.incrementAndGet();
       try {
           super.execute(command);
       } catch (RejectedExecutionException rx) { //大于最大线程数被拒绝任务 重新添加到任务队列
           // retry to offer the task into queue.
           final TaskQueue queue = (TaskQueue) super.getQueue();
           try {
               if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
                   submittedTaskCount.decrementAndGet();
                   throw new RejectedExecutionException("Queue capacity is full.", rx);
               }
           } catch (InterruptedException x) {
               submittedTaskCount.decrementAndGet();
               throw new RejectedExecutionException(x);
           }
       } catch (Throwable t) {
           // decrease any way
           submittedTaskCount.decrementAndGet();
           throw t;
       }
   }
复制代码

TaskQueue

public boolean offer(Runnable runnable) {
       if (executor == null) {
           throw new RejectedExecutionException("The task queue does not have executor!");
       }
       // 获取当前线程池中的线程数量
       int currentPoolThreadSize = executor.getPoolSize();
       // have free worker. put task into queue to let the worker deal with task.
       // 如果已经提交的任务数量小于当前线程池中线程数量(不是很理解这里的操作)
       if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
           return super.offer(runnable);
       }

       // return false to let executor create new worker.
       //当前线程数小于最大线程程数直接创建新worker
       if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
           return false;
       }

       // currentPoolThreadSize >= max
       return super.offer(runnable);
   }
复制代码

优先创建 Worker 线程池。在任务数量大于 corePoolSize 但是小于 maximumPoolSize 时，优先创建 Worker 来处理任务。当任务数量大于 maximumPoolSize 时，将任务放入阻塞队列中。阻塞队列充满时抛出 RejectedExecutionException 。(相比于cached:cached在任务数量超过 maximumPoolSize 时直接抛出异常而不是将任务放入阻塞队列)。

根据以上的代码分析，如果消费者的请求过快很有可能导致服务提供者业务线程池抛出 RejectedExecutionException 异常。这个异常是duboo的采用的线程拒绝策略 AbortPolicyWithReport#rejectedExecution 抛出的,并且会被反馈到消费端，此时简单的解决办法就是将提供者的服务调用线程池数目调大点，例如如下配置：

<dubbo:provider threads="500"/>
或
<dubbo:protocol name="dubbo" port="20882" accepts="10" threads="500"/>
复制代码

为了保证模块内的主要服务有线程可用（防止次要服务抢占过多服务调用线程），可以对次要服务进行并发限制，例如：

<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService" executes="100"/>
复制代码

dubbo的dispatcher 策略默认是all，实际上比较好的处理方式是I/O线程和业务线程分离，所以采取message是比较好得配置。并且如果采用all如果使用的dubo版本比较低很有可能会触发dubbo的bug。一旦业务线程池满了，将抛出执行拒绝异常，将进入caught方法来处理，而该方法使用的仍然是业务线程池，所以很有可能这时业务线程池还是满的，导致下游的一个HeaderExchangeHandler没机会调用，而异常处理后的应答消息正是 HeaderExchangeHandler#caught 来完成的，所以最后 NettyHandler#writeRequested 没有被调用，Consumer只能死等到超时，无法收到Provider的线程池打满异常(2.6.x已经修复该问题)。

• 推荐配置

<dubbo:protocol name="dubbo" port="8888" threads="500" dispatcher="message" />
复制代码

参考文章： manzhizhen.iteye.com/blog/239117…