Tomcat的StandardThreadExecutor

Posted on In , ,

JDK中的ThreadExecutor中的execute方法的处理逻辑应该都知道：

1.小于等于Coresize：创建线程执行；
2.大于CoreSize：加入队列；
3.队列满且小于maxSize：有空闲线程使用空闲线程执行，没有的话，创建线程执行；如果大于maxSize则拒绝策略执行。

这样会造成一个现象，如果设置的不恰当，队列使用LinkedBlockingQueue，那么线程数量是很不可能达到maximumPoolSize，因为线程数量到了corePoolSize之后，之后新的任务是添加到队列里面去了。

Tomcat中有一个StandardThreadExecutor线程池，该线程池execute执行策略是优先扩充线程到maximumPoolSize，再offer到queue，如果满了就reject。来看看StandardThreadExecutor是怎么实现的。

首先Tomcat的StandardThreadExecutor类实现了org.apache.catalina.Executor接口,该接口继承了java.util.concurrent.Executor。StandardThreadExecutor内部组合了一个org.apache.tomcat.util.threads.ThreadPoolExecutor,该类继承了java.util.concurrent.ThreadPoolExecutor。

先来关注一下java.util.concurrent.ThreadPoolExecutor的execute方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    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);
    }
    else if (!addWorker(command, false))
        reject(command);
}

addWorker是创建线程执行任务，该方法中会判断当前的线程数量是否达到maximumPoolSize；workQueue.offer是将任务添加到队列里面。

org.apache.tomcat.util.threads.ThreadPoolExecutor就是对workQueue作了一些改动。StandardThreadExecutor使用的是org.apache.tomcat.util.threads.TaskQueue：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class TaskQueue extends LinkedBlockingQueue<Runnable> {

    private volatile ThreadPoolExecutor parent = null;

    public boolean force(Runnable o) {
        if ( parent==null || parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
        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("Executor not running, can't force a command into the queue");
        return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
    }

    @Override
    public boolean offer(Runnable o) {
      //we can't do any checks
        if (parent==null) return super.offer(o);
        //we are maxed out on threads, simply queue the object
        if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
        //we have idle threads, just add it to the queue
        if (parent.getSubmittedCount()<(parent.getPoolSize())) return super.offer(o);
        //if we have less threads than maximum force creation of a new thread
        if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
        //if we reached here, we need to add it to the queue
        return super.offer(o);
    }

    //……
}

看一下TaskQueue的offer方法，是不是明白了？当parent.getPoolSize()还没达到parent.getMaximumPoolSize()大小时，返回false，这样会走addWorker分支。再来看org.apache.tomcat.util.threads.ThreadPoolExecutor的execute:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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("Queue capacity is full.");
                }
            } catch (InterruptedException x) {
                submittedCount.decrementAndGet();
                throw new RejectedExecutionException(x);
            }
        } else {
            submittedCount.decrementAndGet();
            throw rx;
        }

    }
}

当抛出RejectedExecutionException时，会等待timeout后重试一次加到queue中，如果失败才最终抛出异常。