Java核心复习——线程池ThreadPoolExecutor源码分析

使用线程池的好处
创建线程和销毁线程会带来资源和时间上的消耗，而且线程池可以对线程进行管理。所以好处如下

• 降低资源的消耗
• 提高响应的速度
• 提高线程的可管理性

一、线程池的实现原理

1. 线程池先判断核心线程池中的所有线程是否都在执行任务，若否，创建新任务，若是进入下一个流程
2. 线程池判断工作队列是否已满，未满，任务加入队列，若满，进入下个流程
3. 线程池判断线程池中的线程是否都在执行任务，若是，则创建新任务，若否，交给饱和策略处理任务。

二、ThreadPoolExecutor源码

2.1 类依赖关系

2.2 源码解析

成员变量

public class ThreadPoolExecutor extends AbstractExecutorService {


//工作队列
private final BlockingQueue<Runnable> workQueue;

//核心线程池大小
private volatile int corePoolSize;

//最大线程池大小
private volatile int maximumPoolSize;

private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();

}

execute方法

//ctl, is an atomic integer packing two conceptual fields
//workerCount, indicating the effective number of threads
//runState,    indicating whether running, shutting down etc
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task.  The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
* 如果小于正在运行的核心线程池的线程数，则尝试开启一个新线程运行任务。
* 
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread.  If it fails, we know we are shut down or saturated
* and so reject the task.
*/
 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);
        }
        //抛出RejectedExecutionException异常
        else if (!addWorker(command, false))
            reject(command);
    }

构造方法

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,//核心线程池大小
                              int maximumPoolSize,//最大线程池大小
                              long keepAliveTime,//线程活动保持时间
                              TimeUnit unit,//线程活动保持时间单位
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

参考文档

《Java并发编程的艺术》