面对海量网络请求，Tomcat线程池如何进行扩展？

news2024/9/23 2:36:59

面对海量网络请求，Tomcat线程池如何进行扩展？

上篇文章:深入浅出Tomcat网络通信的高并发处理机制说到Tomcat中EndPoint如何高效处理网络通信，其中离不开Tomcat线程池的大力支持

本篇文章就来聊聊Tomcat中的线程池与JUC下的线程池到底有何不同？

java.util.concurrent.ThreadPoolExecutor 是JUC下提供的线程池

而 org.apache.tomcat.util.threads.ThreadPoolExecutor Tomcat中的线程池对其进行扩展

先回顾下JUC线程池执行流程：

如果工作线程数量小于核心线程数量，创建核心线程执行任务
如果工作线程数量大于等于核心线程数量并且线程池还在运行则尝试将任务加入阻塞队列
如果任务加入阻塞队列失败（说明阻塞队列已满），并且工作线程小于最大线程数，则创建非核心线程执行任务
如果阻塞队列已满、并且工作线程数量达到最大线程数量则执行拒绝策略

不理解JUC下线程池的同学可以查看：12分钟从Executor自顶向下彻底搞懂线程池

在这个过程中，我们可以发现JUC的线程池更偏向于CPU密集型任务

当任务数量超过核心线程时，会把任务放到队列中排队（防止线程过多，上下文开销过大），只有队列已满才会创建非核心线程一起来执行任务

（JUC线程池也是可以通过调整参数满足IO密集型任务的，比如把核心线程数量调整为CPU核心数量的两倍）

在面对IO密集型任务时，JUC线程池还有能够优化、提升吞吐量的地方，Tomcat正在这些地方进行扩展：

提前创建核心线程

JUC线程池创建时并不会提前创建好所有的核心线程，而是使用“懒加载”，任务到达时不够核心线程数再创建

Tomcat可能在刚启动就收到大量网络请求，因此创建线程池时不能再像JUC中的线程池使用“懒加载”的方式，而是在创建线程池时就提前创建核心线程

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    //...
    
	//启动所有核心线程
    prestartAllCoreThreads();
}

创建非核心线程的时机

Tomcat线程池使用自定义的阻塞队列TaskQueue，其继承LinkedBlockingQueue，默认情况下是无界队列（integer最大值）

public class TaskQueue extends LinkedBlockingQueue<Runnable>

在JUC线程池执行流程中，必须等到队列已满才会去创建非核心线程

如果队列无界，那么可能队列堆积太多任务，直到发生OOM也不会创建非核心线程

对于面对IO密集型任务的Tomcat来说，这肯定是不能满足需求的

于是，Tomcat重写TaskQueue队列入队的逻辑，改变创建非核心线程的时机

//parent是该队列对应的线程池，parent就是用来判断什么时候创建非核心线程的
private transient volatile ThreadPoolExecutor parent = null;
//获取当前工作线程数量
parent.getPoolSizeNoLock()
//获取提交的任务数量    
parent.getSubmittedCount()

//调用offer 说明工作线程数量大于等于核心线程数
public boolean offer(Runnable o) {
    //没有parent直接入队
    if (parent==null) {
        return super.offer(o);
    }
    
    //如果线程池的工作线程数量等于最大线程数量则直接入队
    if (parent.getPoolSizeNoLock() == parent.getMaximumPoolSize()) {
        return super.offer(o);
    }
    
    //来到这说明当前工作线程小于最大线程数量，表面可以创建非核心线程
    
    //如果提交的任务数小于等于工作线程数量，说明有的线程空闲 放入队列
    if (parent.getSubmittedCount() <= parent.getPoolSizeNoLock()) {
        return super.offer(o);
    }
    
    //到此说明提交任务数大于工作线程数量
    //如果还没到最大线程数量，则返回false 后续创建非核心线程
    if (parent.getPoolSizeNoLock() < parent.getMaximumPoolSize()) {
        return false;
    }
    
    //默认入队
    return super.offer(o);
}

ThreadPoolExecutor.executeInternal

	private void executeInternal(Runnable command) {
        if (command == null) {
            throw new NullPointerException();
        }
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true)) {
                return;
            }
            c = ctl.get();
        }
    	//工作线程数量大于核心线程数量 调用offer
    	//工作线程数量小于最大线程数量 并且 提交任务数量大于工作线程数量 offer返回false
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command)) {
                reject(command);
            } else if (workerCountOf(recheck) == 0) {
                addWorker(null, false);
            }
        }
    	//调用addWorker创建非核心线程
        else if (!addWorker(command, false)) {
            reject(command);
        }
    }

从源码中可以看到：当线程池的工作线程数量大于核心线程数量、小于最大线程数量并且提交任务数量大于工作线程数量时会去创建非核心线程

也就是说，只要线程数量不超过最大线程，并且任务数量超过当前线程数量，就会去创建非核心线程

这样任务数量过多就去创建非核心线程执行更适合IO密集型的任务

拒绝后再次尝试放入队列

在JUC线程池中，当队列已满并且线程数量达到最大线程数量时会执行拒绝策略

Tomcat线程池捕获拒绝策略的异常后再次尝试把任务放入队列，进而提升吞吐量（拒绝这段时间也可能消费任务，进一步利用队列的容量）

public void execute(Runnable command) {
    //原子类自增记录提交任务数量
    submittedCount.incrementAndGet();
    try {
        //线程核心流程
        executeInternal(command);
    } catch (RejectedExecutionException rx) {
        //捕获拒绝异常
        if (getQueue() instanceof TaskQueue) {
            final TaskQueue queue = (TaskQueue) getQueue();
            //拒绝后再次尝试把任务加入队列
            if (!queue.force(command)) {
                submittedCount.decrementAndGet();
                throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));
            }
        } else {
            //其他队列则继续抛出拒绝异常
            submittedCount.decrementAndGet();
            throw rx;
        }
    }
}