結構化併發應用程式-Executor框架的理解與使用

public class FutureTask<V> implements RunnableFuture<V> {
    /*
     * FutureTask是一個併發安全類,有併發控制機制
     * 先前版本為了簡潔,使用AQS來實現,jdk1.8後的併發控制使用 一個state 域,通過CAS操作state,同時通過一個簡單的stack來維護 waiting threads*/
    private volatile int state;
    // 實際的任務
    private Callable<V> callable;

    // 任務的執行結果或者拋出的異常,通過get()獲取這個欄位
    // 不需要 volatile,不會有併發讀寫該欄位的情況
    private Object outcome;
    // 在run()的時候 會set runner
    private volatile Thread runner;

    // 一個簡單的等待隊列,為什麼不用AQS了? AQS太重了
    private volatile WaitNode waiters;
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;

        WaitNode() {
            thread = Thread.currentThread();
        }
    }
}

public void run() {
        // 先驗條件
        if (state != NEW ||
        // set runner
                !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 outcome
                    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);
        }
    }
 
 public boolean cancel(boolean mayInterruptIfRunning) {
        // cancel: new -> interrupting->interrupted,或者 new -> cancelled
        if (!(state == NEW &&
                UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                        mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            // check then action, 複合操作, 如果失敗則說明此時任務的狀態不是 new了,返回false,即取消失敗
            return false;
        try { // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    // 通過給 執行該任務的線程 發送中斷信號來取消任務
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    // 發送完後預設置為 interrupted, 表示 信號已發過去了,但任務不一定能停下來,需要任務自己判斷這個信號
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }
 
 public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            // 任務未完成前,通過LockSupport.park等待任務完成
            s = awaitDone(false, 0L);
        return report(s);
    }
 
    @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL)
            return (V) x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable) x);
    }
 
/**
     * 可中斷的方法
     * Awaits completion or aborts on interrupt or timeout.
     *
     * @param timed true if use timed waits
     * @param nanos time to wait, if timed
     * @return state upon completion
     */
    private int awaitDone(boolean timed, long nanos)
            throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
        // 可中斷方法的大部分實現,都是通過拋InterruptedException()來響應中斷,註意Thread.interrupted()會清除中斷信號
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }
 
            int s = state;
            // 如果任務 達到了終態,即isDone()了, 即 S>COMPLETIOG,返回 isDone() => s>competing
            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)
                // 任務還未開始, 即 s=NEW 時,此時創建 等待線程節點,再過一次前面的操作 到下一步
                q = new WaitNode();
            else if (!queued)
                // 新增的節點未入隊,將節點入隊,入隊成功後再過一次前面的操作 到下一步
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                        q.next = waiters, q);
            else if (timed) {
                // 如果過了等待時間了,不等了, 把前面構建的節點從等待隊列中刪除,返回 state
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                // 否則 限時阻塞當前線程,等待任務完成時被喚醒
                LockSupport.parkNanos(this, nanos);
            } else
                // timed=false時, 永遠阻塞當前線程,等待任務完成時被喚醒
                LockSupport.park(this);
        }
    }

**
     * 實際的工作者線程,主要維護線程的中斷狀態
     * 這個類為了簡化在運行任務的時候對鎖的獲取和釋放,設計成了 extends AQS
     * 當shutdown的時候,會通過tryLock判斷線程是否正在執行任務,如果為false,表示線程不在執行任務,而是在等待新的任務,通過發送中斷信號,中斷這些線程的等待,這些線程被中斷後會判斷是由於什麼原因喚醒的,如果這個時候線程池狀態為SHUTDOWN,那麼這些線程就會被回收.
     * 註：(線程被喚醒的原因可能是被中斷了,也有可能是有任務了,也有可能是時間到了,喚醒後需要二次判斷go getTask())
     * 註:(線程由於沒有任務掛起(poll()), 掛起期間可能有新的任務過來了(offer())被喚醒,也有可能被中斷信號通知關閉而被喚醒.)
     */
    private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
        /**
         * Worker繼承了AQS,也就是說Worker還有一個state屬性欄位,這個欄位是有必要分析下的:
         * -1: 剛初始化 
         * 0: 剛調用runWorker或者沒任務了 
         * 1: 正在執行任務,正是通過這個state欄位,來判斷線程是否正在執行任務(tryLock)
         */
        final Thread thread;
        // 在Worker初始化時,firstTask可能有值
        Runnable firstTask;
        // 每個工作者線程完成的任務數,任務性質可以不同,即線程是可以復用的
        volatile long completedTasks;
        Worker(Runnable firstTask) {
        // 只有在worker線程已開始的時候中斷才有意義,所以在初始化worker的時候state=-1,這個時候不會被中斷go isLocked()
            setState(-1);
            this.firstTask = firstTask;
            // 初始化Workder的時候 通過 threadFactory創建線程,最終通過系統調用,由OS創建內核線程
            this.thread = getThreadFactory().newThread(this);
        }
        // runWorker實際實現主執行迴圈,接下來就是重點了,任務線程初始化時,拿到了firstTask(有的話),以及一個新的線程,接下來就開始真正地執行任務了
        public void run() {
            runWorker(this);
        }
        // 設計worker類的主要目的,用來中斷線程
        void interruptIfStarted() {
            Thread t;
            // 只有在worker線程已開始的時候中斷才有意義, 所以在初始化worker的時候state=-1,這個時候不會被中斷
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        // 線程池小於core
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 新增任務失敗,可能在addWorker的時候線程數達到了corePoolSize的水平,此時放到workQueue
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            // 判斷如果線程池正在shutdown,拒絕任務
            if (!isRunning(recheck) && remove(command))
                reject(command);
            // 確保任務隊列中的任務可以被線程執行
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
            // 工作隊列滿了,再嘗試增加worker,線程個數判斷使用 maxvalue
        } else if (!addWorker(command, false))
            reject(command);
    }

private boolean addWorker(Runnable firstTask, boolean core) {
        retry: for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                    !(rs == SHUTDOWN &&
                            firstTask == null &&
                            !workQueue.isEmpty()))
                // 當shutdown且 隊列是空的時候就沒必要加worker了
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                // 達到限制數量了也返回false
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get(); // Re-read ctl
                if (runStateOf(c) != rs)
                    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) {
                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 rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        // 非shutdown,或者 是shutdown但是firstTask==null的時候,可以新增線程
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        // 新增worker的時候更新largestPoolSize
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    // start() -> runWorker()->task.run()
                    // 新增成功後 調用start(),如果start()失敗了,比如ntive stack申請失敗,也返回false
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (!workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        // state: -1 => 0 , unlock -> release -> tryRelease -> state=0
        // 這個時候任務線程開始工作,可以被中斷
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            // getTask從隊列中拿任務
            while (task != null || (task = getTask()) != null) {
       // 工作的時候將state置為1,表示正在工作,這個操作一定會成功(正常來說lock是一個基於狀態的方法,可能會阻塞調用線程),因為不會有其他地方調用w.lock
                // 註:(state: 0 => 1 lock -> acquire -> tryAcquire -> state=1)
                w.lock();
                // 線程當且僅當池子stopping(shutdown,shutdownNow的時候)的時候才會interrupted,且一定要interrupted
                // 註:(worker線程是由線程池服務來維護的,只有線程池服務有權對worker線程進行中斷操作)
                if ((runStateAtLeast(ctl.get(), STOP) ||
                // 註:(Thread.interrupted會清除interrupted標記)
                // 這裡表明worker線程只能在STOPING(STOP,TIDING,TERMINATED)時中斷信號有效,其他形式的中斷信號（例如在任務中中斷)會被清除
                        (Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP)))
                        &&
                        !wt.isInterrupted())
                    wt.interrupt();
                try {
                    // hooc 函數
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                  // 保存異常 thrown 到1326處理(給客戶端提供的鉤子函數,afterExecute,使客戶端可以感知到任務失敗併進行特定的處理),同時拋出異常到
                        // 1330 處理(線程池自身對任務異常的處理)
                        thrown = x;
                        throw x;
                    } catch (Error x) {
                        thrown = x;
                        throw x;
                    } catch (Throwable x) {
                        thrown = x;
                        throw new Error(x);
                    } finally {
                        // 將任務執行過程中的異常傳入到hooc函數
                        afterExecute(task, thrown);
                    }
                } finally {
       // beforeExecutehooc函數出錯或者任務出錯了的話,task=null,從而跳到1336,completedAbruptly=true,從而回收線程,即使線程並沒有完成任何工作
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            // 處理task=null的場景或者任務拋處異常時的場景,釋放線程,什麼時候task會為null ,go getTask
            processWorkerExit(w, completedAbruptly);
        }
    }

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

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

            // Check if queue empty only if necessary.
            // stopped 或者 shutdown 且 workQueue.isEmpty 返回null 2,3
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            // allowCoreThreadTimeOut等價於 wc> corePooSize
            // allowCoreThreadTimeOut, wc>corePoolSize, 一起表示 當任務線程獲取任務超時時,被要求中斷(subject to termination)
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            // 1 wc > maxPoolSize 或者 4 獲取任務超時且 要求獲取任務超時的進程被中斷(timed && timedOut)
            if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try {
                // 如果沒有任務,則阻塞在這裡, workQueue.offer後繼續運行
                Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
                if (r != null)
                    return r;
                // r==null,poll返回null,表示timedOut,下次 go 1210
                timedOut = true;
            } catch (InterruptedException retry) {
                // 忽略中斷信號
                timedOut = false;
            }
        }
    }

private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            // getTask時會decrementWorkerCount
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 回收線程前先將線程執行的任務數加一下
            completedTaskCount += w.completedTasks;
            // 通過釋放worker引用來釋放線程
            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;
                // 如果不是由於任務忽然中斷且線程數符合最小值的要求,那麼無需addWorker替換
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            // 如果任務線程是由於任務執行異常退出的 或者 線程池中的數量小於min,addWorker
            addWorker(null, false);
        }
    }

public static void main(String[] args) {
        Thread calPrimeTask = new Thread(InterruptedTest::calPrime);
        calPrimeTask.start();
        ThreadUtil.sleep(1000);
        // 嘗試終止calPrimeTask
        calPrimeTask.interrupt();
    }


    public static void calPrime() {
        while(!Thread.currentThread().isInterrupted()) {
            ThreadUtil.sleep(50);
            log();
            System.out.println("一個耗時50ms的任務完成!");
        }
    }

    public static void log() {
        /**
         * 假設有一段代碼調用了jdk中的某個可能拋出InterruptedException的介面,這段代碼捕獲到這個異常後本意是不會處理這個異常,但是如果它沒有再
         * Thread.currentThread().interrupt(),就會影響其他使用到這個方法的函數,例如calPrime();
         */
        ArrayBlockingQueue<Integer> que = new ArrayBlockingQueue<>(12);
        try {
            System.out.println("do other thing");
            que.poll(30, TimeUnit.MILLISECONDS);
        }catch (InterruptedException e) {
            e.printStackTrace();
            // poll拋出 InterruptedException後,會清空 interrupted標記,這裡返回false
            System.out.println(Thread.currentThread().isInterrupted());
          // 如果這裡不重新設置interrupted標記的話,這回使得calPrimary任務無法取消,我們不知道調用棧的其他地方是否會用到中斷信號,所以必須把中斷信號設置回去
           Thread.currentThread().interrupt();
        }
    }


// 不支持取消但仍可以調用可中斷阻塞方法，忽略中斷信號並重試
public Task getNextTask(BlockingQueue<Task> queue) {
    boolean interrupted = false;
    try {
        while(true) {
            try {
                return queue.take();
            }catch(InterruptedException e) {
                interrupted = true;
                // 忽略並重試
                // 如果我們在這裡調用Thread.currentThread().interrupt()的話會引起死迴圈
            }
        }finally {
            if(interrupted) {
                // 避免屏蔽中斷信號，其他方法可能需要
                Thread.currentThread().interrupt();
            }
        }
    }
}