线程池的应用

了解了Java中线程池的基本实现原理后，那么在一些优秀的Java框架中，开发者又是怎么应用线程池的呢？

以Tomcat、dubbo为例，来分析线程池在实际生产中的应用。

Tomcat中的线程池

Tomcat基于自身的业务场景，对JUC的线程池做了自己的扩展。其主要扩展有两个类：

• ThreadPoolExecute
• TaskQueue

ThreadPoolExecute

要扩展线程池，那么首先肯定要自定义一个ThreadPoolExecute 来继承java.util.concurrent.ThreadPoolExecutor

public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor 

然后重写execute 方法

@Override
public void execute(Runnable command) {execute(command,0,TimeUnit.MILLISECONDS);
}
public void execute(Runnable command, long timeout, TimeUnit unit) {//计数加一submittedCount.incrementAndGet();try {super.execute(command);} catch (RejectedExecutionException rx) {if (super.getQueue() instanceof TaskQueue) {final TaskQueue queue = (TaskQueue)super.getQueue();try {//任务被拒绝后，再次尝试往队列中添加if (!queue.force(command, timeout, unit)) {submittedCount.decrementAndGet();throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));}} catch (InterruptedException x) {submittedCount.decrementAndGet();throw new RejectedExecutionException(x);}} else {submittedCount.decrementAndGet();throw rx;}}
}

在tomcat线程池中，定义了一个计数变量submittedCount ，它代表提交到线程池中的任务总数，包括正在执行的线程数和队列中的任务数。

它的值会在执行execute 方法和自行实现的afterExecute 的方法时，实时更新。

此外，还何以增加一些额外的功能，例如stopCurrentThreadIfNeeded ，关掉当前线程以防止内存泄漏。

TaskQueue

TaskQueue是tomcat自行实现的一个阻塞队列，主要是重写了offer 方法。

@Override
public boolean offer(Runnable o) {if (parent==null) return super.offer(o);if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);//提交到线程池中任务数小于线程数，则说明存在空闲线程，那么就把任务放到队列中，等待线程来取if (parent.getSubmittedCount()<=(parent.getPoolSize())) return super.offer(o);//如果线程池中的线程数小于最大线程数，则添加任务到队列失败if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;return super.offer(o);
}

在JUC线程池中，execute 方法的第二步是当线程数大于核心数时，就会尝试往队列中添加，如果添加失败，那么就创建一个新的线程来执行任务。

那么在tomcat中，队列的做法是：如果线程数小于最大线程数，那么调用offer 方法时就会失败，也就说，此时就会执行execute 方法的第三步：创建一个新的线程。

总结

tomcat线程池相较于JUC原生的线程池，它是率先创建线程来执行任务，只有当线程数量达到最大线程数或者存在空闲线程时，才会将任务提交到队列中。

为什么？

tomcat为什么要做这个改变呢？

线程池的配置，通常是基于CPU密集型和IO密集型两种业务场景的考量。

JUC线程池这种尽量保持少量核心线程来执行任务的，显然更适合CPU密集型的场景。

而Tomcat作为一个web容器，需要处理大量的网络请求，所以一般情况下，它的场景是属于IO密集型的，这种场景下，线程执行时间短，增加线程数可以增强并发处理能力。

dubbo中的线程池

说到dubbo中线程池的应用，那么必须先了解dubbo中的线程模型。

dubbo线程模型

dubbo默认的底层网络通信是使用Netty的，Netty默认是基于Java Nio来实现和扩展的，所以其采用Selector网络通信模式，在Netty中的实现就是：服务方使用两级线程池，EventLoopGroup(boss)和EventLoopGroup(worker)，前者处理IO连接，后者处理IO读写事件。

那么在服务提供方的Netty IO线程该如何处理业务逻辑？

如果处理逻辑比较简单，并且不会发起新的IO请求，那么直接在当前线程上处理肯定会比较快；如果处理逻辑较复杂或者比较耗时，那么就应该使用使用业务线程来处理，否则就会导致IO线程阻塞，处理效率变低。

dubbo中根据请求的消息是直接在IO线程上处理还是被业务线程池处理，提供了五种线程模型：

• all(AllDispatcher) 所有的消息都派发到业务线程池
• direct(DirectDispatcher) 所有消息都不派发到业务线程池，全部都在IO线程上处理
• message(MessageOnlyDispatcher) 只有请求的响应消息派发到业务线程池
• execution(ExecutionDispatcher) 只有请求消息派发到业务线程池
• connection(ConnectionOrderedDispatcher) 在IO线程上将连接、断开连接事件依次放入队列，顺序执行，其它事件派发到业务线程

dubbo的设计中，大量使用SPI机制以供使用者可以自由配置各种场景下使用的策略，对于线程模型策略，也是如此，默认的策略便是all 。

@SPI(AllDispatcher.NAME)
public interface Dispatcher{}

线程池策略

dubbo中同样对于线程池，也是提供了多种实现。

@SPI("fixed")
public interface ThreadPool {}

服务提供方默认策略是固定线程数的线程池。

• fixed(FixedThreadPool) 一个固定线程数的线程池，默认线程数是200，同JUC中的newFixedThreadPool
• cached(CachedThreadPool) 一个自适应线程池 ，为每个请求创建一个线程，空闲线程1分钟后销毁，同JUC中的newCachedThreadPool
• limited(LimitedThreadPool) 线程数量不断增加，直到限制值（默认200），且空闲线程不会销毁
• eager(EagerThreadPool) 优先使用线程处理任务，直达线程数达最大，才放到队列中，同tomcat中的线程池

ThreadLessExecutor

Dubbo 发布里程碑版本，性能提升30%

Dubbo 2.7.5版本中对于消费端线程池模型做了优化，原因是：

“对 2.7.5 版本之前的 Dubbo 应用，尤其是一些消费端应用，当面临需要消费大量服务且并发数比较大的大流量场景时（典型如网关类场景），经常会出现消费端线程数分配过多的问题”。

这个改动就是引入了org.apache.dubbo.common.threadpool.ThreadlessExecutor 。

消费端的线程池策略默认是使用cached ，线程池会为每次的消费请求创建一个线程，那么当消费者应用并发较大或者提供者响应的时间较长时，就会出现消费者线程很多的情况。

消费者线程

2.7.5版本之前消费者模型

1. 消费者业务线程发起请求，拿到一个Future实例
2. 业务线程紧接着调用future.get() 阻塞等待业务结果返回
3. 服务提供方返回业务结果，交由独立的消费者线程池进行反序列化等处理，然后调用future.set() 回写结果
4. 消费者业务线程拿到到业务结果

2.7.5版本后的消费者线程模型

1. 消费者发起请求，拿到一个Future 实例
2. 在调用future.get() 之前，先调用ThreadlessExecutor.waitAndDrain() 方法，这会使得消费者业务线程在一个阻塞队列上等待着
3. 服务提供方返回业务结果，并将结果封装成一个Runnable Task 加入到ThreadlessExecutor队列中
4. 消费者业务线程从队列中获取Task并在本线程执行：反序列化并set到future中
5. 业务线程从future中拿到结果

相较于老版本，2.7.5版本的消费者业务线程自己负责监控和解析结果，从而免去了额外的消费端线程池开销，也就解决了之前会出现的消费端大量线程的问题。

//org.apache.dubbo.rpc.AsyncRpcResult#get()，即消费者调用future.get()的地方
@Override
public Result get() throws InterruptedException, ExecutionException {if (executor != null && executor instanceof ThreadlessExecutor) {ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;//调用future.get()之前，先调用wait方法，使业务线程在队列中等待threadlessExecutor.waitAndDrain();}return responseFuture.get();
}

Threadless

和普通的线程池最大的不同就是：ThreadlessExecutor不管理任何线程。

调用execute() 方法，提交到TreadlessExecutor的任务不会被调度到具体的线程。

@Override
public void execute(Runnable runnable) {runnable = new RunnableWrapper(runnable);synchronized (lock) {if (!waiting) {sharedExecutor.execute(runnable);} else {//服务提供方返回结果后，将任务加入到队列中，然后消费端业务线程从队列中取出结果，并执行queue.add(runnable);}}
}

在调用future.get()方法之前，消费者业务线程一直在一个队列中等待，直到服务提供方返回结果

public void waitAndDrain() throws InterruptedException {if (finished) {return;}//阻塞等待服务提供方调用excute(),将结果写入队列中Runnable runnable = queue.take();synchronized (lock) {waiting = false;runnable.run();}runnable = queue.poll();while (runnable != null) {runnable.run();runnable = queue.poll();}finished = true;
}

总结

因为2.7.5版本之前，dubbo的消费者端在一些情况下（调用并发较大且服务提供方返回结果的时间较长）会出现消费端线程数过多，从而发生oom，所以2.7.5版本后，dubbo对消费端的线程模型做了调整，通过本地线程复用的方式（即业务线程阻塞等待）来避免额外的线程消耗。