Dubbo源码 之 事件派发线程模型

​一、背景

Dubbo 将底层通信框架中接收请求的线程称为 IO 线程。如果一些事件处理逻辑可以很快执行完，比如只在内存打一个标记，此时直接在 IO 线程上执行该段逻辑即可。但如果事件的处理逻辑比较耗时，比如该段逻辑会发起数据库查询或者 HTTP 请求。此时我们就不应该让事件处理逻辑在 IO 线程上执行，而是应该派发到线程池中去执行。原因也很简单，IO 线程主要用于接收请求，如果 IO 线程被占满，将导致它不能接收新的请求。

二、源码分析

首先通过 一张Dubbo一次请求大致流程图，看一下线程派发器所处的位置。

如上图，红框中的 Dispatcher 就是线程派发器。需要说明的是，Dispatcher 真实的职责创建具有线程派发能力的 ChannelHandler，比如 AllChannelHandler、MessageOnlyChannelHandler 和 ExecutionChannelHandler 等，其本身并不具备线程派发能力。Dubbo 支持 5 种不同的线程派发策略，下面通过一个表格列举一下。

默认配置下，Dubbo 使用 all 派发策略，即将所有的消息都派发到线程池中。下面我们来分析一下 AllChannelHandler 的代码。

/**
* @Author: wenyixicodedog
* @Date:  2020-07-09
* @Param:
* @return:
* @Description:  所有消息都派发到线程池，包括请求，响应，连接事件，断开事件等
*/
public class AllChannelHandler extends WrappedChannelHandler {
​
    public AllChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
    }
​
    /** 处理连接事件 */
    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);
        }
    }
​
    /** 处理断开事件 */
    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);
        }
    }
​
    /** 处理请求和响应消息，这里的 message 变量类型可能是 Request，也可能是 Response */
    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) {
            // 如果通信方式为双向通信，此时将 Server side ... threadpool is exhausted
            // 错误信息封装到 Response 中，并返回给服务消费方。
          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);
                    // 返回包含错误信息的 Response 对象
              channel.send(response);
              return;
            }
          }
            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
        }
    }
​
    //服务端获取线程池
    private ExecutorService getExecutorService() {
        ExecutorService cexecutor = executor;
        if (cexecutor == null || cexecutor.isShutdown()) {
            cexecutor = SHARED_EXECUTOR;
        }
        return cexecutor;
    }
}

如上，connected主要负责请求、响应的发起和接收，received方法主要负责请求或者相应的处理业务。先大致描述下有关AllChannelHandler在这一部分的一次请求流程：client发起请求connected->server接收请求connected->nettyHandler.receive方法接收请求->AllChannelHandler#receive处理请求->server发起响应请求connected->client接收请求connected。

ChannelState.CONNECTED、ChannelState.RECEIVED都是利用枚举声明的请求状态。

另外，AllChannelHandler#getExecutorService用来获取线程池，这个首先通过SPI技术获取线程池，如果为空则赋值一个shared_executor,这部分稍等再细看下。->⭐️⭐️

回到AllChannelHandler，获取连接之后将连接事件交给线程池处理，请求对象会被封装 ChannelEventRunnable 中，ChannelEventRunnable 将会是服务调用过程的新起点。所以接下来我们以 ChannelEventRunnable 为起点向下探索。

该类的主要代码如下：

/**
* @Author: wenyixicodedog
* @Date:  2020-07-09
* @Param:
* @return:
* @Description:  开启线程执行业务逻辑
* 请求对象会被封装 ChannelEventRunnable 中，ChannelEventRunnable 将会是服务端调用过程的新起点。
*/
public class ChannelEventRunnable implements Runnable {
​
    private static final Logger logger = LoggerFactory.getLogger(ChannelEventRunnable.class);
​
    private final ChannelHandler handler;
    private final Channel channel;
    private final ChannelState state;
    private final Throwable exception;
    private final Object message;
​
    。。。
​
    public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Object message, Throwable exception) {
        this.channel = channel;
        this.handler = handler;
        this.state = state;
        this.message = message;
        this.exception = exception;
    }
​
    /**
    * @Author: wenyixicodedog
    * @Date:  2020-07-09
    * @Param:
    * @return:
    * @Description:  ChannelEventRunnable 仅是一个中转站，它的 run 方法中并不包含具体的调用逻辑，
     * 仅用于将参数传给其他 ChannelHandler 对象进行处理，该对象类型为 DecodeHandler。
    */
    public void run() {
        // 检测通道状态，对于请求或响应消息，此时 state = RECEIVED
        switch (state) {
            case CONNECTED:
                try {
                    handler.connected(channel);
                } catch (Exception e) {
                    logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
                }
                break;
            case DISCONNECTED:
                try {
                    handler.disconnected(channel);
                } catch (Exception e) {
                    logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
                }
                break;
            case SENT:
                try {
                    handler.sent(channel, message);
                } catch (Exception e) {
                    logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                            + ", message is " + message, e);
                }
                break;
            //请求和响应消息出现频率明显比其他类型消息高，所以新版dubbo对RECEIVED单独拿出来作为一个处理方式放在switch前面。
            case RECEIVED:
                try {
                    handler.received(channel, message);
                } catch (Exception e) {
                    logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                            + ", message is " + message, e);
                }
                break;
            case CAUGHT:
                try {
                    handler.caught(channel, exception);
                } catch (Exception e) {
                    logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                            + ", message is: " + message + ", exception is " + exception, e);
                }
                break;
            default:
                logger.warn("unknown state: " + state + ", message is " + message);
        }
    }
​
}

因为ChannelEventRunnable实现了 Runnable 接口，所以他的核心逻辑就是run方法执行的内容。请求和响应消息出现频率明显比其他类型消息高，所以新版本的dubbo这里对该类型的消息进行了针对性判断，我的dubbo是2.5.x，所以暂时还是放在了一起进行判断。ChannelEventRunnable 其实仅仅是一个中转站，它的 run 方法中并不包含具体的调用逻辑，仅用于将参数传给其他 ChannelHandler 对象进行处理，该对象类型为 DecodeHandler。这里先看一下他们各个类之间的关系。

WrapperChannelHandler负责的是事件派发，AbstractChannelHandlerDelegate的子类负责解码、心跳相关操作，HeaderExchangeHandler的子类负责后续请求的处理。

我们进入到DecodeHandler

public class DecodeHandler extends AbstractChannelHandlerDelegate {
​
    private static final Logger log = LoggerFactory.getLogger(DecodeHandler.class);
​
    public DecodeHandler(ChannelHandler handler) {
        super(handler);
    }
​
    public void received(Channel channel, Object message) throws RemotingException {
        if (message instanceof Decodeable) {
            // 对 Decodeable 接口实现类对象进行解码
            decode(message);
        }
​
        if (message instanceof Request) {
            // 对 Request 的 data 字段进行解码
            decode(((Request) message).getData());
        }
​
        if (message instanceof Response) {
            // 对 Response 的 result 字段进行解码
            decode(((Response) message).getResult());
        }
        // 执行后续逻辑
        handler.received(channel, message);
    }
​
    private void decode(Object message) {
        // Decodeable 接口目前有两个实现类，
        // 分别为 DecodeableRpcInvocation 和 DecodeableRpcResult
        if (message != null && message instanceof Decodeable) {
            try {
                // 执行解码逻辑
                ((Decodeable) message).decode();
                if (log.isDebugEnabled()) {
                    log.debug(new StringBuilder(32).append("Decode decodeable message ")
                            .append(message.getClass().getName()).toString());
                }
            } catch (Throwable e) {
                if (log.isWarnEnabled()) {
                    log.warn(
                            new StringBuilder(32)
                                    .append("Call Decodeable.decode failed: ")
                                    .append(e.getMessage()).toString(),
                            e);
                }
            } 
        } 
    } 
​
}

DecodeHandler 主要是包含了一些解码逻辑。请求解码可在 IO 线程上执行，也可在线程池中执行，这个取决于运行时配置。DecodeHandler 存在的意义就是保证请求或响应对象可在线程池中被解码。解码完毕后，完全解码后的 Request 对象会继续向后传递，下一站是 HeaderExchangeHandler。

/**
 * 交换接收器
 */
public class HeaderExchangeHandler implements ChannelHandlerDelegate {
​
    protected static final Logger logger = LoggerFactory.getLogger(HeaderExchangeHandler.class);
​
    public static String KEY_READ_TIMESTAMP = HeartbeatHandler.KEY_READ_TIMESTAMP;
​
    public static String KEY_WRITE_TIMESTAMP = HeartbeatHandler.KEY_WRITE_TIMESTAMP;
​
    private final ExchangeHandler handler;
​
    public HeaderExchangeHandler(ExchangeHandler handler) {
        if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        }
        this.handler = handler;
    }
​
  ......
  
    Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
        Response res = new Response(req.getId(), req.getVersion());
          // 检测请求是否合法，不合法则返回状态码为 BAD_REQUEST 的响应
          if (req.isBroken()) {
            Object data = req.getData();
​
            String msg;
              if (data == null) msg = null;
              else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
              else msg = data.toString();
            res.setErrorMessage("Fail to decode request due to: " + msg);
              // 设置 BAD_REQUEST 状态
            res.setStatus(Response.BAD_REQUEST);
​
              return res;
        }
          // 获取 data 字段值，也就是 RpcInvocation 对象
        Object msg = req.getData();
          try {
              // 继续向下调用
            Object result = handler.reply(channel, msg);
              // 设置 OK 状态码
            res.setStatus(Response.OK);
              // 设置调用结果
            res.setResult(result);
        } catch (Throwable e) {
              // 若调用过程出现异常，则设置 SERVICE_ERROR，表示服务端异常
            res.setStatus(Response.SERVICE_ERROR);
            res.setErrorMessage(StringUtils.toString(e));
        }
          return res;
    }
​
      public void received(Channel channel, Object message) throws RemotingException {
        channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
        ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
        try {  
              // 处理请求对象
              if (message instanceof Request) {
                  // handle request.
                Request request = (Request) message;
                  // 处理事件
                  if (request.isEvent()) {
                    handlerEvent(channel, request);
                } else {
                      // TODO  双向通信
                      if (request.isTwoWay()) {
                          // TODO  向后调用服务，并得到调用结果
                        Response response = handleRequest(exchangeChannel, request);
                          // TODO  将调用结果返回给服务消费端
                        channel.send(response);
                    } else {
                          // TODO  如果是单向通信，仅向后调用指定服务即可，无需返回调用结果
                        handler.received(exchangeChannel, request.getData());
                    }
                }
              // 处理响应对象，服务消费方会执行此处逻辑，后面分析
            } else if (message instanceof Response) {
                handleResponse(channel, (Response) message);
              // telnet 相关，忽略
            } else if (message instanceof String) {
                  if (isClientSide(channel)) {
                    Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
                    logger.error(e.getMessage(), e);
                } else {
                    String echo = handler.telnet(channel, (String) message);
                      if (echo != null && echo.length() > 0) {
                        channel.send(echo);
                    }
                }
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }
    ......
}

到这里，我们看到了比较清晰的请求和响应逻辑。对于双向通信，HeaderExchangeHandler 首先向后进行调用，得到调用结果。然后将调用结果封装到 Response 对象中，最后再将该对象返回给服务消费方。如果请求不合法，或者调用失败，则将错误信息封装到 Response 对象中，并返回给服务消费方。否则，获取 data 字段值，也就是 RpcInvocation 对象，继续进行向下调用。

三、服务端接口调用

接下来我们继续向后分析，看一下我们的手写的实现类最终调用的流程是什么样的。从HeaderExchangeHandler#handleRequest#reply继续往下走。

来到定义在 DubboProtocol 类中的匿名类对象逻辑，如下：

 //创建匿名实现类实现reply方法
    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
​
        // TODO 从HeaderExchangeHandler#handleRequest#Object result = handler.reply(channel, msg);调用进来
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                //根据客户端请求信息封装key值从exporterMap取出接口实现类对象invoker 获取 Invoker 实例
                Invoker<?> invoker = getInvoker(channel, inv);
                // 如果是回调，则需要考虑向后兼容性
                if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                    String methodsStr = invoker.getUrl().getParameters().get("methods");
                    boolean hasMethod = false;
                    if (methodsStr == null || methodsStr.indexOf(",") == -1) {
                        hasMethod = inv.getMethodName().equals(methodsStr);
                    } else {
                        //循环验证客户端请求的方法，服务端是否存在对应的实现
                        String[] methods = methodsStr.split(",");
                        for (String method : methods) {
                            if (inv.getMethodName().equals(method)) {
                                hasMethod = true;
                                break;
                            }
                        }
                    }
                    //不存在直接抛出异常
                    if (!hasMethod) {
                        logger.warn(new IllegalStateException("The methodName " + inv.getMethodName() + " not found in callback service interface ,invoke will be ignored. please update the api interface. url is:" + invoker.getUrl()) + " ,invocation is :" + inv);
                        return null;
                    }
                }
                RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                //利用反射调用实现类对象的方法响应请求
                return invoker.invoke(inv);
            }
            throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
        }
​
        /**
        * @Author: wenyixicodedog
        * @Date:  2020-07-03
        * @Param:  [channel, message]
        * @return:  void
        * @Description:  接受客户端请求并返回响应消息
        */
        @Override
        public void received(Channel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                //服务端处理并返回响应消息
                reply((ExchangeChannel) channel, message);
            } else {
                super.received(channel, message);
            }
        }
​
        @Override
        public void connected(Channel channel) throws RemotingException {
            invoke(channel, Constants.ON_CONNECT_KEY);
        }
        ......
    };

      首先根据客户端请求信息封装key值从exporterMap取出接口实现类对象invoker 获取 Invoker 实例，并通过 Invoker 的 invoke 方法调用服务逻辑。invoke 方法定义在 AbstractProxyInvoker 中，代码如下。

首先看下getInvoker逻辑

核心逻辑就是计算service key，然后从exporterMap缓存中查找与serviceKey相对应的 DubboExporter 对象，exporterMap是exporter的缓存，这个在服务暴露通过export方法将wrapperInvoker暴露的时候会获取到exporter，然后将 <serviceKey, DubboExporter> 映射关系存储到 exporterMap 集合中。

        当然，DubboProtocol#reply方法执行完了还要进行一些过滤器链的调用，比如监视器monitor、trace等，就不细看啦。可以自己debug分析。

对于invoker.invoke调用，invoke 方法定义在 AbstractProxyInvoker 中，代码如下。

public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
​
    ......
​
    @Override
    public Result invoke(Invocation invocation) throws RpcException {
        try {
            // 调用 doInvoke 执行后续的调用，并将调用结果封装到 RpcResult 中，并
            return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
        } catch (InvocationTargetException e) {
            return new RpcResult(e.getTargetException());
        } catch (Throwable e) {
            throw new RpcException("Failed to invoke remote proxy method ...");
        }
    }
​
    protected abstract Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable;
}

如上，invoke方法进行doInvoke的调用，然后将结果封装在了RPCResult对象之中。doInvoke 是一个抽象方法，这个需要由具体的 Invoker 实例实现。

这个Invoker 实例是在运行时通过 JavassistProxyFactory 中以内部类的方式创建的，创建逻辑如下：

Wrapper 是一个抽象类，其中 invokeMethod 是一个抽象方法。Dubbo 会在运行时通过 Javassist 框架为 Wrapper 生成实现类，并实现 invokeMethod 方法，该方法最终会根据调用信息调用具体的服务。以 DemoServiceImpl 为例，Javassist 为其生成的代理类如下。

public class Wrapper0 extends Wrapper implements ClassGenerator.DC {
    public static String[] pns;
    public static Map pts;
    public static String[] mns;
    public static String[] dmns;
    public static Class[] mts0;
​
    // 省略其他方法
​
    public Object invokeMethod(Object object, String string, Class[] arrclass, Object[] arrobject) throws InvocationTargetException {
        DemoService demoService;
        try {
            // 类型转换
            demoService = (DemoService)object;
        }
        catch (Throwable throwable) {
            throw new IllegalArgumentException(throwable);
        }
        try {
            // 根据方法名调用指定的方法
            if ("sayHello".equals(string) && arrclass.length == 1) {
                return demoService.sayHello((String)arrobject[0]);
            }
        }
        catch (Throwable throwable) {
            throw new InvocationTargetException(throwable);
        }
        throw new NoSuchMethodException(new StringBuffer().append("Not found method "").append(string).append("" in class com.alibaba.dubbo.demo.DemoService.").toString());
    }
}

到这里，整个服务调用过程就分析完了。

四、AllChannelHandler#getExecutorService

还记得之前说过AllChannelHandler#getExecutorService用来服务端获取线程池，先看下大致逻辑结构。

private ExecutorService getExecutorService() {
        ExecutorService cexecutor = executor;
        if (cexecutor == null || cexecutor.isShutdown()) {
            cexecutor = SHARED_EXECUTOR;
        }
        return cexecutor;
    }

public class WrappedChannelHandler implements ChannelHandlerDelegate {
​
    protected static final Logger logger = LoggerFactory.getLogger(WrappedChannelHandler.class);
​
    protected static final ExecutorService SHARED_EXECUTOR = Executors.newCachedThreadPool(new NamedThreadFactory("DubboSharedHandler", true));
​
    protected final ExecutorService executor;
​
    protected final ChannelHandler handler;
​
    protected final URL url;
​
    public WrappedChannelHandler(ChannelHandler handler, URL url) {
        this.handler = handler;
        this.url = url;
        executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
​
        ......        
    }
​
    ......
​
}

默认的executor是WrappedChannelHandler构造中dubbo通过SPI从url中获取线程池参数然后动态加载进来的，而shared_executor则是作为成员变量通过SPI加载。我的目的就是想搞明白服务端处理消费端请求使用的线程池模型。

项目启动进行访问，日志显示线程名称为DubboServerHandler

首先在AllChannelHandler#getExecutorService打断点不断进行debug

找到threadFactory#mPrefix为DubboServerHandler

这个时候executor发现被赋值，说明是在之前已经初始化完成。

然后在WrappedChannelHandler#WrappedChannelHandler#executor断点

程序执行到@WrapperChannelHandler构造方法，然后getAdaptiveExtension进入动态生成的线程池类ThreadPool$Adaptive

这里我生成了一个ThreadPool运行时的动态类。

/**
 * @author: wenyixicodedog
 * @create: 2020-07-09
 * @description:
 */
public class ThreadPool$Adaptive implements com.alibaba.dubbo.common.threadpool.ThreadPool {
​
    public java.util.concurrent.Executor getExecutor(com.alibaba.dubbo.common.URL arg0) {
        if (arg0 == null) throw new IllegalArgumentException("url == null");
        com.alibaba.dubbo.common.URL url = arg0;
        String extName = url.getParameter("threadpool", "fixed");
        if (extName == null)
            throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.common.threadpool.ThreadPool) name from url" +
                    "(" + url.toString() + ") use keys([threadpool])");
        com.alibaba.dubbo.common.threadpool.ThreadPool extension =
                ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.common.threadpool.ThreadPool.class).getExtension(extName);
        return extension.getExecutor(arg0);
    }
}

因为根据url获取到的threadpool为空所以用的是默认配置的fixed

然后getExecutor就是进入到fixedThreadPool的执行逻辑了

NamedThreadFactory该线程工厂为我们的线程池创建了名称，为线程创建了名称，方便我们查看堆栈信息时进行 debug ，线程为守护/用户的决定权由使用者决定。

队列容量queues默认为0，该线程池的workQueue为SynchronousQueue，queues如果小于0，使用无界阻塞队列LinkedBlockingQueue，queues如果大于0，使用以queues为capacity的有界阻塞队列LinkedBlockingQueue。

queues默认为0，workQueue为SynchronousQueue，但是阻塞队列虽然默认是 SynchronousQueue，如果用户配置了 queues 变量，且其值较大，使用的阻塞队列就是 LinkedBlockingQueue，此时一旦 core 配置较小，就会导致事件阻塞。

我的dubbo版本为2.5.x，在Dubbo 的最新源码中又提供了一个新的线程池

EagerThreadPool，当核心线程池核心线程都在忙碌的时候，新来的任务不是放在workQueue当中，而是新开启线程进行执行任务。dubbo的线程池好像在 java.util.concurrent.Executors 都为我们提供了。只不过 Dubbo 又自己实现了一下，设定了下自定义参数。

dubbo线程池选择可以自定义配置

<dubbo:protocol name="dubbo" dispatcher="all" threadpool="fixed" threads="200"/>

dubbo的优化很多可以针对线程池入手，有关dubbo线程池详细内容后续在进行详细分析吧。????

这里再来回顾下Executor框架中FixedThreadPool和CachedThreadPool特性作为结束。【Java并发编程的艺术】

①、FixedThreadPool

FixedThreadPool：称为可重用固定线程数的线程池。

FixedThreadPool的corePoolSize和maximumPoolSize都被设置为创建FixedThreadPool时指定的参数nThreads。

当线程池中的线程数大于corePoolSize时，keepAliveTime为多余的空闲线程等待新任务的最长时间，超过这个时间后多余的线程将被终止。如果把keepAliveTime设置为0L，意味着多余的空闲线程会被立即终止。

FixedThreadPool使用无界阻塞队列LinkedBlockingQueue作为线程池的工作队列（队列的容量为Integer.MAX_VALUE）。使用无界阻塞队列作为工作队列会对线程池带来如下影响。

1）当线程池中的线程数达到corePoolSize后，新任务将在队列中等待，因此线程池中的线程数不会超过corePoolSize。

2）由于1，使用无界阻塞队列时maximumPoolSize将是一个无效参数。

3）由于1和2，使用无界阻塞队列时keepAliveTime将是一个无效参数。

4）由于使用无界阻塞队列，运行中的FixedThreadPool（未执行方法shutdown()或shutdownNow()）不会拒绝任务（不会调用RejectedExecutionHandler.rejectedExecution方法）。

②、CachedThreadPool

CachedThreadPool：是一个会根据需要创建新线程的线程池。

CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列，但CachedThreadPool的maximumPool是无界的。这意味着，如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时，CachedThreadPool会不断创建新线程。极端情况下，CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。

个人才疏学浅、信手涂鸦，dubbo框架更多模块解读相关源码持续更新中，感兴趣的朋友请移步至个人公众号，谢谢支持????????......

公众号：wenyixicodedog

最后

以上就是无奈金毛为你收集整理的Dubbo源码 之 事件派发线程模型的全部内容，希望文章能够帮你解决Dubbo源码 之 事件派发线程模型所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。