天天看點
傳回

【精通高并發系列】兩種異步模型與深度解析Future接口！

大家好，我是冰河~~

本文有點長，但是滿滿的幹貨，以實際案例的形式分析了兩種異步模型，并從源碼角度深度解析Future接口和FutureTask類，希望大家踏下心來，打開你的IDE，跟着文章看源碼，相信你一定收獲不小！

一、兩種異步模型

在Java的并發程式設計中，大體上會分為兩種異步程式設計模型，一類是直接以異步的形式來并行運作其他的任務，不需要傳回任務的結果資料。一類是以異步的形式運作其他任務，需要傳回結果。

1.無傳回結果的異步模型

無傳回結果的異步任務，可以直接将任務丢進線程或線程池中運作，此時，無法直接獲得任務的執行結果資料，一種方式是可以使用回調方法來擷取任務的運作結果。

具體的方案是：定義一個回調接口，并在接口中定義接收任務結果資料的方法，具體邏輯在回調接口的實作類中完成。将回調接口與任務參數一同放進線程或線程池中運作，任務運作後調用接口方法，執行回調接口實作類中的邏輯來處理結果資料。這裡，給出一個簡單的示例供參考。

  • 定義回調接口 
package io.binghe.concurrent.lab04;

/**
 * @author binghe
 * @version 1.0.0
 * @description 定義回調接口
 */
public interface TaskCallable<T> {
    T callable(T t);
}

便于接口的通用型，這裡為回調接口定義了泛型。

  • 定義任務結果資料的封裝類 
package io.binghe.concurrent.lab04;

import java.io.Serializable;

/**
 * @author binghe
 * @version 1.0.0
 * @description 任務執行結果
 */
public class TaskResult implements Serializable {
    private static final long serialVersionUID = 8678277072402730062L;
    /**
     * 任務狀态
     */
    private Integer taskStatus;

    /**
     * 任務消息
     */
    private String taskMessage;

    /**
     * 任務結果資料
     */
    private String taskResult;
    
    //省略getter和setter方法
    @Override
    public String toString() {
        return "TaskResult{" +
                "taskStatus=" + taskStatus +
                ", taskMessage='" + taskMessage + '\'' +
                ", taskResult='" + taskResult + '\'' +
                '}';
    }
}
  • 建立回調接口的實作類

回調接口的實作類主要用來對任務的傳回結果進行相應的業務處理，這裡，為了友善示範，隻是将結果資料傳回。大家需要根據具體的業務場景來做相應的分析和處理。

package io.binghe.concurrent.lab04;

/**
 * @author binghe
 * @version 1.0.0
 * @description 回調函數的實作類
 */
public class TaskHandler implements TaskCallable<TaskResult> {
    @Override
public TaskResult callable(TaskResult taskResult) {
//TODO 拿到結果資料後進一步處理
    System.out.println(taskResult.toString());
        return taskResult;
    }
}
  • 建立任務的執行類

任務的執行類是具體執行任務的類，實作Runnable接口，在此類中定義一個回調接口類型的成員變量和一個String類型的任務參數（模拟任務的參數），并在構造方法中注入回調接口和任務參數。在run方法中執行任務，任務完成後将任務的結果資料封裝成TaskResult對象，調用回調接口的方法将TaskResult對象傳遞到回調方法中。

package io.binghe.concurrent.lab04;

/**
 * @author binghe
 * @version 1.0.0
 * @description 任務執行類
 */
public class TaskExecutor implements Runnable{
    private TaskCallable<TaskResult> taskCallable;
    private String taskParameter;

    public TaskExecutor(TaskCallable<TaskResult> taskCallable, String taskParameter){
        this.taskCallable = taskCallable;
        this.taskParameter = taskParameter;
    }

    @Override
    public void run() {
        //TODO 一系列業務邏輯,将結果資料封裝成TaskResult對象并傳回
        TaskResult result = new TaskResult();
        result.setTaskStatus(1);
        result.setTaskMessage(this.taskParameter);
        result.setTaskResult("異步回調成功");
        taskCallable.callable(result);
    }
}

到這裡，整個大的架構算是完成了，接下來，就是測試看能否擷取到異步任務的結果了。

  • 異步任務測試類 
package io.binghe.concurrent.lab04;

/**
 * @author binghe
 * @version 1.0.0
 * @description 測試回調
 */
public class TaskCallableTest {
    public static void main(String[] args){
        TaskCallable<TaskResult> taskCallable = new TaskHandler();
        TaskExecutor taskExecutor = new TaskExecutor(taskCallable, "測試回調任務");
        new Thread(taskExecutor).start();
    }
}

在測試類中，使用Thread類建立一個新的線程，并啟動線程運作任務。運作程式最終的接口資料如下所示。

TaskResult{taskStatus=1, taskMessage='測試回調任務', taskResult='異步回調成功'}

大家可以細細品味下這種擷取異步結果的方式。這裡，隻是簡單的使用了Thread類來建立并啟動線程，也可以使用線程池的方式實作。大家可自行實作以線程池的方式通過回調接口擷取異步結果。

2.有傳回結果的異步模型

盡管使用回調接口能夠擷取異步任務的結果，但是這種方式使用起來略顯複雜。在JDK中提供了可以直接傳回異步結果的處理方案。最常用的就是使用Future接口或者其實作類FutureTask來接收任務的傳回結果。

  • 使用Future接口擷取異步結果

使用Future接口往往配合線程池來擷取異步執行結果，如下所示。

package io.binghe.concurrent.lab04;

import java.util.concurrent.*;

/**
 * @author binghe
 * @version 1.0.0
 * @description 測試Future擷取異步結果
 */
public class FutureTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future<String> future = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return "測試Future擷取異步結果";
            }
        });
        System.out.println(future.get());
        executorService.shutdown();
    }
}

運作結果如下所示。

測試Future擷取異步結果
  • 使用FutureTask類擷取異步結果

FutureTask類既可以結合Thread類使用也可以結合線程池使用，接下來，就看下這兩種使用方式。

結合Thread類的使用示例如下所示。

package io.binghe.concurrent.lab04;

import java.util.concurrent.*;

/**
 * @author binghe
 * @version 1.0.0
 * @description 測試FutureTask擷取異步結果
 */
public class FutureTaskTest {

    public static void main(String[] args)throws ExecutionException, InterruptedException{
        FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return "測試FutureTask擷取異步結果";
            }
        });
        new Thread(futureTask).start();
        System.out.println(futureTask.get());
    }
}           
測試FutureTask擷取異步結果

結合線程池的使用示例如下。

package io.binghe.concurrent.lab04;

import java.util.concurrent.*;

/**
 * @author binghe
 * @version 1.0.0
 * @description 測試FutureTask擷取異步結果
 */
public class FutureTaskTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return "測試FutureTask擷取異步結果";
            }
        });
        executorService.execute(futureTask);
        System.out.println(futureTask.get());
        executorService.shutdown();
    }
}           
測試FutureTask擷取異步結果

可以看到使用Future接口或者FutureTask類來擷取異步結果比使用回調接口擷取異步結果簡單多了。注意：實作異步的方式很多，這裡隻是用多線程舉例。

接下來，就深入分析下Future接口。

二、深度解析Future接口

1.Future接口

Future是JDK1.5新增的異步程式設計接口，其源代碼如下所示。

package java.util.concurrent;

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

可以看到，在Future接口中，總共定義了5個抽象方法。接下來，就分别介紹下這5個方法的含義。

  • cancel(boolean)

取消任務的執行，接收一個boolean類型的參數，成功取消任務，則傳回true，否則傳回false。當任務已經完成，已經結束或者因其他原因不能取消時，方法會傳回false，表示任務取消失敗。當任務未啟動調用了此方法，并且結果傳回true（取消成功），則目前任務不再運作。如果任務已經啟動，會根據目前傳遞的boolean類型的參數來決定是否中斷目前運作的線程來取消目前運作的任務。

  • isCancelled()

判斷任務在完成之前是否被取消，如果在任務完成之前被取消，則傳回true；否則，傳回false。

這裡需要注意一個細節：隻有任務未啟動，或者在完成之前被取消，才會傳回true，表示任務已經被成功取消。其他情況都會傳回false。

  • isDone()

判斷任務是否已經完成，如果任務正常結束、抛出異常退出、被取消，都會傳回true，表示任務已經完成。

  • get()

當任務完成時，直接傳回任務的結果資料；當任務未完成時，等待任務完成并傳回任務的結果資料。

  • get(long, TimeUnit)

當任務完成時，直接傳回任務的結果資料；當任務未完成時，等待任務完成，并設定了逾時等待時間。在逾時時間内任務完成，則傳回結果；否則，抛出TimeoutException異常。

2.RunnableFuture接口

Future接口有一個重要的子接口，那就是RunnableFuture接口，RunnableFuture接口不但繼承了Future接口，而且繼承了java.lang.Runnable接口，其源代碼如下所示。

package java.util.concurrent;

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

這裡，問一下，RunnableFuture接口中有幾個抽象方法？想好了再說！哈哈哈。。。

這個接口比較簡單run()方法就是運作任務時調用的方法。

3.FutureTask類

FutureTask類是RunnableFuture接口的一個非常重要的實作類，它實作了RunnableFuture接口、Future接口和Runnable接口的所有方法。FutureTask類的源代碼比較多，這個就不粘貼了，大家自行到java.util.concurrent下檢視。

（1）FutureTask類中的變量與常量

在FutureTask類中首先定義了一個狀态變量state，這個變量使用了volatile關鍵字修飾，這裡，大家隻需要知道volatile關鍵字通過記憶體屏障和禁止重排序優化來實作線程安全，後續會單獨深度分析volatile關鍵字是如何保證線程安全的。緊接着，定義了幾個任務運作時的狀态常量，如下所示。

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;

其中，代碼注釋中給出了幾個可能的狀态變更流程，如下所示。

NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED

接下來，定義了其他幾個成員變量，如下所示。

private Callable<V> callable;
private Object outcome; 
private volatile Thread runner;
private volatile WaitNode waiters;

又看到我們所熟悉的Callable接口了，Callable接口那肯定就是用來調用call()方法執行具體任務了。

  • outcome：Object類型，表示通過get()方法擷取到的結果資料或者異常資訊。
  • runner：運作Callable的線程，運作期間會使用CAS保證線程安全，這裡大家隻需要知道CAS是Java保證線程安全的一種方式，後續文章中會深度分析CAS如何保證線程安全。
  • waiters：WaitNode類型的變量，表示等待線程的堆棧，在FutureTask的實作中，會通過CAS結合此堆棧交換任務的運作狀态。

看一下WaitNode類的定義，如下所示。

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

可以看到，WaitNode類是FutureTask類的靜态内部類，類中定義了一個Thread成員變量和指向下一個WaitNode節點的引用。其中通過構造方法将thread變量設定為目前線程。

（2）構造方法

接下來，是FutureTask的兩個構造方法，比較簡單，如下所示。

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

public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;
}

（3）是否取消與完成方法

繼續向下看源碼，看到一個任務是否取消的方法，和一個任務是否完成的方法，如下所示。

public boolean isCancelled() {
    return state >= CANCELLED;
}

public boolean isDone() {
    return state != NEW;
}

這兩方法中，都是通過判斷任務的狀态來判定任務是否已取消和已完成的。為啥會這樣判斷呢？再次檢視FutureTask類中定義的狀态常量發現，其常量的定義是有規律的，并不是随意定義的。其中，大于或者等于CANCELLED的常量為CANCELLED、INTERRUPTING和INTERRUPTED，這三個狀态均可以表示線程已經被取消。當狀态不等于NEW時，可以表示任務已經完成。

通過這裡，大家可以學到一點：以後在編碼過程中，要按照規律來定義自己使用的狀态，尤其是涉及到業務中有頻繁的狀态變更的操作，有規律的狀态可使業務處理變得事半功倍，這也是通過看别人的源碼設計能夠學到的，這裡，建議大家還是多看别人寫的優秀的開源架構的源碼。

（4）取消方法

我們繼續向下看源碼，接下來，看到的是cancel(boolean)方法，如下所示。

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

接下來，拆解cancel(boolean)方法。在cancel(boolean)方法中，首先判斷任務的狀态和CAS的操作結果，如果任務的狀态不等于NEW或者CAS的操作傳回false，則直接傳回false，表示任務取消失敗。如下所示。

if (!(state == NEW &&
      UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
          mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
    return false;

接下來，在try代碼塊中，首先判斷是否可以中斷目前任務所在的線程來取消任務的運作。如果可以中斷目前任務所在的線程，則以一個Thread臨時變量來指向運作任務的線程，當指向的變量不為空時，調用線程對象的interrupt()方法來中斷線程的運作，最後将線程标記為被中斷的狀态。如下所示。

try {
    if (mayInterruptIfRunning) {
        try {
            Thread t = runner;
            if (t != null)
                t.interrupt();
        } finally { // final state
            UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
        }
    }
}

這裡，發現變更任務狀态使用的是UNSAFE.putOrderedInt()方法，這個方法是個什麼鬼呢？點進去看一下，如下所示。

public native void putOrderedInt(Object var1, long var2, int var4);

可以看到，又是一個本地方法，嘿嘿，這裡先不管它，後續文章會詳解這些方法的作用。

接下來，cancel(boolean)方法會進入finally代碼塊，如下所示。

finally {
    finishCompletion();
}

可以看到在finallly代碼塊中調用了finishCompletion()方法，顧名思義，finishCompletion()方法表示結束任務的運作，接下來看看它是如何實作的。點到finishCompletion()方法中看一下，如下所示。

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
}

在finishCompletion()方法中，首先定義一個for循環，循環終止因子為waiters為null，在循環中，判斷CAS操作是否成功，如果成功進行if條件中的邏輯。首先，定義一個for自旋循環，在自旋循環體中，喚醒WaitNode堆棧中的線程，使其運作完成。當WaitNode堆棧中的線程運作完成後，通過break退出外層for循環。接下來調用done()方法。done()方法又是個什麼鬼呢？點進去看一下，如下所示。

protected void done() { }

可以看到，done()方法是一個空的方法體，交由子類來實作具體的業務邏輯。

當我們的具體業務中，需要在取消任務時，執行一些額外的業務邏輯，可以在子類中覆寫done()方法的實作。

（5）get()方法

繼續向下看FutureTask類的代碼，FutureTask類中實作了兩個get()方法，如下所示。

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

沒參數的get()方法為當任務未運作完成時，會阻塞，直到傳回任務結果。有參數的get()方法為當任務未運作完成，并且等待時間超出了逾時時間，會TimeoutException異常。

兩個get()方法的主要邏輯差不多，一個沒有逾時設定，一個有逾時設定，這裡說一下主要邏輯。判斷任務的目前狀态是否小于或者等于COMPLETING，也就是說，任務是NEW狀态或者COMPLETING，調用awaitDone()方法，看下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);
    }
}

接下來，拆解awaitDone()方法。在awaitDone()方法中，最重要的就是for自旋循環，在循環中首先判斷目前線程是否被中斷，如果已經被中斷，則調用removeWaiter()将目前線程從堆棧中移除，并且抛出InterruptedException異常，如下所示。

if (Thread.interrupted()) {
    removeWaiter(q);
    throw new InterruptedException();
}

接下來，判斷任務的目前狀态是否完成，如果完成，并且堆棧句柄不為空，則将堆棧中的目前線程設定為空，傳回目前任務的狀态，如下所示。

int s = state;
if (s > COMPLETING) {
    if (q != null)
        q.thread = null;
    return s;
}

當任務的狀态為COMPLETING時，使目前線程讓出CPU資源，如下所示。

else if (s == COMPLETING)
    Thread.yield();

如果堆棧為空，則建立堆棧對象，如下所示。

else if (q == null)
    q = new WaitNode();

如果queued變量為false，通過CAS操作為queued指派，如果awaitDone()方法傳遞的timed參數為true，則計算逾時時間，當時間已逾時，則在堆棧中移除目前線程并傳回任務狀态，如下所示。如果未逾時，則重置逾時時間，如下所示。

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

接下來，回到get()方法中，當awaitDone()方法傳回結果，或者任務的狀态不滿足條件時，都會調用report()方法，并将目前任務的狀态傳遞到report()方法中，并傳回結果，如下所示。

return report(s);

看來，這裡還要看下report()方法啊，點進去看下report()方法的實作，如下所示。

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

可以看到，report()方法的實作比較簡單，首先，将outcome資料指派給x變量，接下來，主要是判斷接收到的任務狀态，如果狀态為NORMAL，則将x強轉為泛型類型傳回；當任務的狀态大于或者等于CANCELLED，也就是任務已經取消，則抛出CancellationException異常，其他情況則抛出ExecutionException異常。

至此，get()方法分析完成。注意：一定要了解get()方法的實作，因為get()方法是我們使用Future接口和FutureTask類時，使用的比較頻繁的一個方法。

（6）set()方法與setException()方法

繼續看FutureTask類的代碼，接下來看到的是set()方法與setException()方法，如下所示。

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

通過源碼可以看出，set()方法與setException()方法整體邏輯幾乎一樣，隻是在設定任務狀态時一個将狀态設定為NORMAL，一個将狀态設定為EXCEPTIONAL。

至于finishCompletion()方法，前面已經分析過。

（7）run()方法與runAndReset()方法

接下來，就是run()方法了，run()方法的源代碼如下所示。

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

可以這麼說，隻要使用了Future和FutureTask，就必然會調用run()方法來運作任務，掌握run()方法的流程是非常有必要的。在run()方法中，如果目前狀态不是NEW，或者CAS操作傳回的結果為false，則直接傳回，不再執行後續邏輯，如下所示。

if (state != NEW ||
    !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
    return;

接下來，在try代碼塊中，将成員變量callable指派給一個臨時變量c，判斷臨時變量不等于null，并且任務狀态為NEW，則調用Callable接口的call()方法，并接收結果資料。并将ran變量設定為true。當程式抛出異常時，将接收結果的變量設定為null，ran變量設定為false，并且調用setException()方法将任務的狀态設定為EXCEPTIONA。接下來，如果ran變量為true，則調用set()方法，如下所示。

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代碼塊中，如下所示。

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

這裡，将runner設定為null，如果任務的目前狀态大于或者等于INTERRUPTING，也就是線程被中斷了。則調用handlePossibleCancellationInterrupt()方法，接下來，看下handlePossibleCancellationInterrupt()方法的實作。

private void handlePossibleCancellationInterrupt(int s) {
    if (s == INTERRUPTING)
        while (state == INTERRUPTING)
            Thread.yield();
}

可以看到，handlePossibleCancellationInterrupt()方法的實作比較簡單，當任務的狀态為INTERRUPTING時，使用while()循環，條件為目前任務狀态為INTERRUPTING，将目前線程占用的CPU資源釋放，也就是說，當任務運作完成後，釋放線程所占用的資源。

runAndReset()方法的邏輯與run()差不多，隻是runAndReset()方法會在finally代碼塊中将任務狀态重置為NEW。runAndReset()方法的源代碼如下所示，就不重複說明了。

protected boolean runAndReset() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
        return false;
    boolean ran = false;
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                c.call(); // don't set result
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
            }
        }
    } 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
        s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
    return ran && s == NEW;
}

（8）removeWaiter()方法

removeWaiter()方法中主要是使用自旋循環的方式來移除WaitNode中的線程，比較簡單，如下所示。

private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;
        retry:
        for (;;) {          // restart on removeWaiter race
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                s = q.next;
                if (q.thread != null)
                    pred = q;
                else if (pred != null) {
                    pred.next = s;
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                      q, s))
                    continue retry;
            }
            break;
        }
    }
}

最後，在FutureTask類的最後，有如下代碼。

// Unsafe mechanics
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);
    }
}

關于這些代碼的作用，會在後續深度解析CAS文章中詳細說明，這裡就不再探讨。

至此，關于Future接口和FutureTask類的源碼就分析完了。

好了，今天就到這兒吧，我是冰河，我們下期見~~

安全 ide java 開發工具 java開發工具 jdk工具java 工具支援java 釋出java安全 java工具
上一篇: 每秒處理107萬張圖檔！阿裡雲打破MLPerf推理性能測試紀錄
下一篇: 再續前緣 阿裡雲基礎設施網絡團隊ACM SIGCOMM 2021續創佳績

繼續閱讀