線程池源碼分析-FutureTask

時間 2019-12-08

標籤線程源碼分析 futuretask 简体版

原文原文鏈接

#1 系列目錄多線程

該系列打算從一個最簡單的Executor執行器開始一步一步擴展到ThreadPoolExecutor，但願能粗略的描述出線程池的各個實現細節。針對JDK1.7中的線程池異步

#2 Executor接口說明函數

Executor執行器，就是執行一個Runnable任務，可同步可異步，接口定義以下：源碼分析

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the <tt>Executor</tt> implementation.
     *
     * [@param](http://my.oschina.net/u/2303379) command the runnable task
     * [@throws](http://my.oschina.net/throws) RejectedExecutionException if this task cannot be
     * accepted for execution.
     * [@throws](http://my.oschina.net/throws) NullPointerException if command is null
     */
    void execute(Runnable command);
}

ExecutorService則繼承了Executor，描述了線程池應該具備的功能特性，來詳細看下接口，這些接口都有詳細的文檔能夠閱讀，這裏就再也不列出來了，目前只說明咱們重點關注的接口。this

<T> Future<T> submit(Callable<T> task);

能夠提交一個Callable，而且返回一個Future用於追蹤提交的任務。如何追蹤一個任務的狀態和返回數據呢？那就須要將提交的任務進行封裝，對任務的執行、執行過程當中的異常、中斷、返回結果進行統一的監控處理。下面就來看看AbstractExecutorService對上述submit的實現.net

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

從上面看到就是對Callable封裝成一個新的任務，即FutureTask，調用Executor的原始接口execute方法來執行FutureTask，而且返回給用戶FutureTask對象，用於追蹤任務的狀態和數據，下面就須要咱們來詳細看看FutureTask如何對任務進行封裝的線程

#3 FutureTask的實現細節設計

##3.1 FutureTask的屬性和構造函數code

private volatile int state;
private static final int NEW          = 0;
private static final int COMPLETING   = 1;
private static final int NORMAL       = 2;
private static final int EXCEPTIONAL  = 3;
private static final int CANCELLED    = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED  = 6;

/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
/** The thread running the callable; CASed during run() */
private volatile Thread runner;
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}

有一個狀態變量state,一個Callable callable即原始任務，Object outcome存放原始任務的輸出結果或者異常，Thread runner運行該任務的線程，WaitNode waiters等待獲取任務結果的等待者對象

##3.2 FutureTask的get方法實現

使用FutureTask阻塞式等待任務執行結果，一種是永遠阻塞另外一種就是阻塞必定時間不然報超時異常，以下2個方法

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

阻塞式等待的核心邏輯就在上述awaitDone方法中，來詳細看看

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) {
            if (q != null)
                q.thread = null;
            return s;
        }
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        else if (q == null)
            q = new WaitNode();
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        }
        else
            LockSupport.park(this);
    }
}

能夠看到有一個for循環不斷處理着各類狀況：

1 從最開始的WaitNode q = null，構建了一個WaitNode，即表明着當前線程做爲一個等待者，WaitNode就是一個簡單的鏈表，以下

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

2 構建好WaitNode以後就要將該WaitNode放入鏈表中，這時候就會涉及多線程問題，使用UNSAFE的CAS來解決，這種方式也是AtomicLong等衆多原子類的底層實現方式

3 成功放入WaitNode鏈表以後，採用LockSupport的park阻塞當前線程，要麼只阻塞必定時間要麼一直阻塞，直到被LockSupport的unpark喚醒。LockSupport在鎖的底層實現AQS中也很是常見，使用了LockSupport就能夠不用在for循環裏不斷判斷當前任務狀態而浪費CPU，只須要當前任務完成以後，使用LockSupport對等待線程進行unpark，就可使等待的線程退出等待繼續往下執行

4 若是LockSupport阻塞時間到了，還未收到unpark，則須要從等待者鏈表中刪除當前線程表明的等待者

##3.3 FutureTask的任務執行過程

public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

1 一旦FutureTask任務開始執行了，就須要將當前執行線程設置到FutureTask的volatile Thread runner屬性中

2 執行原始任務Callable的call方法，可能成功也可能失敗也可能被中斷被取消

文檔中有以下狀態的遷移過程：

Possible state transitions:
 * NEW -> COMPLETING -> NORMAL
 * NEW -> COMPLETING -> EXCEPTIONAL
 * NEW -> CANCELLED
 * NEW -> INTERRUPTING -> INTERRUPTED

來看下成功和失敗方法

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();
    }
}

都是首先將狀態變成COMPLETING正在結束中，而後設置outcome，成功則設置正常的返回值，失敗則設置成異常，而後根據劃定最終的狀態結果，成功就是NORMAL，失敗就是EXCEPTIONAL，最後呢調用finishCompletion，去unpark以前說的WaitNode中對應的線程們

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }

    done();

    callable = null;        // to reduce footprint
}

這裏就是遍歷WaitNode鏈表，對每個WaitNode對應的線程依次進行LockSupport.unpark(t)，使其結束阻塞。WaitNode通知完畢後，調用一個done方法，目前該方法是空的實現，因此你若是想在任務完成後執行一些動做的時候就能夠重寫該方法

有一個問題就是：爲何必定要加入COMPLETING狀態呢？能不能直接過分到NORMAL或者EXCEPTIONAL？

目前個人理解是：NORMAL或者EXCEPTIONAL是一種最終狀態，因此在出現該狀態前，outcome必須已經被設置了，即有以下代碼：

protected void set(V v) {
	outcome = v;
	UNSAFE.compareAndSwapInt(this, stateOffset, NEW, NORMAL)
    finishCompletion();
}

可是由於存在外部直接取消該任務，因此結果狀態的設置和outcome必須是同步的，且outcome在前，爲了保證代碼的同步可使用鎖

protected void set(V v) {
	synchronized(){
		outcome = v;
		UNSAFE.compareAndSwapInt(this, stateOffset, NEW, NORMAL)
        finishCompletion();
	}
}

爲了減小鎖帶來的開支，就能夠引入一箇中間狀態COMPLETING，經過CAS來間接實現鎖的競爭，同時又保證outcome在最終狀態NORMAL或者EXCEPTIONAL以前被設置

##3.4 FutureTask任務的取消

public boolean cancel(boolean mayInterruptIfRunning) {
    if (state != NEW)
        return false;
    if (mayInterruptIfRunning) {
        if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
            return false;
        Thread t = runner;
        if (t != null)
            t.interrupt();
        UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
    }
    else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
        return false;
    finishCompletion();
    return true;
}

取消任務，有2種狀況，一種該任務正在運行，一種就是非運行狀態，因此須要用戶給出明示是否中斷正在運行的任務，即須要一個參數mayInterruptIfRunning

中斷任務就是經過中斷運行該任務的線程，即直接調用該線程的interrupt()方法

#4 結束語

FutureTask大部分就簡單分析完了，其餘的本身去看下就好了。至此咱們瞭解了一個任務被提交通過了封裝，變成了一個新的任務FutureTask,同時FutureTask也明確了該任務的整個執行過程，只留出核心execute(futureTask)方法須要被子類來實現，下一篇文章就重點介紹下ThreadPoolExecutor對該核心方法的實現

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。