初识Worker

线程池顾名思义是存放线程的池子，ThreadPoolExecutor是语言级别上对它的定义，既然要存放线程，那么其内部就需要某种数据结构来存储代表线程的Thread对象。Java API定义的容器类可以用来完成存储的功能，由于线程集中线程的松散性和不可重复性的特点，选用Set来存储比较合适。

那么，就可以在ThreadPoolExecutor类中这样定义：

private final HashSet<Thread> threads = new HashSet<Thread>();

存储的问题我们解决了，那么下一个问题就是线程执行用户任务的问题，虽然现在还没提到任务队列，不过就当成我们已经知道了，需要执行的用户任务会被放入到ThreadPoolExecutor内部定义的任务队列里，线程池的线程会从这个任务队列里取出用户任务来执行。

ThreadPoolExecutor-Worker-1.png

线程池中的线程要取用户任务并执行，那么线程池中的线程类要这样定义：

public class ThreadPoolThread extends Thread {
    @Override
    public void run() {
        // 获取队列中的用户任务
        Runnable userTask =  getUserTaskFromQueue();  
        userTask.run();
    }
}

我们基于Runnable接口编程，"获取队列中的用户任务并执行"的逻辑代码可以剥离出来，定义成一个新的任务类：

public class ThreadTask implements Runnable {
    @Override
    public void run() {
        // 获取并执行队列中的任务
        Runnable userTask =  getUserTashFromQueue();
        userTask.run();
    }
}

这样定义后，创建线程并放到线程池中，就可以这样做：

ThreadTask threadTask = new ThreadTask();
Thread thread = new Thread(threadTask);
thread.start();
threads.put(thread);

线程池中的线程需要类似上面代码那样创建后放入到线程池中的，那么上面的这段代码什么时候执行？

有两种方案：一种是在线程池启动的时候我们就创建一定数量的线程，也就是所谓的"预先启动线程"，这种方案的策略是不管有没有任务需要执行，预先启动一定数量的线程然后等着任务到来；还有一种方案就是提交任务的时候再创建线程，只要数量不达到一个预定的值即可。提交任务的时候再创建线程，那么想一下，这个任务有必要放到任务队列中吗？用刚创建的线程来执行这个任务不就行了嘛？是的，的确是这样。回头看一下上面创建线程的代码，也就是说：

Thread thread = new Thread(threadTask);

这里操作的时候就希望把用户任务设置进去，但是threadTask本身就是一个任务对象，没有方法设置两个任务对象。那就需要对ThreadTask改造了，改造也很简单，类似这样：

public class ThreadTask implements Runnable {
    
    Runnalbe userTask;

    public ThreadTask (Runnable userTask) {
        this.userTask = userTask;
    }

    @Override
    public void run() {
        if (userTask != null || (userTask = getUserTaskFromQueue() ) != null)  {
            userTask.run();
        }
    }
  
    public Runnalbe getUserTaskFromQueue() {
        Runnable queueTask =  getUserTashFromQueue();
        return queueTask;
    }

}

上面的代码还有个问题，就是线程需要不断地从队列中取用户任务执行，而我们知道线程执行完run方法后就会停止，所以上面的代码达不到循环取任务的效果，需要对run方法的逻辑进行修改，如下：

while (userTask != null || (userTask = getUserTaskFromQueue() ) != null)  {
     userTask.run();
}

重新定义ThreadTask之后，创建线程池线程的代码，修改成如下所示：

Runnable userTask = new UserTask();
ThreadTask threadTask = new ThreadTask(userTask);
Thread thread = new Thread(threadTask);
threads.put(thread);
thread.start();

ThreadTask需要Thread的调用，也可以说其实就是Thread的run方法中的一部分，因为我们基于Runnable编程所以把它给提取出来了，既然ThreadTask是依托于Thread的，那么创建ThreadTask的时候就把要运行它的线程确定下来可以吗？是可以的，对ThreadTask改造如下：

public class ThreadTask implements Runnable {
    
    Runnalbe userTask;

    Thread thread;

    public ThreadTask (Runnable userTask) {
        this.userTask = userTask;
        thread = new Thread(this);
    }

    @Override
    public void run() {
        while (userTask != null || (userTask = getUserTaskFromQueue() ) != null)  {
            userTask.run();
        }
    }
  
    public Runnalbe getUserTaskFromQueue() {
        Runnable queueTask =  getUserTashFromQueue();
        return queueTask;
    }

}

上面的代码中我们将Thread作为了ThreadTask的一个属性，在ThreadTask的构造方法中通过this将自身对象设置进去。这样改造后，创建线程池线程的代码，修改成如下所示：

Runnable userTask = new UserTask();
ThreadTask threadTask  =  new ThreadTask(userTask);
threadTask.thread.start();// 启动线程

经过上面的一系列改造，ThreadTask具有了获取队列用户任务以及执行用户任务的功能，那么ThreadPoolExecutore只要持有ThreadTask的集合集合，也就是说我们上面的Thread集合可以修改成：

private final HashSet<ThreadTask> threads = new HashSet<ThreadTask>();

内部任务的执行就是通过ThreadTask这个对象的操作来完成，将用户交给它，它会启动内部持有的线程，然后执行用户任务。

有没有觉得ThreadTask这个名字并不完全符合了。这里不用费心思考合适的名字，它就是这篇文章要介绍的类——Worker的简化版。

JDK中Worker类的定义：

ThreadPoolExecutor$Worker

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.

    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) {
            }
        }
    }
}

中文翻译把它称之为：工作线程或工作者。

上面贴出来的代码可以看到，run方法的执行逻辑交给了外部定义的runWorker方法。

runWorker

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        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);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

Worker锁

可以看到上面贴出来的Worker类继承了AQS，重写了锁获取与释放的方法，Worker具有了锁的特性，而且是一把互斥锁。runWorker使用了Worker这把互斥锁，通过它保护用户任务的执行。搜索ThreadPoolExecutor的源码可以发现除了Worker本身的run方法调用了runWorker之外，没有其他地方执行了调用，而run方法的调用本身就是线程封闭式的，不会存在一个线程闯入进来打乱线程的执行。也就是runWorker的执行不存在竞争，为什么要用锁来保护？

目的是中断的控制, 正如Worker类注释锁描述的那样：

Class Worker mainly maintains interrupt control state for threads running tasks...

控制中断？怎么说？

这样说，现在问你，Worker代表工作线程，它正在执行任务，这时候想要中断它怎么做？Woker严格意义来说不是线程对象，但是它的内部包含了线程对象，这个线程对象就是运行Worker称之为工作线程的体现。想中断Worker的操作，那么中断它内部的这个线程就行了，于是我们可以这样操作：

Thread t = worker.thread;
t.interrupt();

通过这样的方式就可以中断Worker的执行。前提是可以持有想中断的线程对象的Worker对象。

但是现在的需求是在Worker工作的时候不允许中断它。也就是在执行runWorker内部逻辑的时候不允许中断正在运行它的线程。这时候runWorker使用的Worker锁就起到作用了。

对某一个线程的中断通常是另外一个线程发起的，也就是

Thread t = worker.thread;
t.interrupt();

这段代码是另外一个线程调用的，于是我们可以这样操作：

if (worker.tryLock()) {
    Thread t = worker.thread;
    t.interrupt();
}

如果worker锁现在被想中断的那个线程持有，那么上面的执行线程想获取锁就会失败，无法执行中断操作。

这样通过Worker锁的控制保护了正在执行的任务不能被中断。

还有一点在Worker的构造还是是这样定义的：

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

这里可以看到state的初始值为-1，为什么定义这个数字，其实后面的解释已经很明确了：在进入runWorker方法之前不允许中断。在runWorker进入后用户任务执行之前，做了释放锁的操作，将state的值改为了0，也就是说runWorker锁保护块代码之前可以被中断：

w.unlock(); // allow interrupts

用户任务没执行前被中断ThreadPoolExecutor是允许的。

非重入

看先Worker的tryAcquire方法定义：

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

这里的逻辑规定了不能重复获取锁，这个是为什么呢？

首先要明确一点Worker是来执行用户任务的，用户任务的开始执行的执行点在在runWorker方法中：

task.run();

假设现在执行这个调用的是线程A，这时候线程A是持有Worker锁的，我们在自己调用的这个task的run方法中执行这样的操作：

class MyTask implements Runnable {
    @Override
    public void run() {
        threadPoolExecutor.setCorePoolSize(10);
    }
}

执行run方法的当前线程还是线程A，来看看setCorePoolSize方法：

setCorePoolSize

public void setCorePoolSize(int corePoolSize) {
    if (corePoolSize < 0)
        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;
        }
    }
}

这里有个interruptIdleWorkers方法的调用：

interruptIdleWorkers

private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}

interruptIdleWorkers

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

现在线程A执行到这里，刚好w就是它在runWorker执行时候持有的锁，

if (!t.isInterrupted() && w.tryLock()) 

如果允许重入，那么w.tryLock就是true，那么线程A就把它自己它打断了。

所以Worker类在设计的时候就设计成是不可重入。

Worker执行异常处理

Worker执行出现异常指的是什么？

还是来看一下runWorker代码：

runWorker

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        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);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

注意第2个try...catch块，task是用户自定的任务，在运行的时候可能会出现异常，出现异常后，从上面代码中可以看到捕获异常后被抛出了，那么下面这一句就无法执行：

completedAbruptly = false;

也就是completedAbruptly为初始值true，while循环结束，执行processWorkerExit方法，现在来看一下processWorkerExit的处理：

• processWorkerExit*

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        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) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

#第1步：减workerCount（减1）

if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
    decrementWorkerCount();

#第2步：将当前Worker完成的任务数加到总任务数中，并从Worker集合中移除当前Worker

completedTaskCount += w.completedTasks;
workers.remove(w);

#第3步：尝试终止线程池

tryTerminate();

在对线程池有负效益的操作时，都需要“尝试终止”线程池。

终止线程池不在这里做说明。

#第4步：如果线程池状态是running或shutdown就尝试增加一个新的Worker

int c = ctl.get();
if (runStateLessThan(c, STOP)) {
    ......
    // completedAbruptly为true，上面的代码忽略
    addWorker(null, false);
}

这里调用addWorker方法不一定能增加成功，addWorker内部有具体的判断逻辑。

addWorker逻辑不在这里做说明。