Tomcat实现ThreadPoolExecutor和JDK线程池区别

1.1 tomcat线程池和juc线程池流程

jdk 线程池策略：

• 当线程池中线程数量小于 corePoolSize，每来一个任务，就会创建一个线程执行这个任务
• 当前线程池线程数量大于等于 corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列中；若是添加成功，则该任务会等待线程将其取出去执行；若添加失败(一般来说任务缓存队列已满)，则会尝试创建新的线程执行
• 当前线程池线程数量等于 maximumPoolSize，则会采取任务拒绝策略进行处理

tomcat 线程池策略：

• 当前线程数小于 corePoolSize，则去创建工作线程
• 当前线程数大于 corePoolSize，但小于 maximumPoolSize，则去创建工作线程
• 当前线程数大于 maximumPoolSize，则将任务放入到阻塞队列中，当阻塞队列满了之后，则调用拒绝策略丢弃任务

1.2 tomcat线程池和juc线程池的区别

tomcat 线程池是在 juc线程池的基础上，修改少量代码来实现，tomcat 线程池执行流程和 juc的线程池执行流程有着很大的区别，那么tomcat为什么要这样设计？

使用线程池的任务有两种：

• IO 密集型任务(如调用接口、查询数据库)

• CPU 密集型任务

• JDK 的线程池 ThreadPoolExecutor 主要目的解决的便是CPU密集型任务的并发处理

• Tomcat 若使用原生的JDK线程池，一旦接收的请求数量大于线程池的核心线程数，这些请求就会被放到队列中，等待核心线程处理，这样会降低请求的总体处理速度，所以 Tomcat并没有使用JDK原生线程池的策略 (例如：10个核心线程，20个最大线程，15个请求过来的时候，有5个请求放入队列进行等待处理，这也是为什么 tomcat 要使 work 线程优先 queue的原因)

• JDK线程池：当线程数达到 corePoolSize 后，任务首先被放到 queue，发挥CPU多核的并行优势，减少多个线程导致的上下文切换，适合的场景是：CPU密集型任务

• Tomcat线程池：当大量请求达到时，接收的请求数量大于核心线程池的 corePoolSize 时，会继续创建 worker 线程去处理请求，而后续请求量变少时，只会销毁 maximumPoolSize 线程数，适合的场景是：IO密集型

1.3 IO密集型和CPU密集型任务核心参数的设置

• 场景：JDK线程池执行IO密集型的任务，解决方案：可以提高corePoolSize的大小，引入问题：系统中线程数过多

• 线程数过多问题的解决方案：指定 ThreadPoolExecutor 的 allowCoreThreadTimeout=true，那么核心线程若处于闲置状态的话，超过一定的时间(KeepAliveTime)，就会销毁掉

• 引入问题：当核心线程数被销毁时，而有大量请求到达时系统重新创建 worker 线程，会使得前期接口响应时间变长(这也是为什么系统上线需要预热)

• 所以JDK线程池无法完美的去处理IO密集型的任务，这也就是为什么Tomcat需要重写JDK线程池的原因

Tomcat线程池执行流程

• Tomcat 重写 JDK 线程池(即改动的是少量的源码)实现的功能的增强

• 思想：主要流程还是JDK线程池流程，即先开启 corePoolSize 线程，然后在 queue，最后在开启 maximumPoolSize 线程，Tomcat 重点改造的是 queue 的 offer()，即在向 queue 放入任务时，若发现未达到最大线程数，那么 offer() 返回 false，即放入队列失败，此时，便继续开启 maximumPoolSize 线程

自定义队列

Tomcat 主要是通过实现自定义队列来完成逻辑的改造。

/**
 * 实现Tomcat特有逻辑的自定义队列
 */
public class TaskQueue extends LinkedBlockingQueue<Runnable> {
    private static final long serialVersionUID = 1L;

    private transient volatile ThreadPoolExecutor parent = null;

    private static final int DEFAULT_FORCED_REMAINING_CAPACITY = -1;

    /**
     * 强制遗留的容量
     */
    private int forcedRemainingCapacity = -1;

    /**
     * 队列的构建方法
     */
    public TaskQueue() {
    }

    public TaskQueue(int capacity) {
        super(capacity);
    }

    public TaskQueue(Collection<? extends Runnable> c) {
        super(c);
    }

    /**
     * 设置核心变量
     */
    public void setParent(ThreadPoolExecutor parent) {
        this.parent = parent;
    }

    /**
     * put：向阻塞队列填充元素，当阻塞队列满了之后，put时会被阻塞。
     * offer：向阻塞队列填充元素，当阻塞队列满了之后，offer会返回false。
     *
     * @param o 当任务被拒绝后，继续强制的放入到线程池中
     * @return 向阻塞队列塞任务，当阻塞队列满了之后，offer会返回false。
     */
    public boolean force(Runnable o) {
        if (parent == null || parent.isShutdown()) {
            throw new RejectedExecutionException("taskQueue.notRunning");
        }
        return super.offer(o);
    }

    /**
     * 带有阻塞时间的塞任务
     */
    @Deprecated
    public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
        if (parent == null || parent.isShutdown()) {
            throw new RejectedExecutionException("taskQueue.notRunning");
        }
        return super.offer(o, timeout, unit); //forces the item onto the queue, to be used if the task is rejected
    }

    /**
     * 当线程真正不够用时，优先是开启线程（直至最大线程），其次才是向队列填充任务。
     *
     * @param runnable 任务
     * @return false 表示向队列中添加任务失败，
     */
    @Override
    public boolean offer(Runnable runnable) {
        if (parent == null) {
            return super.offer(runnable);
        }
        //若是达到最大线程数，进队列。
        if (parent.getPoolSize() == parent.getMaximumPoolSize()) {
            return super.offer(runnable);
        }
        //当前活跃线程为10个，但是只有8个任务在执行，于是，直接进队列。
        if (parent.getSubmittedCount() < (parent.getPoolSize())) {
            return super.offer(runnable);
        }
        //当前线程数小于最大线程数，那么直接返回false，去创建最大线程
        if (parent.getPoolSize() < parent.getMaximumPoolSize()) {
            return false;
        }
        //否则的话，将任务放入到队列中
        return super.offer(runnable);
    }

    /**
     * 获取任务
     */
    @Override
    public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
        Runnable runnable = super.poll(timeout, unit);
        //取任务超时，会停止当前线程，来避免内存泄露
        if (runnable == null &amp;&amp; parent != null) {
            parent.stopCurrentThreadIfNeeded();
        }
        return runnable;
    }

    /**
     * 阻塞式的获取任务，可能返回null。
     */
    @Override
    public Runnable take() throws InterruptedException {
        //当前线程应当被终止的情况下：
        if (parent != null &amp;&amp; parent.currentThreadShouldBeStopped()) {
            long keepAliveTime = parent.getKeepAliveTime(TimeUnit.MILLISECONDS);
            return poll(keepAliveTime, TimeUnit.MILLISECONDS);
        }
        return super.take();
    }

    /**
     * 返回队列的剩余容量
     */
    @Override
    public int remainingCapacity() {
        if (forcedRemainingCapacity > DEFAULT_FORCED_REMAINING_CAPACITY) {
            return forcedRemainingCapacity;
        }
        return super.remainingCapacity();
    }


    /**
     * 强制设置剩余容量
     */
    public void setForcedRemainingCapacity(int forcedRemainingCapacity) {
        this.forcedRemainingCapacity = forcedRemainingCapacity;
    }

    /**
     * 重置剩余容量
     */
    void resetForcedRemainingCapacity() {
        this.forcedRemainingCapacity = DEFAULT_FORCED_REMAINING_CAPACITY;
    }
}

2.2 自定义线程池ThreadPoolExecutor

Tomcat 线程池 ThreadPoolExecutor 是继承的 AbstractExecutorService 类，但是很多代码依旧使用的是 JDK 的 ThreadPoolExecutor，只是稍微改造了一部分

    public void execute(Runnable command, long timeout, TimeUnit unit) {

        /**
         * 提交任务的数量+1
         */
        submittedCount.incrementAndGet();
        try {
            /**
             * 线程池内部方法，真正执行的方法。就是JDK线程池原生的方法。
             *
             * 因为重写了阻塞队列，才完成Tomcat特有逻辑的实现。
             *
             * 1. 重写队列方法；达到核心线程数，然后向阻塞队列中放置，阻塞队列直接返回false
             * 2. 返回false，则开启maximum pool size；
             * 3. maximum pool size到达极限时，会抛出RejectedExecutionException方法。
             *
             */
            executeInternal(command);
            /**
             * 任务被拒绝
             */
        } catch (RejectedExecutionException rx) {
            /**
             * 在将被拒绝的任务放入到队列中。
             */
            if (getQueue() instanceof TaskQueue) {

                //如果Executor接近最大线程数，应该将任务添加到队列中，而不是拒绝。
                final TaskQueue queue = (TaskQueue) getQueue();
                try {
                    //强制的将任务放入到阻塞队列中
                    if (!queue.force(command, timeout, unit)) {
                        //放入失败，则继续抛出异常
                        submittedCount.decrementAndGet();
                        throw new RejectedExecutionException("threadPoolExecutor.queueFull");
                    }
                } catch (InterruptedException x) {
                    //被中断也抛出异常
                    submittedCount.decrementAndGet();
                    throw new RejectedExecutionException(x);
                }
            } else {
                //不是这种队列，那么当任务满了之后，直接抛出去。
                submittedCount.decrementAndGet();
                throw rx;
            }
        }
    }
    
    /**
     * JDK线程池的任务执行的逻辑
     */
    private void executeInternal(Runnable command) {
        if (command == null) {
            throw new NullPointerException();
        }
        int c = ctl.get();
        //未达到corePoolSize数量，则去开启线程
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true)) {
                return;
            }
            c = ctl.get();
        }
        //开启到corePoolSize数量的工作线程，则将任务放入队列。
        //但是Tomcat重写了阻塞队列。
        //当放入workQueue.offer(command)返回false，则继续开启线程数
        if (isRunning(c) &amp;&amp; workQueue.offer(command)) {
            int recheck = ctl.get();
            if (!isRunning(recheck) &amp;&amp; remove(command)) {
                reject(command);
            } else if (workerCountOf(recheck) == 0) {
                addWorker(null, false);
            }
        } else if (!addWorker(command, false)) {
            //开启最大线程失败，则任务被拒绝。
            reject(command);
        }
    }

注意事项

tomcat 的优化

    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) {
            decrementWorkerCount();
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }

        tryTerminate();

        int c = ctl.get();
        if (runStateLessThan(c, STOP)) {
            if (!completedAbruptly) {
                //获取剩余的最小线程数，若配置了允许最小线程数关闭的参数，min=0
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                //当配置了允许最小线程关闭参数，且队列不为空的情况下，允许保留的最小线程数为1
                /**
                 * 这是关闭空闲核心线程数时的判断。
                 */
                if (min == 0 && !workQueue.isEmpty()) {
                    min = 1;
                }
                // https://bz.apache.org/bugzilla/show_bug.cgi?id=65454
                // If the work queue is not empty, it is likely that a task was
                // added to the work queue between this thread timing out and
                // the worker count being decremented a few lines above this
                // comment. In this case, create a replacement worker so that
                // the task isn't held in the queue waiting for one of the other
                // workers to finish.
                /**
                 * 若当前工作数量>允许的最小线程数，那么关闭改线程。
                 * Tomcat对此处进行了优化。
                 */
                if (workerCountOf(c) >= min && workQueue.isEmpty()) {
                    return; // replacement not needed
                }
            }
            /**
             * 没有终止线程，又重新开启线程
             */
            addWorker(null, false);
        }
    }

代码007普通

打赏 收藏 海报 链接