Java线程池三：ThreadPoolExecutor

概述

ThreadPoolExecutor是实现线程池的类，它实现了线程池的所有功能。

解析

继承关系

ThreadPoolExecutor

1	public class ThreadPoolExecutor extends AbstractExecutorService {}

ThreadPoolExecutor继承了抽象类AbstractExecutorService.

ThreadPoolExecutor任务运行逻辑

成员变量

静态属性：运行状态值

// 29位
private static final int COUNT_BITS = Integer.SIZE - 3;
//二进制为29个1
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// 左移29位，高3位作为状态
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 准备停止，等待队列中没有任务了才能停止
private static final int STOP       =  1 << COUNT_BITS; // 停止
private static final int TIDYING    =  2 << COUNT_BITS; // 准备中断
private static final int TERMINATED =  3 << COUNT_BITS; // 中断

非静态属性

// 记录运行信息，高3位代表运行状态，第29位代表工作线程数workerCount
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 提交的等待运行任务队列
private final BlockingQueue<Runnable> workQueue;
// 主锁，操作线程相关属性时获取，例如workers
private final ReentrantLock mainLock = new ReentrantLock();
// 一个work对应一个线程，workers代表所有工作线程，仅在获取主锁时变更
private final HashSet<Worker> workers = new HashSet<>();
// 等待队列，仅在获取主锁时变更
private final Condition termination = mainLock.newCondition();
// 最大池容量，仅在获取主锁时变更
private int largestPoolSize;
// 已完成的任务数，仅在获取主锁时变更
private long completedTaskCount;
// 线程创建工厂
private volatile ThreadFactory threadFactory;
// 拒绝执行器
private volatile RejectedExecutionHandler handler;
// 线程最大等待时间（大于核心线程数或者开启allowCoreThreadTimeOut时采用）
private volatile long keepAliveTime;
// 允许核心线程超时结束
private volatile boolean allowCoreThreadTimeOut;
// 核心线程数
private volatile int corePoolSize;
// 最大线程数
private volatile int maximumPoolSize;
// 默认拒绝执行器（默认是抛出RejectedExecutionException）
private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();
// shutdown权限，System.getSecurityManager()管理权限
private static final RuntimePermission shutdownPerm =
        new RuntimePermission("modifyThread");

运行信息

// 取高3位的运行状态
private static int runStateOf(int c)     { return c & ~COUNT_MASK; }
// 取低29位的工作线程数
private static int workerCountOf(int c)  { return c & COUNT_MASK; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

private static boolean runStateLessThan(int c, int s) {
    return c < s;
}

private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}

private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

private boolean compareAndIncrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect + 1);
}

private boolean compareAndDecrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect - 1);
}

private void decrementWorkerCount() {
    ctl.addAndGet(-1);
}

构造器

构造器中没有复杂逻辑将对应的数据赋值。

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;
}

public ThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
            Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
            threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
            Executors.defaultThreadFactory(), handler);
}

任务工作类Worker（内部类）

线程池中一个线程对应一个Worker。Worker实现了Runnable接口，实际线程运行的Runnable对象是Worker本身，Worker来调度各个任务的运行。

继承关系

Worker继承了AQS，所以自带锁的能力

1	private final class Worker extends AbstractQueuedSynchronizer implements Runnable{}

成员变量

// 运行的线程
final Thread thread;
// 初始的任务，可能为null
Runnable firstTask;
// 完成任务计数
volatile long completedTasks;

构造器

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

方法

其中核心的run方法依赖ThreadPoolExecutor的runWorker方法。

因为Worker是ThreadPoolExecutor的内部类，所以Worker可以使用ThreadPoolExecutor的属性和方法。

public void run() {
    runWorker(this);
}

protected boolean isHeldExclusively() {
    return getState() != 0;
}

protected boolean tryAcquire(int unused) {
    if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

protected boolean tryRelease(int unused) {
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
}

public void lock()        { acquire(1); }
public boolean tryLock()  { return tryAcquire(1); }
public void unlock()      { release(1); }
public boolean isLocked() { return isHeldExclusively(); }

void interruptIfStarted() {
    Thread t;
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
            t.interrupt();
        } catch (SecurityException ignore) {
        }
    }
}

ThreadPoolExecutor的runWorker(Worker w)方法

runWorker方法是线程池线程复用的核心方法，Worker不断循环获取任务队列中的任务，然后执行。

注意：当getTask()方法返回null时，该线程会结束运行

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //  getTask()从等待任务队列中获取
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                    (Thread.interrupted() &&
                    runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task); // 空方法，扩展点
                try {
                    task.run();
                    afterExecute(task, null); // 空方法，扩展点
                } catch (Throwable ex) {
                    afterExecute(task, ex);   // 空方法，扩展点
                    throw ex;
                }
            } finally {
                task = null;
                w.completedTasks++; // 完成任务数累加
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly); // 退出处理
    }
}

获取任务getTask()

任务从任务队列workQueue中获取，对于当前线程数>核心线程数就可以超时，或者允许核心线程超时利用BlockingQueue的poll(long timeout, TimeUnit unit)固定等待时间获取。

getTask()方法会处理线程存活时间，也就是timedOut字段控制，当线程存活时间超时后，getTask()方法返回null，然后runWorker方法判断到null结束循环运行，终止线程。

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();

        // Check if queue empty only if necessary.
        // 线程池结束或者队列为空
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        // 超时时间
        // 当前线程数>核心线程数就可以超时，或者允许核心线程超时
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // 线程数大于最大线程数或者处于超时状态，则返回null结束这个worker
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();// 堵塞获取
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

Worker退出处理processWorkerExit

completedAbruptly为true则代表是异常结束的，不然则是正常结束的。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 异常结束则减worker数
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();
    // 移除worker
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    
    tryTerminate();

    int c = ctl.get();
    // 如果还没停止且worker数太少，则新添加一个worker
    if (runStateLessThan(c, STOP)) {
        // 自然结束的判断一下是不是小于核心线程数或者根本没有线程了
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

核心方法

执行任务入口 execute(Runnable command)

execute是执行任务的入口，抽象类AbstractExecutorService中未实现，该方法交由实现类来实现。

未达到核心线程数上限：执行addWorker新建核心线程,如果成功则返回，不然走后续逻辑；
线程池在运行则加入等待队列，如果成功：
1. 再次判断线程池是否在运行，如果没有运行则将任务从队列中移除；
2. 线程池在运行，但是线程数为0，则执行addWorker新建非核心线程，但是不传入初始任务；
没有成功存入队列，则执行addWorker新建非核心线程；

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();
        // 如果线程池没有running则拒绝
        if (! isRunning(recheck) && remove(command))
            reject(command); 
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 没有存进队列再次尝试执行
    else if (!addWorker(command, false))
        reject(command); // 失败就拒绝
}

添加新线程addWorker

firstTask：新建Worker后立即运行的任务，可以为null
core：代表当前新建的Worker的运行线程是否是核心线程

逻辑：

retry循环：
1. 当前线程池停止或者中断了则返回false；
2. 无限循环：
  1. 如果core参数为true则判断是否达到了核心线程数，不然判断是否达到最大线程数，如果到达了则返回false；
  2. cas线程数加一，如果成功则retry循环break；
  3. 如果线程池没有中断，则重新开始retry循环；
新建工作对象worker；
如果worker.thread不为空：
1. 获取mainLock锁；
2. 线程池在运行或者没有停止且初始任务firstTask为空：
  1. worker.thread已经在运行了，则抛出IllegalThreadStateException；
  2. 将worker加入workers列表；
  3. 如果workers的列表长度大于largestPoolSize，则将largestPoolSize更新为workers的列表长度；
  4. 添加成功标志位workerAdded设置为true;
3. 释放锁；
如果添加成功标志位workerAdded为true，则worker.thread start()且workerStarted标志设置为true；
如果workerStarted标志不为true，则执行addWorkerFailed(worker)方法。

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (int c = ctl.get();;) {
        // Check if queue empty only if necessary.
        // 线程池在运行,等待队列为空则返回false
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP)
                || firstTask != null
                || workQueue.isEmpty()))
            return false;

        for (;;) {
            // workerCountOf(c)当前工作线程数
            // core为true表示当前任务是核心线程运行
            // 则线程数不能超过核心线程数，core为false则线程数不能超过最大线程上限
            if (workerCountOf(c)
                >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                return false;
            // cas线程数加一
            if (compareAndIncrementWorkerCount(c))
                break retry; // 加成功后跳出retry循环
            c = ctl.get();  // Re-read ctl
            // 没有中断
            if (runStateAtLeast(c, SHUTDOWN))
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);// 构建新工作类
        final Thread t = w.thread;// 构造器中调用线程工厂类生成
        if (t != null) {
            // workers操作加锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int c = ctl.get();

                if (isRunning(c) ||
                    (runStateLessThan(c, STOP) && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 添加成功则开始运行
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

添加Worker失败addWorkerFailed(Worker w)

将添加失败的worker移出workers列表，然后更新线程数尝试中断(tryTerminate)

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (w != null)
            workers.remove(w);
        decrementWorkerCount();
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

尝试中断tryTerminate()

只能是SHUTDOWN且线程池和等待队列为空或者 STOP且线程池为空。

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        // 正在运行/中断中/已经中断/有任务而且处于结束状态
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateLessThan(c, STOP) && ! workQueue.isEmpty()))
            return;
        // worker数不为0，则中断workers列表中的第一个worker
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated(); // 空方法
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    // Condition termination = mainLock.newCondition();
                    // 唤醒termination中的等待队列中的线程
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

停止shutdown()

SHUTDOWN状态必须等等待队列中没有任务了才能停止

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();// SecurityManager判断是否有停止权限
        advanceRunState(SHUTDOWN); // 设置状态
        interruptIdleWorkers(); // 中断所有worker
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

立即停止shutdownNow()

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();// SecurityManager判断是否有停止权限
        advanceRunState(STOP);// 设置状态
        interruptWorkers();   // 中断所有worker
        tasks = drainQueue(); // 返回任务队列中还未执行的任务
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

更新状态advanceRunState

private void advanceRunState(int targetState) {
    // assert targetState == SHUTDOWN || targetState == STOP;
    for (;;) {
        int c = ctl.get();
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

中断workers interruptIdleWorkers

private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}
// 如果传true则中断第一个worker
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

等待任务中断

利用mainLock.newCondition()的等待队列，中断是执行tryTerminate()成功后唤醒等待线程。

public boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            while (runStateLessThan(ctl.get(), TERMINATED)) {
                if (nanos <= 0L)
                    return false;
                nanos = termination.awaitNanos(nanos);
            }
            return true;
        } finally {
            mainLock.unlock();
        }
    }

设置核心线程数setCorePoolSize

public void setCorePoolSize(int corePoolSize) {
    if (corePoolSize < 0 || maximumPoolSize < corePoolSize)
        throw new IllegalArgumentException();
    int delta = corePoolSize - this.corePoolSize;
    this.corePoolSize = corePoolSize;
    if (workerCountOf(ctl.get()) > corePoolSize)
        interruptIdleWorkers();
    else if (delta > 0) {
        // We don't really know how many new threads are "needed".
        // As a heuristic, prestart enough new workers (up to new
        // core size) to handle the current number of tasks in
        // queue, but stop if queue becomes empty while doing so.
        int k = Math.min(delta, workQueue.size());
        while (k-- > 0 && addWorker(null, true)) {
            if (workQueue.isEmpty())
                break;
        }
    }
}