Tomcat线程池的优化

Tomcat线程池的优化


1. 简介

Tomcat继承并重写了JAVA原生的java.util.concurrent.ThreadPoolExecutor,增加了一些更有效率的方法,并且定义了默认拒绝策略。


2. 原理分析


2.1 实例化

与阿里巴巴JAVA规范中定义的线程池创建规范类似,Tomcat实例化线程池时,同样限制了任务队列的长度,maxQueueSize默认为Integer.MAX_VALUE,但是可以通过修改配置xml注入新值。

另一方面,Tomcat同样设置了线程池的核心线程数与最大线程数,默认为25和200。

    /**
     * max number of threads
     */
    protected int maxThreads = 200;

    /**
     * min number of threads
     */
    protected int minSpareThreads = 25;

    protected void startInternal() throws LifecycleException {

        taskqueue = new TaskQueue(maxQueueSize);
        TaskThreadFactory tf = new TaskThreadFactory(namePrefix,daemon,getThreadPriority());
        executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), maxIdleTime, TimeUnit.MILLISECONDS,taskqueue, tf);
        executor.setThreadRenewalDelay(threadRenewalDelay);
        if (prestartminSpareThreads) {
            executor.prestartAllCoreThreads();
        }
        taskqueue.setParent(executor);

        setState(LifecycleState.STARTING);
    }


2.2 构造函数

Tomcat ThreadPoolExecutor构造函数,均调用了java.util.concurrent.ThreadPoolExecutor的prestartAllCoreThreads方法,创建线程池对象后,直接启动所有核心线程,以提高系统效率。

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        prestartAllCoreThreads();
    }

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
            RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        prestartAllCoreThreads();
    }

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new RejectHandler());
        prestartAllCoreThreads();
    }

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new RejectHandler());
        prestartAllCoreThreads();
    }


    /**
     * Starts all core threads, causing them to idly wait for work. This
     * overrides the default policy of starting core threads only when
     * new tasks are executed.
     *
     * @return the number of threads started
     */
    public int prestartAllCoreThreads() {
        int n = 0;
        while (addWorker(null, true))
            ++n;
        return n;
    }



2.3 execute

Tomcat ThreadPoolExecutor主要改动在于任务提交方法execute(Runnable command),修改后的逻辑如下:

  • CAS修改submittedCount,submittedCount为已提交且未结束的任务数量,包括任务队列中的任务和执行中的任务
  • 调用super.execute方法提交任务
  • 捕获RejectedExecutionException异常,尝试将任务添加到任务队列中
  • 如果添加失败,则抛出RejectedExecutionException
    public void execute(Runnable command, long timeout, TimeUnit unit) {
        // CAS submittedCount+1
        submittedCount.incrementAndGet();
        try {
            // 提交任务
            super.execute(command);
        } catch (RejectedExecutionException rx) {
            if (super.getQueue() instanceof TaskQueue) {
                final TaskQueue queue = (TaskQueue)super.getQueue();
                try {
                    // 入队,如果队列满则阻塞,直到timeout
                    if (!queue.force(command, timeout, unit)) {
                        // 入队失败,则CAS submittedCount-1
                        submittedCount.decrementAndGet();
                        throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));
                    }
                } catch (InterruptedException x) {
                    // 阻塞中被中断,则CAS submittedCount-1
                    submittedCount.decrementAndGet();
                    throw new RejectedExecutionException(x);
                }
            } else {
                submittedCount.decrementAndGet();
                throw rx;
            }

        }
    }

Copy


2.4 TaskQueue

TaskQueue继承了LinkedBlockingQueue类,主要添加了两个方法force(Runnable o)和force(Runnable o, long timeout, TimeUnit unit),根据线程池状态决定是否调用队列的offer方法。

    public boolean force(Runnable o) {
        if ( parent==null || parent.isShutdown() ) throw new RejectedExecutionException(sm.getString("taskQueue.notRunning"));
        return super.offer(o); //forces the item onto the queue, to be used if the task is rejected
    }

    public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
        if ( parent==null || parent.isShutdown() ) throw new RejectedExecutionException(sm.getString("taskQueue.notRunning"));
        return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
    }



2.5 拒绝策略

当任务队列满时,父类的execute直接抛出RejectedExecutionException异常。

    private static class RejectHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r,
                java.util.concurrent.ThreadPoolExecutor executor) {
            throw new RejectedExecutionException();
        }

    }



3. 总结

总得来说,Tomcat在原生的ThreadPoolExecutor的基础上,做了小范围的修改,部分提升了高并发下的性能,并减小了错误率。比较而言,Jetty自研的QTP与Java原生的线程池差别要大的多。

评论区
Rick ©2018