摘要：表示一個(gè)異步任務(wù)的結(jié)果，就是向線程池提交一個(gè)任務(wù)后，它會(huì)返回對(duì)應(yīng)的對(duì)象。它們分別提供兩個(gè)重要的功能阻塞當(dāng)前線程等待一段時(shí)間直到完成或者異常終止取消任務(wù)。此時(shí)，線程從中返回，然后檢查當(dāng)前的狀態(tài)已經(jīng)被改變，隨后退出循環(huán)。

前段時(shí)間需要把一個(gè)C++的項(xiàng)目port到Java中，因此時(shí)隔三年后重新熟悉了下Java。由于需要一個(gè)通用的線程池，自然而然就想到了Executors。

用了后，感覺很爽... 于是忍不住摳了下源碼。因此就有了這篇學(xué)習(xí)筆記。

言歸正傳，Java Executor是一個(gè)功能豐富，接口設(shè)計(jì)很好的，基于生產(chǎn)者-消費(fèi)者模式的通用線程池。這種線程池的設(shè)計(jì)思想也在很多地方被應(yīng)用。

在這篇文章中，我并不打算介紹java線程池的使用，生產(chǎn)者-消費(fèi)者模式，并發(fā)編程基本概念等。

通常來說，一個(gè)線程池的實(shí)現(xiàn)包括四個(gè)部分：

執(zhí)行任務(wù)的線程

用于封裝任務(wù)的task對(duì)象

存儲(chǔ)任務(wù)的數(shù)據(jù)結(jié)構(gòu)

線程池本身

Thread 并不是concurrent包的一部分。Thread包含著name, priority等成員和對(duì)應(yīng)的操作方法。

它是繼承自runable的，也就是說線程的入口函數(shù)是run。它的繼承體系和重要操作函數(shù)如下圖：

它實(shí)現(xiàn)了一系列包括sleep, yield等靜態(tài)方法。以及獲取當(dāng)前線程的靜態(tài)方法currentThread()。這些都是native方法。

值得注意的是它的中斷機(jī)制（雖然它也實(shí)現(xiàn)了suspend和resume方法，但是這兩個(gè)方法已被棄用）：

通過調(diào)用interrupt來觸發(fā)一個(gè)中斷

isInterrupted() 用來查詢線程的中斷狀態(tài)

interrupted() 用來查詢并清除線程的中斷狀態(tài)

public void interrupt() {
    if (this != Thread.currentThread())
        checkAccess();
    synchronized (blockerLock) {
        Interruptible b = blocker;
        if (b != null) {
            interrupt0();           // Just to set the interrupt flag
            b.interrupt(this);
            return;
        }
    }
    interrupt0();
}

在默認(rèn)的情況下，blocker (Interruptible 成員變量）的值為null, 這時(shí)調(diào)用interrupt，僅僅是調(diào)用interrupt0設(shè)置一個(gè)標(biāo)志位。

而如果blocker的值不為null，則會(huì)調(diào)用其interrupt方法實(shí)現(xiàn)真正的中斷。

(關(guān)于blocker值何時(shí)被設(shè)置，在后面會(huì)看到一個(gè)使用場(chǎng)景。)

當(dāng)線程處于可中斷的阻塞狀態(tài)時(shí)，比如說阻塞在sleep, wait, join,select等操作時(shí)，調(diào)用interrupt方法會(huì)讓線程從阻塞狀態(tài)退出，并拋出InterruptedException。

值得注意的一點(diǎn)是：interrupt讓我們從阻塞的方法中退出，但線程的中斷狀態(tài)卻并不會(huì)被設(shè)置！

try {
    Thread.sleep(10);
}
catch (InterruptedException e) {
    System.out.println("IsInterrupted: " + Thread.currentThread().isInterrupted());
}

如上述示例代碼，此時(shí)你得到的輸出是： IsInterrupted : false 。這是一個(gè)有點(diǎn)令人意外的地方。
上述代碼并不是一個(gè)好的示例，因?yàn)閕nterrupt被我們“吃”掉了！除非你明確的知道這是你想要的。否則的話請(qǐng)考慮在異常捕獲中(catch段中)加上：

Thread.currentThread.interrupt();

Java可執(zhí)行的接口類有兩種，Runnable和Callable，它們的區(qū)別是Callable可以帶返回值，一個(gè)需要實(shí)現(xiàn)Run()方法，另一個(gè)需要實(shí)現(xiàn)帶返回值的Call() 方法。

在java.util.concurret中還有另外一個(gè)接口類Future。

Future表示一個(gè)異步任務(wù)的結(jié)果，就是user code向線程池提交一個(gè)任務(wù)后，它會(huì)返回對(duì)應(yīng)的 Future對(duì)象。用以觀察任務(wù)執(zhí)行的狀態(tài)(isCancelled, isDone)，取消任務(wù)(Cancel)或者等待任務(wù)執(zhí)行(get, timeout get)。

如上圖，RunnableFuture是一個(gè)中間類，它將Runnable和Future的功能糅合到一起。FutureTask 則是真正的實(shí)現(xiàn)。

FutureTask可以從一個(gè)Runnable和Callable構(gòu)造，當(dāng)通過Runnable構(gòu)造時(shí)，它會(huì)調(diào)用Excutors.callable接口將其轉(zhuǎn)為Callable對(duì)象保存起來。

從上面的類圖中可以看出，F(xiàn)utureTask除了簡(jiǎn)單的狀態(tài)查詢等接口外，還具有兩個(gè)重要的接口：get() 和 get(long timeout, TimeUnit unit))， cancel(bool mayInterruptIfRunning)。

它們分別提供兩個(gè)重要的功能：阻塞（當(dāng)前線程）等待（一段時(shí)間）直到task完成或者異常終止；取消任務(wù)。

一個(gè)任務(wù)具有三種狀態(tài)：尚未運(yùn)行，正在運(yùn)行，已經(jīng)執(zhí)行完畢。

在調(diào)用cancel后，如果任務(wù)處于已經(jīng)執(zhí)行完畢了，則不需要做任何事情直接返回；

如果任務(wù)尚未運(yùn)行，將其狀態(tài)設(shè)為cancelled；
如果任務(wù)正在執(zhí)行，而且user以cancel(true)的方式取消這個(gè)任務(wù)。那么FutureTask會(huì)通過調(diào)用Thread.interrupt來終止當(dāng)前任務(wù)。

public boolean cancel(boolean mayInterruptIfRunning) {
    // 任務(wù)已經(jīng)完成或者被中斷等其他狀態(tài)
    if (state != NEW)
        return false;
    if (mayInterruptIfRunning) {
        // 正在運(yùn)行，或者尚未運(yùn)行
        if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
            return false;
        Thread t = runner;
        if (t != null)
            t.interrupt();
        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
    }
    // 設(shè)置cancel標(biāo)志位
    else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
        return false;
    finishCompletion();
    return true;
}

注意到： FutureTask并沒有一個(gè)RUNNING的狀態(tài)來標(biāo)識(shí)該任務(wù)正在執(zhí)行。正常的情況下，任務(wù)從開始創(chuàng)建直到運(yùn)行完畢，這段過程的狀態(tài)都是NEW。

user code可以調(diào)用get() 接口等待任務(wù)完成或者調(diào)用get(long, TimeUnit)等待一段時(shí)間。但get()接口被調(diào)用，當(dāng)前的線程將被掛起，直到條件滿足（任務(wù)完成或者異常退出）。

在前文中我們了解到，Thread并沒有提供掛起和阻塞的方法。在這里，Java利用LockSupport類來實(shí)現(xiàn)目的。（我猜測(cè)其中用了類似條件變量的方法來實(shí)現(xiàn)）。

LockSupport也屬于concurrent。FutureTask利用它的park (parkNanos)和unpark方法來實(shí)現(xiàn)線程的掛起和恢復(fù):

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    unsafe.park(false, 0L);
    setBlocker(t, null);
}

public static void unpark(Thread thread) {
    if (thread != null)
        unsafe.unpark(thread);
}

其中parkNanos跟park方法并無本質(zhì)區(qū)別，只是多了一個(gè)timeout參數(shù)。FutureTask分別用它們來實(shí)現(xiàn)get和timeout的get。

注意到上面的setBlocker方法了嗎？沒錯(cuò)，它就是給在上文Thread.interrupt方法中出現(xiàn)過的Thread成員變量blocker賦值。從這我們可以看出，它是可中斷的。

而它真正實(shí)現(xiàn)掛起的則是依賴unsafe類。unsafe類在concurrent中頻繁出現(xiàn)，但sun去并不建議使用它。

它除了提供park，unpark方法外，還提供了一些內(nèi)存和同步原語。比如CAS等。

調(diào)用get()的線程可以是一個(gè)，也可以是多個(gè)。為了能夠在恰當(dāng)?shù)臅r(shí)機(jī)將它們一一恢復(fù)，F(xiàn)utureTask內(nèi)部需要維護(hù)一個(gè)鏈表來記錄所有的等待線程：waiters.

static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
}

至此，我們終于了解get的全貌了。get會(huì)調(diào)用awaitDone方法來實(shí)現(xiàn)阻塞。當(dāng)然，只有兩個(gè)狀態(tài)需要處理：NEW, COMPLETING。

NEW的狀態(tài)在前文已經(jīng)有介紹過。COMPLETING狀態(tài)通常持續(xù)較短，在FutureTask 內(nèi)部的callable 的call方法調(diào)用完畢后，會(huì)需要將call的返回值設(shè)置到outcome這個(gè)成員變量。隨后將狀態(tài)設(shè)為NORMAL。這期間的狀態(tài)就是COMPLETING。

顯而易見，對(duì)于這種狀態(tài)，我們只需要調(diào)用yield讓出線程資源，使得FutureTask完成這一過程即可。

private int awaitDone(boolean timed, long nanos)
     throws InterruptedException {
     final long deadline = timed ? System.nanoTime() + nanos : 0L;
     WaitNode q = null;
     boolean queued = false;
     for (;;) {
         if (Thread.interrupted()) {
             removeWaiter(q);
             throw new InterruptedException();
         }
         int s = state;
         if (s > COMPLETING) { // 1
             if (q != null)
                 q.thread = null;
             return s;
         }
         else if (s == COMPLETING) // cannot time out yet  2
             Thread.yield();
         else if (q == null)  // 3
             q = new WaitNode();
         else if (!queued)  // 4
             queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                  q.next = waiters, q);
         else if (timed) { // 5
             nanos = deadline - System.nanoTime();
             if (nanos <= 0L) {
                 removeWaiter(q);
                 return state;
             }
             LockSupport.parkNanos(this, nanos);
         }
         else  // 6
             LockSupport.park(this);
     }
 }

當(dāng)任務(wù)處于NEW狀態(tài)正在被執(zhí)行時(shí)，其他線程調(diào)用get而進(jìn)入awaitdone函數(shù)。

此時(shí)的流程是 3 -> 4 -> 5 或者 3 -> 4 -> 6。

它會(huì)首先分配一個(gè)WaitNode對(duì)象 --> 把它插入到waiters鏈表的表頭 --> 然后開始等待。那么park函數(shù)何時(shí)返回呢？

對(duì)應(yīng)的unpark被調(diào)用（或者在這之前已經(jīng)被調(diào)用）

如果設(shè)置了timeout的，會(huì)在時(shí)間到達(dá)后退出。

被中斷。

其他異常。

當(dāng)任務(wù)執(zhí)行完畢(或者被cancel)時(shí)，F(xiàn)utureTask會(huì)調(diào)用最終調(diào)用finishcompletion，改函數(shù)會(huì)改變FutureTask狀態(tài)，并調(diào)用LockSupport.unpark方法。

此時(shí)，awaitDone線程從park中返回，然后檢查當(dāng)前的狀態(tài)已經(jīng)被改變，隨后退出for循環(huán)。

FutureTask是會(huì)被多個(gè)線程訪問的，涉及到臨界區(qū)的保護(hù)，但是其內(nèi)部卻并沒有任何的鎖操作。而在該類定義的末尾，有這樣的代碼。

private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
    try {
        UNSAFE = sun.misc.Unsafe.getUnsafe();
        Class k = FutureTask.class;
        stateOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("state"));
        runnerOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("runner"));
        waitersOffset = UNSAFE.objectFieldOffset
            (k.getDeclaredField("waiters"));
    } catch (Exception e) {
        throw new Error(e);
    }
}

這段代碼會(huì)在類被加載時(shí)執(zhí)行一次。注意到它利用getDeclaredField反射機(jī)制來保存了三個(gè)offset:
stateOffset,runnerOffset,waitersOffset分別對(duì)應(yīng)著state,runner,waiters這三個(gè)成員的偏移量。

FutureTask真是對(duì)這三個(gè)成員變量進(jìn)行CAS操作來保證原子性和無鎖化的。實(shí)現(xiàn)CAS的類正是上文出現(xiàn)過的sun.misc.Unsafe類。

UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())

第一個(gè)參數(shù)是對(duì)象指針，第二個(gè)是偏移量，第三個(gè)是舊值，最后一個(gè)是新值。詳細(xì)可參考Unsafe文檔。

java實(shí)現(xiàn)了生產(chǎn)者-消費(fèi)者模式的隊(duì)列。由于隊(duì)列的容量有限，因此涉及到在隊(duì)列為空的時(shí)候取task和在隊(duì)列已滿的時(shí)候存task的策略，連同一系列的查詢函數(shù)一起，BlockingQueue包含著11個(gè)靜態(tài)方法。

BlockingQueue只是一個(gè)interface，它的實(shí)現(xiàn)類包括鏈表方式的LinkedBlockingQueue 、數(shù)組方式的ArrayBlockingQueue以及PriorityBlockingQueue等。

下面以LinkedBlockingQueue為例來了解一下它的實(shí)現(xiàn)。

LinkedBlockingQueue是一個(gè)FIFO的隊(duì)列，它真正用來存儲(chǔ)元素的節(jié)點(diǎn)類型是Node ：

static class Node {
    E item;
    Node next;
    Node(E x) { item = x; }
}

對(duì)應(yīng)的，在LinkedBlockingQueue中保存了頭節(jié)點(diǎn)和尾節(jié)點(diǎn) :

/**
 * Head of linked list.
 * Invariant: head.item == null
 */
private transient Node head;
/**
 * Tail of linked list.
 * Invariant: last.next == null
 */
private transient Node last;

在LinkedBlockingQueue中，Java使用了雙鎖機(jī)制，分別對(duì)頭節(jié)點(diǎn)和尾節(jié)點(diǎn)加鎖。這樣取和存的操作就可以同時(shí)進(jìn)行了。

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

以take為例，獲取并移除此隊(duì)列的頭部，在元素變得可用之前一直等待（可被打斷）。

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        while (count.get() == 0) {
            notEmpty.await();
        }
        x = dequeue();
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}

它將會(huì)一直阻塞在notEmpty.await()上，直到信號(hào)到達(dá)或者被中斷。注意到它只需要對(duì)takeLock加鎖，而無需對(duì)putLock加鎖。

相應(yīng)的，put操作也只需要鎖上putLock就可以了。

有的操作則需要兩個(gè)鎖都鎖上，比如說remove，因?yàn)槲覀儾淮_定要?jiǎng)h除的元素的位置。

public boolean remove(Object o) {
    if (o == null) return false;
    fullyLock();
    try {
        for (Node trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
            if (o.equals(p.item)) {
                unlink(p, trail);
                return true;
            }
        }
        return false;
    } finally {
        fullyUnlock();
    }
}

可以看到LinkedBlockingQueue 并沒有直接調(diào)用lock，而是通過fullyLock和fullyUnLock來加解鎖以保證一致性，避免死鎖:

/**
 * Lock to prevent both puts and takes.
 */
void fullyLock() {
    putLock.lock();
    takeLock.lock();
}
/**
 * Unlock to allow both puts and takes.
 */
void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

當(dāng)然，雙鎖隊(duì)列在插入第一個(gè)元素和最后一個(gè)元素出隊(duì)的時(shí)候會(huì)有沖突。這里的解決辦法是加了一個(gè)哨兵，開始的時(shí)候，頭尾節(jié)點(diǎn)都指向這個(gè)哨兵，在隨后的操作中，頭結(jié)點(diǎn)始終指向哨兵，而尾節(jié)點(diǎn)指向真正有效的值。

有了前面這些零件，我們就可以開始組裝線程池對(duì)象了。java里面Executors的真正實(shí)現(xiàn)類主要包括兩個(gè)ThreadPollExecutors和ScheduledThreadPoolExecutor。其中ScheduledThreadPoolExecutor通過實(shí)現(xiàn)其基類ScheduledExecutorService擴(kuò)展了ThreadPoolExecutor類。

SheduledExecutorsService主要用于執(zhí)行周期性的或者定時(shí)的任務(wù)。其他情況下我們更多使用ThreadPoolExecutor。

ThreadPoolExecutor總共有七個(gè)構(gòu)造參數(shù)：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {

從其注釋和參數(shù)名不難猜測(cè)各個(gè)參數(shù)的用途。唯一有點(diǎn)麻煩的是corePoolSize, maximumPoolSize這兩個(gè)參數(shù)的區(qū)別。你可以參考這里或者這里。

但大多數(shù)情況我們并不需要直接調(diào)用它的構(gòu)造函數(shù)，在Executors里面定義了一系列的靜態(tài)方法供我們使用。包括newFixedThreadPool、newSingleThreadExecutor等。

由于ThreadPoolExecutor是一個(gè)通用的線程池，因此它需要為各種各樣的情況預(yù)留足夠的接口。ThreadPoolExecutor除了提供豐富的接口外，還提供了一些“什么都不做”的函數(shù)，為user預(yù)留接口。
比如每個(gè)任務(wù)在執(zhí)行之前會(huì)調(diào)用beforeExecute，執(zhí)行完畢后又會(huì)調(diào)用afterExecute。又比如terminate用來通知用戶代碼該線程將要結(jié)束。

這些接口java都提供了及其豐富的文檔。

Executor接口設(shè)計(jì)的目的或許也在于此，為簡(jiǎn)單的情況提供盡量簡(jiǎn)單的使用方法，同時(shí)為復(fù)雜的情況或者說高級(jí)用戶提供足夠多的接口。

在最初使用ThreadPoolExecutor 時(shí)候，用到FutrueTask的cancel接口，我總是擔(dān)心一個(gè)問題：

由于cancel是依賴線程的interrupt方法來實(shí)現(xiàn)的，也就是說cancel的狀態(tài)保持在線程中而不是task中。那么當(dāng)這個(gè)線程執(zhí)行下一個(gè)task會(huì)不會(huì)被影響？為了驗(yàn)證這一點(diǎn)，我做了個(gè)小小的實(shí)驗(yàn):

public class InterruptTest
{
    public static class MyTask implements Runnable {
        @Override
        public void run() {
            System.out.println(Thread.currentThread());
            System.out.println("before interrupt " + Thread.currentThread().isInterrupted());
            Thread.currentThread().interrupt();

            System.out.println("after interrupt " + Thread.currentThread().isInterrupted());
        }
    }
    public static void main(String[] str)
    {
        ExecutorService service = Executors.newFixedThreadPool(1);
        // MyTask task1 = new MyTask();
        Future future1 = service.submit(new InterruptTest.MyTask());
        Future future2 = service.submit(new InterruptTest.MyTask());
    }
}  

輸出結(jié)果說明，我的擔(dān)心是多余的：

Thread[pool-1-thread-1,5,main]
before interrupt false
after interrupt true
Thread[pool-1-thread-1,5,main]
before interrupt false
after interrupt true

其關(guān)鍵代碼就在ThreadPoolExecutor.runWorker 方法中，線程的中斷狀態(tài)會(huì)被清除(shutDown例外)。

final void runWorker(Worker w) {
     ...
                // 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();
    ...
}

參見 SO 的提問

其中Executors還有很多的東西，但是看看文章的長(zhǎng)度，我決定把那些關(guān)于Executors的筆記先“藏”起來。

如果感興趣的可以翻看源碼: ThreadFactory, RejectHandler, worker, task, shutDown策略，鎖機(jī)制... 看看ThreadPoolExecutor 把這些積木堆成一個(gè)房子的吧。