Android Handler機(jī)制12之Callable、Future和FutureTask

Android Handler機(jī)制系列文章整體內(nèi)容如下：

本片文章的主要內(nèi)容如下：

  • 1、概述
  • 2、Callable
  • 3、Future
  • 4、FutureTask
  • 5、總結(jié)

一、概述

  • Java從發(fā)布的一個(gè)版本開始就可以很方便地編寫多線程的應(yīng)用程序，并在設(shè)計(jì)中引入異步處理。創(chuàng)建線程有兩種方式，一種是直接繼承Thread，另外一種就是實(shí)現(xiàn)Runnable接口。Thread類、Runnable接口和Java內(nèi)存管理模型使得多線程編程簡(jiǎn)單直接。但是大家知道Thread和Runnable接口都不允許聲明檢查性異常，也不能返定義返回值。沒有返回值這點(diǎn)稍微有點(diǎn)麻煩。
  • 不能聲明拋出檢查型異常則更麻煩一些。public void run()方法契約意味著你必須捕獲并處理檢查型異常。即使你小心地保存了異常信息(在捕獲異常時(shí))以便稍后檢查， 但也不能保證這個(gè)類(Runnnable)的所有使用者都讀取這個(gè)異常信息。你也可以修改Runnable實(shí)現(xiàn)的getter，讓他們都能拋出任務(wù)執(zhí)行中的異常。但這種方法除了繁瑣也不是十分安全可靠，你不能強(qiáng)迫使用者調(diào)用這些方法，程序員很可能會(huì)調(diào)用join()方法等待線程結(jié)束后就不管了。
  • 但是現(xiàn)在不用擔(dān)心了，以上問題終于在1.5中解決了。Callable接口和Future接口接口的引入以及他們對(duì)線程池的支持優(yōu)雅地解決了這個(gè)兩個(gè)問題。

二、Callable

Callable.java源碼地址

(一) Runnable

說到Callable就不能不說下java.lang.Runnable，它是一個(gè)接口，它只聲明了一個(gè)run()方法，由于這個(gè)run()方法的返回值是void的，所以在執(zhí)行完任務(wù)之后無法返回任何結(jié)果。

public  interface Runnable {
    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see     java.lang.Thread#run()
     */
    public abstract void run();
}

(二) Callable

Callable位于java.util.concurrent包下，它也是一個(gè)接口，在它里面也只聲明了一個(gè)方法，只不過這個(gè)方法叫做call()

/**
 * A task that returns a result and may throw an exception.
 * Implementors define a single method with no arguments called
 * {@code call}.
 *
 * <p>The {@code Callable} interface is similar to {@link
 * java.lang.Runnable}, in that both are designed for classes whose
 * instances are potentially executed by another thread.  A
 * {@code Runnable}, however, does not return a result and cannot
 * throw a checked exception.
 *
 * <p>The {@link Executors} class contains utility methods to
 * convert from other common forms to {@code Callable} classes.
 *
 * @see Executor
 * @since 1.5
 * @author Doug Lea
 * @param <V> the result type of method {@code call}
 */
@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
     //
    V call() throws Exception;
}

可以看到，這一個(gè)泛型的接口，call()函數(shù)返回的類型就是傳遞進(jìn)來的V類型。

(三) Callable的類注釋

為了讓我們更好的理解Callable，還是從類的注釋開始，翻譯如下：

  • 有執(zhí)行結(jié)果的，并且可以引發(fā)異常的任務(wù)
    它的實(shí)現(xiàn)類需要去實(shí)現(xiàn)沒有參數(shù)的的方法，即call()方法
  • Callable有點(diǎn)類似于Runnable接口，因?yàn)樗鼈兌际窃O(shè)計(jì)成被線程去執(zhí)行的可執(zhí)行代碼的實(shí)例。由于Runnable是沒有返回值的，并且不能拋出一個(gè)檢查出的異常。
  • Executors類包含方法可以使得Callable轉(zhuǎn)化成其他普通形式。

(四)、 Runnable和Callable的區(qū)別：

  • 1、Runnable是Java 1.1有的，而Callable是1.5之后才加上去的
  • 2、Callable規(guī)定的方法是call()，Runnable規(guī)定的方法是run()
  • 3、Callable的任務(wù)執(zhí)行后可返回值，而Runnable的任務(wù)是不能返回(因?yàn)槭莢oid)
  • 4、call()方法是可以拋出異常的，而run()方法不可以
  • 5、運(yùn)行Callable任務(wù)可以拿到一個(gè)Future對(duì)象，表示異步計(jì)算的結(jié)果，它提供了檢查計(jì)算是否完成的方法，以等待計(jì)算的完成，并檢索計(jì)算的結(jié)果，通過Future對(duì)象可以了解任務(wù)的執(zhí)行情況，可取消任務(wù)的執(zhí)行，還可獲取執(zhí)行結(jié)果
  • 6、加入線程池運(yùn)行，Runnable使用ExecutorService的execute()方法，Callable使用submit()方法

三、Future

Future.java源碼地址

(一) Future類詳解

Future就是對(duì)于具體的Runnable或者Callable任務(wù)的執(zhí)行結(jié)果進(jìn)行取消、查詢是否完成、獲取結(jié)果。必要時(shí)可以通過get方法獲取執(zhí)行結(jié)果，該方法會(huì)阻塞知道任務(wù)返回結(jié)果。Future類位于java.util.concurrent包下，它是一個(gè)接口：

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個(gè)方法，下面依次解釋下每個(gè)方法的作用：

  • boolean cancel(boolean mayInterruptIfRunning)：方法用來取消任務(wù)，如果取消任務(wù)成功則返回true，如果取消任務(wù)失敗則返回false。參數(shù)mayInterruptIfRunning表示是否允許取消正在執(zhí)行卻沒有執(zhí)行完畢的任務(wù)，如果設(shè)置為true，則表示可以取消正在執(zhí)行過程中任務(wù)。如果任務(wù)已經(jīng)完成，則無論mayInterruptIfRunning為true還是false，此方法肯定返回false，即如果取消已完成的任務(wù)會(huì)返回false；如果任務(wù)正在執(zhí)行，若mayInterruptIfRunning設(shè)置為true，則返回true，若mayInterruptIfRunning設(shè)置為false，則返回false；如果任務(wù)還沒有執(zhí)行，則無論mayInterruptIfRunning為true還是false，肯定返回true。
  • boolean isCancelled()：表示任務(wù)是否被取消成功，如果在任務(wù)正常完成之前取消成功則返回true.
  • isDone()：方法表示任務(wù)是否已經(jīng)完成，若任務(wù)完成，則返回true。
  • V get()：方法用來獲取執(zhí)行結(jié)果，這個(gè)方法會(huì)產(chǎn)生阻塞，會(huì)一直等到任務(wù)執(zhí)行完畢才返回
  • V get(long timeout, TimeUnit unit)：用來獲取執(zhí)行結(jié)果，如果在指定時(shí)間內(nèi)，還沒獲取到結(jié)果，就直接返回null。

總結(jié)一下，F(xiàn)uture提供了三種功能:

  • 判斷任務(wù)是否完成
  • 能夠中斷任務(wù)
  • 能夠獲取任務(wù)的執(zhí)行結(jié)果

(二) Future類注釋

  • Future可以表示異步計(jì)算的結(jié)果。Future提供一個(gè)方法用來檢查這個(gè)計(jì)算是否已經(jīng)完成，還提供一個(gè)方法用來檢索計(jì)算結(jié)果。get()方法可以獲取計(jì)算結(jié)果，這個(gè)方法里面可能產(chǎn)生阻塞，如果產(chǎn)生了阻塞了，就阻塞到計(jì)算結(jié)束。cancel()方法可以取消執(zhí)行。還有一些方法可以用來確定任務(wù)是否已經(jīng)完成、是否已經(jīng)取消成功了。如果任務(wù)已經(jīng)執(zhí)行完畢，則是不能取消的。如果你想使用Future并且，希望它是不可撤銷的，同時(shí)不關(guān)心執(zhí)行的結(jié)果，可以聲明Future的泛型，并且基礎(chǔ)任務(wù)返回值結(jié)果為null。
  • 使用示例
   class App {
       ExecutorService executor = ...
      ArchiveSearcher searcher = ...

        void showSearch(final String target)
                throws InterruptedException {
            Future<String> future
                    = executor.submit(new Callable<String>() {
                public String call() {
                    return searcher.search(target);
                }
            });
            displayOtherThings(); // do other things while searching
            try {
                displayText(future.get()); // use future
            } catch (ExecutionException ex) {
                cleanup();
                return;
            }
        }
   }

FutureTask是Future的具體實(shí)現(xiàn)類，同時(shí)也實(shí)現(xiàn)了Runnable。所以FutureTask可以被Executor執(zhí)行。例如Executor的submit方法可以換成如下寫法

FutureTask<String> future =
  new FutureTask<>(new Callable<String>() {
    public String call() {
      return searcher.search(target);
  }});
executor.execute(future);

內(nèi)存一致性效應(yīng)：如果想在另一個(gè)線程調(diào)用 Future.get()方法，則在調(diào)用該方法之前應(yīng)該先執(zhí)行其自己的操作。

(三) boolean cancel(boolean mayInterruptIfRunning)方法注釋

翻譯如下：

嘗試去關(guān)閉正在執(zhí)行的任務(wù)，如果任務(wù)已經(jīng)完成，或者任務(wù)已經(jīng)被取消，或者任務(wù)因?yàn)槟撤N原因而無法被取消，則關(guān)閉事失敗。當(dāng)這個(gè)任務(wù)還沒有被執(zhí)行，則調(diào)用此方法會(huì)成功，并且這個(gè)任務(wù)將來不會(huì)被執(zhí)行。如果任務(wù)已經(jīng)開始了，mayInterruptIfRunning這個(gè)入?yún)Q定是否應(yīng)該中斷該任務(wù)。這個(gè)方法被執(zhí)行返回后，再去調(diào)用isDone()方法，將一直返回true。如果調(diào)用這個(gè)方法后返回true，再去調(diào)用isCancelled()方法，則isCancelled()方法一直返回true。mayInterruptIfRunning這個(gè)參數(shù)表示的是該任務(wù)的線程是否可以被中斷，true表示可以中斷，如果這個(gè)任務(wù)不能被取消，則返回false，而大多數(shù)這種情況是任務(wù)已完成。

上面已經(jīng)提到了Future只是一個(gè)接口，所以是無法直接用來創(chuàng)建對(duì)象使用的，在注釋里面推薦使用FutureTask，那我們就來看下FutureTask

四、FutureTask

FutureTask.java源碼地址
我們先來看一下FutureTask的實(shí)現(xiàn)：

(一)、RunnableFuture

通過代碼我們知道

public class FutureTask<V> implements RunnableFuture<V>

說明FutureTask類實(shí)現(xiàn)了RunnableFuture接口，我們看一下RunnableFuture接口

RunnableFuture.java源碼地址

/**
 * A {@link Future} that is {@link Runnable}. Successful execution of
 * the {@code run} method causes completion of the {@code Future}
 * and allows access to its results.
 * @see FutureTask
 * @see Executor
 * @since 1.6
 * @author Doug Lea
 * @param <V> The result type returned by this Future's {@code get} method
 */
public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

通過代碼我知道RunnableFuture繼承了Runnable接口和Future接口，而FutureTask實(shí)現(xiàn)了RunnableFuture接口。所以它既可以作為Runnable被線程執(zhí)行，也可以作為Future得到Callable的返回值

(二)、FutureTask的類注釋

/**
 * A cancellable asynchronous computation.  This class provides a base
 * implementation of {@link Future}, with methods to start and cancel
 * a computation, query to see if the computation is complete, and
 * retrieve the result of the computation.  The result can only be
 * retrieved when the computation has completed; the {@code get}
 * methods will block if the computation has not yet completed.  Once
 * the computation has completed, the computation cannot be restarted
 * or cancelled (unless the computation is invoked using
 * {@link #runAndReset}).
 *
 * <p>A {@code FutureTask} can be used to wrap a {@link Callable} or
 * {@link Runnable} object.  Because {@code FutureTask} implements
 * {@code Runnable}, a {@code FutureTask} can be submitted to an
 * {@link Executor} for execution.
 *
 * <p>In addition to serving as a standalone class, this class provides
 * {@code protected} functionality that may be useful when creating
 * customized task classes.
 *
 * @since 1.5
 * @author Doug Lea
 * @param <V> The result type returned by this FutureTask's {@code get} methods
 */

為了更好的理解作者設(shè)計(jì)，先來看下類注釋，翻譯如下：

  • 一個(gè)可以取消的異步執(zhí)行，這個(gè)類是Future的基礎(chǔ)實(shí)現(xiàn)類，提供一下方法，比如可以去開啟和關(guān)閉執(zhí)行，查詢是否已經(jīng)執(zhí)行完畢，以及檢索計(jì)算的結(jié)果。這個(gè)執(zhí)行結(jié)果只能等執(zhí)行完畢才能獲取，如果還未執(zhí)行完畢則處于阻塞狀態(tài)。除非調(diào)用runAndReset()方法，否則一旦計(jì)算完畢后則無法重啟啟動(dòng)或者取消。
  • Callable或者Runnable可以包裝FutureTask，由于FutureTask實(shí)現(xiàn)Runnable，所以在Executor可以執(zhí)行FutureTask
  • 除了可以作為一個(gè)獨(dú)立的類外，F(xiàn)utureTask還提供一些protected方法，這樣在自定義任務(wù)類是，就會(huì)很方便。

(三)、FutureTask的結(jié)構(gòu)

結(jié)構(gòu)如下圖


FutureTask結(jié)構(gòu)圖.png

繼承關(guān)系如下:

繼承關(guān)系.png

(四)、靜態(tài)final類WaitNode

    /**
     * Simple linked list nodes to record waiting threads in a Treiber
     * stack.  See other classes such as Phaser and SynchronousQueue
     * for more detailed explanation.
     */
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }

先翻譯一下注釋

精簡(jiǎn)版的鏈表結(jié)構(gòu)，其中每一個(gè)節(jié)點(diǎn)代表堆棧中等待的線程。如果想了解更多的詳細(xì)說明，請(qǐng)參考其他類比如Phaser和SynchronousQueue

結(jié)構(gòu)如下圖


WaitNode.png

再來看下構(gòu)造函數(shù)和成員變量

  • thread：代表等待的線程
  • next：代表下一個(gè)節(jié)點(diǎn)，通過這個(gè)節(jié)點(diǎn)我們也能退出這個(gè)鏈表是單向鏈表
  • 構(gòu)造函數(shù)：無參的構(gòu)造函數(shù)里面將thread設(shè)置為當(dāng)前線程。
    總結(jié)：
    WaitNode就是一個(gè)鏈表結(jié)構(gòu)，用于記錄等待當(dāng)前FutureTask結(jié)果的線程。

(五)、FutureTask的狀態(tài)

FutureTask一共有7種狀態(tài)，代碼如下：

    /*
     * Revision notes: This differs from previous versions of this
     * class that relied on AbstractQueuedSynchronizer, mainly to
     * avoid surprising users about retaining interrupt status during
     * cancellation races. Sync control in the current design relies
     * on a "state" field updated via CAS to track completion, along
     * with a simple Treiber stack to hold waiting threads.
     *
     * Style note: As usual, we bypass overhead of using
     * AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
     */

    /**
     * The run state of this task, initially NEW.  The run state
     * transitions to a terminal state only in methods set,
     * setException, and cancel.  During completion, state may take on
     * transient values of COMPLETING (while outcome is being set) or
     * INTERRUPTING (only while interrupting the runner to satisfy a
     * cancel(true)). Transitions from these intermediate to final
     * states use cheaper ordered/lazy writes because values are unique
     * and cannot be further modified.
     *
     * Possible state transitions:
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
     */
    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;

老規(guī)矩先來翻譯一下注釋，我們看到注釋有兩部分，我們依次翻譯如下：

  • 上半部分注釋：
    修訂說明：和之前版本的AbstractQueuedSynchronizer不同，主要是為了避免令人驚訝的用戶在取消競(jìng)爭(zhēng)遷建保留中斷的狀態(tài)。在當(dāng)前設(shè)計(jì)中，同步的控制是通過CAS中的更新字段——"state"來完成跟蹤的。并且通過一個(gè)Treiber堆棧來保存這些等待的線程。風(fēng)格筆記(筆者注：這個(gè)真心不知道怎么翻譯，誰知道請(qǐng)?jiān)谙旅媪粞?：和往常一樣，我們直接使用不安全的內(nèi)在函數(shù)， 并且忽略使用AtomicXFieldUpdaters的開銷。

簡(jiǎn)單的說就是，F(xiàn)utureTask中使用state表示任務(wù)狀態(tài)，state變更由CAS操作保證原子性。

  • 下半部分注釋：
    這個(gè)任務(wù)的運(yùn)行狀態(tài)，最初是NEW狀態(tài)，在調(diào)用set()方法或者setException()方法或者cancel()方法后，運(yùn)行的狀態(tài)就變?yōu)榻K端的狀態(tài)。在運(yùn)行期間，如果計(jì)算出結(jié)果后，狀態(tài)變更為COMPLETING，如果通過調(diào)用cancel(true)安全的中斷運(yùn)行，則狀態(tài)變更為INTERRUPTING。由于值是唯一且不能被進(jìn)一步修改，所以從中間狀態(tài)到最終狀態(tài)的轉(zhuǎn)化是有序的。
  • 可能的狀態(tài)變更流程
  • NEW -> COMPLETING -> NORMAL
  • NEW -> COMPLETING -> EXCEPTIONAL
  • NEW -> CANCELLED
  • NEW -> INTERRUPTING -> INTERRUPTED

上面翻譯我感覺不是很好，用白話解釋一下：

FutureTask對(duì)象初始化時(shí)，在構(gòu)造器把state設(shè)置為NEW，之后狀態(tài)變更依據(jù)具體執(zhí)行情況來定。

  • 任務(wù)執(zhí)行正常，并且還沒結(jié)束，state為COMPLETING，代表任務(wù)正在執(zhí)行即將完成，接下來很快會(huì)被設(shè)置為NORMAL或者EXCEPTIONAL，這取決于調(diào)用Runnable中的call()方法是否拋出異常，沒有異常則是NORMAL，拋出異常是EXCEPTIONAL。
  • 任務(wù)提交后、任務(wù)結(jié)束前取消任務(wù)，都有可能變?yōu)镃ANCELLED或者INTERRUPTED。在調(diào)用cancel(boolean) 是，如果傳入false表示不中斷線程，state會(huì)變成CANCELLED，如果傳入true，則state先變?yōu)?br> INTERRUPTING，中斷完成后，變?yōu)镮NTERRUPTED。

總結(jié)一下就是：FutureTask的狀態(tài)變化過程為，以下4種情況：

  • 任務(wù)正常執(zhí)行并返回： NEW -> COMPLETING -> NORMAL
  • 任務(wù)執(zhí)行中出現(xiàn)異常：NEW -> COMPLETING -> EXCEPTIONAL
  • 任務(wù)執(zhí)行過程中被取消，并且不中斷線程：NEW -> CANCELLED
  • 任務(wù)執(zhí)行過程中被取消，并且中斷線程：NEW -> INTERRUPTING -> INTERRUPTED

那我們就簡(jiǎn)單的解釋下幾種狀態(tài)

  • NEW：任務(wù)初始化狀態(tài)
  • COMPLETING：任務(wù)正在完成狀態(tài)(任務(wù)已經(jīng)執(zhí)行完成，但是結(jié)果還沒有賦值給outcome)
  • NORMAL：任務(wù)完成(結(jié)果已經(jīng)賦值給outcome)
  • EXCEPTIONAL：任務(wù)執(zhí)行異常
  • CANCELLED：任務(wù)被取消
  • INTERRUPTING：任務(wù)被中斷中
  • INTERRUPTED：任務(wù)被中斷

(六)、FutureTask的成員變量

    /** 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;
  • callable：任務(wù)具體執(zhí)行體，具體要做的事
  • outcome：任務(wù)的執(zhí)行結(jié)果，get()方法的返回值
  • runner：任務(wù)的執(zhí)行線程
  • waiters：獲取任務(wù)結(jié)果的等待線程(是一個(gè)鏈?zhǔn)搅斜?

(七)、FutureTask的構(gòu)造函數(shù)

FutureTask有兩個(gè)構(gòu)造函數(shù)
分別是FutureTask(Callable<V> callable)和FutureTask(Runnable runnable, V result)，那我們來依次分析

1、構(gòu)造函數(shù) FutureTask(Callable<V> callable)
    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Callable}.
     *
     * @param  callable the callable task
     * @throws NullPointerException if the callable is null
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

老規(guī)矩先翻譯一下注釋：

創(chuàng)建一個(gè)FutureTask，并在將來執(zhí)行的時(shí)候，運(yùn)行傳入的Callable。

看下代碼，我們知道：

  • 1、通過代碼我們知道如果傳入的Callable為空直接拋出異常，說明構(gòu)造時(shí)傳入的Callable不能為空。
  • 2、設(shè)置當(dāng)前狀態(tài)為NEW。

所以總結(jié)一句話就是，通過傳入Callable來構(gòu)造一個(gè)FutureTask。

2、構(gòu)造函數(shù) FutureTask(Runnable runnable, V result)
    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Runnable}, and arrange that {@code get} will return the
     * given result on successful completion.
     *
     * @param runnable the runnable task
     * @param result the result to return on successful completion. If
     * you don't need a particular result, consider using
     * constructions of the form:
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @throws NullPointerException if the runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

老規(guī)矩先翻譯一下注釋：

創(chuàng)建一個(gè)FutureTask，并在將來執(zhí)行的時(shí)候，運(yùn)行傳入的Runnable，并且將成功完成后的結(jié)果返給傳入的result。

看下代碼，我們知道：

  • 1、先通過調(diào)用Executors的callable(Runnable, T)方法返回的Callable
  • 2、將上面返回的Callable指向本地變量callable
  • 3、設(shè)置當(dāng)前狀態(tài)為NEW。

所以總結(jié)一句話就是，通過傳入Runnable來構(gòu)造一個(gè)任務(wù)

這里順帶說下Executors.callable(runnable, result)方法的內(nèi)部實(shí)現(xiàn)

2.1、Executors.callable(Runnable, T)
    /**
     * Returns a {@link Callable} object that, when
     * called, runs the given task and returns the given result.  This
     * can be useful when applying methods requiring a
     * {@code Callable} to an otherwise resultless action.
     * @param task the task to run
     * @param result the result to return
     * @param <T> the type of the result
     * @return a callable object
     * @throws NullPointerException if task null
     */
    public static <T> Callable<T> callable(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
        return new RunnableAdapter<T>(task, result);
    }

方法內(nèi)部很簡(jiǎn)單，就是new了一個(gè)RunnableAdapter并返回，那我們來看下RunnableAdapterr適配器

    /**
     * A callable that runs given task and returns given result.
     */
    private static final class RunnableAdapter<T> implements Callable<T> {
        private final Runnable task;
        private final T result;
        RunnableAdapter(Runnable task, T result) {
            this.task = task;
            this.result = result;
        }
        public T call() {
            task.run();
            return result;
        }
    }

通過上面代碼我們知道：

  • RunnableAdapter是FutureTask的一個(gè)靜態(tài)內(nèi)部類并且實(shí)現(xiàn)了Callable，也就是說RunnableAdapter是Callable子類。
  • call方法實(shí)現(xiàn)代碼是，執(zhí)行Runnable的run方法，并返回構(gòu)造函數(shù)傳入的result參數(shù)。

這里實(shí)際上是將一個(gè)Runnable對(duì)象偽裝成一個(gè)Callable對(duì)象，是適配器對(duì)象。

3、構(gòu)造函數(shù)總結(jié)

通過分析上面兩個(gè)構(gòu)造函數(shù)，我們知道無論采用第一個(gè)構(gòu)造函數(shù)，還是第二個(gè)構(gòu)造函數(shù)，其結(jié)果都是給本地變量callable初始化賦值，所以說FutureTask最終都是執(zhí)行Callable類型的任務(wù)。然后設(shè)置狀態(tài)為NEW。

(八)、FutureTask的幾個(gè)核心方法

FutureTask有幾個(gè)核心方法：

  • public void run()：表示任務(wù)的執(zhí)行
  • public V get()和public V get(long timeout, TimeUnit unit)：表示獲取任務(wù)的結(jié)果
  • public boolean cancel(boolean mayInterruptIfRunning)：表示取消任務(wù)

那我們就依次來看下

1、run()方法
    /**
     * 判斷任務(wù)的狀態(tài)是否是初始化的狀態(tài)
     * 判斷執(zhí)行任務(wù)的線程對(duì)象runner是否為null，為空就將當(dāng)前執(zhí)行線程賦值給runner屬性
     *  不為空說明應(yīng)有線程準(zhǔn)備執(zhí)行這個(gè)任務(wù)了
     */
    public void run() {
        if (state != NEW ||
            !U.compareAndSwapObject(this, RUNNER, null, Thread.currentThread()))
            return;
        try {
           // 任務(wù)狀態(tài)時(shí)NEW，并且callable不為空，則執(zhí)行任務(wù)
           // 如果認(rèn)為被cancel了，callable會(huì)被置空
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                
                V result;  // 結(jié)果的變量
                
                boolean ran;  // 執(zhí)行完畢的變量
                try {
                    // 執(zhí)行任務(wù)并返回結(jié)果
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    // 執(zhí)行異常
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    //任務(wù)執(zhí)行完畢就設(shè)置結(jié)果
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            // 將執(zhí)行任務(wù)的執(zhí)行線程清空 
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)
                //判斷線程的狀態(tài)
                handlePossibleCancellationInterrupt(s);
        }
    }

通過上面代碼和注釋我們知道run方法內(nèi)部的流程如下：

  • 第一步：檢查當(dāng)前任務(wù)是否是NEW以及runner是否為空，這一步是防止任務(wù)被取消
  • 第二步：double-check任務(wù)狀態(tài)和state
  • 第三步：執(zhí)行業(yè)務(wù)邏輯，也就是c.call()方法被執(zhí)行
  • 第四步：如果業(yè)務(wù)邏輯異常，則調(diào)用setException方法將異常對(duì)象賦值給outcome，并更新state的值
  • 第五步：如果業(yè)務(wù)正常，則調(diào)用set方法將執(zhí)行結(jié)果賦給outcome，并更新state值。
1.1、Unsafe類

Java不能夠直接訪問操作系統(tǒng)底層，而是通過本地方法來訪問。Unsafe提供了硬件級(jí)別的原子訪問，主要提供以下功能：

  • 分配釋放內(nèi)存
  • 定位某個(gè)字段的內(nèi)存位置
  • 掛起一個(gè)線程和恢復(fù)，更多的是通過LockSupport來訪問。park和unpark
  • CAS操作，比較一個(gè)對(duì)象的某個(gè)位置的內(nèi)存值是否與期望值一致。

主要方法是compareAndSwap()

1.1.1UNSAFE.compareAndSwapObject(this,runnerOffset,null, Thread.currentThread())方法

UNSAFE.compareAndSwapObject(this,RUNNER,null, Thread.currentThread())這行代碼什么意思？

compareAndSwapObject可以通過反射，根據(jù)偏移量去修改對(duì)象，第一個(gè)參數(shù)表示要修改的對(duì)象，第二個(gè)表示偏移量，第三個(gè)參數(shù)用于和偏移量對(duì)應(yīng)的值進(jìn)行比較，第四個(gè)參數(shù)表示如何偏移量對(duì)應(yīng)的值和第三個(gè)參數(shù)一樣時(shí)要把偏移量設(shè)置成的值。

翻譯成白話文就是：

如果this對(duì)象的RUNNER偏移地址的值是null，那就把它設(shè)置為Thread.currentThread()。

上面提到了一個(gè)概念是RUNNER，那這個(gè)RUNNER 是什么東東？
我們?cè)谠创a中找到

    // Unsafe mechanics
    private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
    private static final long STATE;
    private static final long RUNNER;
    private static final long WAITERS;
    static {
        try {
            STATE = U.objectFieldOffset
                (FutureTask.class.getDeclaredField("state"));
            RUNNER = U.objectFieldOffset
                (FutureTask.class.getDeclaredField("runner"));
            WAITERS = U.objectFieldOffset
                (FutureTask.class.getDeclaredField("waiters"));
        } catch (ReflectiveOperationException e) {
            throw new Error(e);
        }

        // Reduce the risk of rare disastrous classloading in first call to
        // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
        Class<?> ensureLoaded = LockSupport.class;
    }

它對(duì)應(yīng)的就是runner的成員變量，也就是說如果狀態(tài)不是NEW或者runner不是null，run方法直接返回。

所以我們知道

  • private static final long STATE：表示state這個(gè)成員變量
  • private static final long RUNNER：表示的是runner這個(gè)成員變量
  • rivate static final long WAITERS：表示的是waiters這個(gè)成員變量

在這個(gè)run方法里面分別調(diào)用了setException(Throwable )和set(V)方法，那我們就來詳細(xì)看下

1.2、setException(Throwable t)方法
    /**
     * Causes this future to report an {@link ExecutionException}
     * with the given throwable as its cause, unless this future has
     * already been set or has been cancelled.
     *
     * <p>This method is invoked internally by the {@link #run} method
     * upon failure of the computation.
     *
     * @param t the cause of failure
     */
    protected void setException(Throwable t) {
         // state狀態(tài) NEW-> COMPLETING
        if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
            outcome = t;
             // // COMPLETING -> EXCEPTIONAL 到達(dá)穩(wěn)定狀態(tài)
            U.putOrderedInt(this, STATE, EXCEPTIONAL); // final state
            // 一些 結(jié)束工作
            finishCompletion();
        }
    }

簡(jiǎn)單翻譯一下方法的注釋:

  • 除非這個(gè)Future已經(jīng)設(shè)置過了，或者被取消了，否則這個(gè)產(chǎn)生的異常將會(huì)匯報(bào)到ExecutionException里面
  • 如果在run()方法里面產(chǎn)生了異常，則會(huì)調(diào)用這個(gè)方法

所以總結(jié)一下就是：當(dāng)任務(wù)執(zhí)行過程中出現(xiàn)異常時(shí)候，對(duì)異常的處理方式

PS：這個(gè)方法是protected，所以可以重寫

1.3、set(V v)方法

這個(gè)方法主要是：執(zhí)行結(jié)果的賦值操作

    /**
     * Sets the result of this future to the given value unless
     * this future has already been set or has been cancelled.
     *
     * <p>This method is invoked internally by the {@link #run} method
     * upon successful completion of the computation.
     *
     * @param v the value
     */
    protected void set(V v) {
        // state 狀態(tài)  NEW->COMPLETING
        if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
            outcome = v;
            // COMPLETING -> NORMAL 到達(dá)穩(wěn)定狀態(tài)
            U.putOrderedInt(this, STATE, NORMAL); // final state
            // 一些結(jié)束工作
            finishCompletion();
        }
    }

通過上面我們知道這個(gè)方法內(nèi)部的流程如下：

  • 首先 將任務(wù)的狀態(tài)改變
  • 其次 將結(jié)果賦值
  • 再次 改變?nèi)蝿?wù)狀態(tài)
  • 最后 處理等待線程隊(duì)列(將線程阻塞狀態(tài)改為喚醒，這樣等待線程就拿到結(jié)果了)

PS：這里使用的是 UNSAFE的putOrderedInt方法，其實(shí)就是原子量的LazySet內(nèi)部使用的方法，為什么要用這個(gè)方法？首先LazySet相對(duì)于Volatile-Write來說更加廉價(jià)，因?yàn)樗鼪]有昂貴的Store/Load屏障，其次后續(xù)線程不會(huì)及時(shí)的看到state從COMPLETING變?yōu)镹ORMAL，但這沒有什么關(guān)系，而且NORMAL是state最終的狀態(tài)，不會(huì)再變化了。

在這個(gè)方法里面調(diào)用了finishCompletion()方法，那我們就來看下這個(gè)方法

1.4、finishCompletion()方法

這個(gè)方法是：

在任務(wù)執(zhí)行完成(包括取消、正常結(jié)束、發(fā)生異常)，將等待線程隊(duì)列喚醒，同時(shí)讓任務(wù)執(zhí)行體清空。

代碼如下：

    /**
     * Removes and signals all waiting threads, invokes done(), and
     * nulls out callable.
     */
    private void finishCompletion() {
        // assert state > COMPLETING;
        // 遍歷等待節(jié)點(diǎn)
        for (WaitNode q; (q = waiters) != null;) {
            if (U.compareAndSwapObject(this, WAITERS, 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;
            }
        }
        // 這里可以自定義實(shí)現(xiàn)任務(wù)完成后要做的事情(在子類重寫done()方法)
        done();
        // 清空callable
        callable = null;        // to reduce footprint
    }

由代碼和注釋可以看出來，這里就是遍歷WaitNode鏈表，對(duì)每一個(gè)WaitNode對(duì)應(yīng)的線程依次進(jìn)行LockSupport.unpark(t)，使其結(jié)束阻塞。WaitNode通知完畢后，調(diào)用done方法。目前該方法是空實(shí)現(xiàn)，所以如果你想在任務(wù)完成后執(zhí)行一些業(yè)務(wù)邏輯可以重寫這個(gè)方法。所以這個(gè)方法主要是在于喚醒等待線程。由前面知道，當(dāng)任務(wù)正常結(jié)束或者異常結(jié)束時(shí)，都會(huì)調(diào)用finishCompletion()去喚醒等待線程。這時(shí)候等待線程就可以醒來，可以獲取結(jié)果了。

·

1.4.1、LockSupport簡(jiǎn)介

這里首先說下LockSupport，很多新手對(duì)這個(gè)東西，不是很熟悉，我先簡(jiǎn)單說下，這里就不詳細(xì)說明了。

LockSupport是構(gòu)建concurrent包的基礎(chǔ)之一

####### ① 操作對(duì)象
LockSupport調(diào)用Unsafe的natvie代碼：

public native void unpark(Thread jthread); 
public native void park(boolean isAbsolute, long time);

這兩個(gè)函數(shù)聲明清楚地說明了操作對(duì)象：park函數(shù)是將當(dāng)前Thread阻塞，而unPark函數(shù)則是將另一個(gè)Thread喚醒。

與Object類的wait/notify 機(jī)制相比，park/unpark有兩個(gè)優(yōu)點(diǎn)：

  • 1、以thread為操作對(duì)象更符合阻塞線程的直觀定義
  • 2、操作更精準(zhǔn)，可以準(zhǔn)確地喚醒某一個(gè)線程(notify隨機(jī)喚醒一個(gè)線程，notifyAll喚醒所有等待的線程)，增加了靈活性

####### ② 關(guān)于許可
在上面的文件，使用了阻塞和喚醒，是為了和wait/notify做對(duì)比。其實(shí)park/unpark的設(shè)計(jì)原理核心是"許可"。park是等待一個(gè)許可。unpark是為某線程提供一個(gè)"許可"。如果說某線程A調(diào)用park，那么除非另外一個(gè)線程unpark(A)給A一個(gè)許可，否則線程A將阻塞在park操作上。

1.5、handlePossibleCancellationInterrupt(int) 方法
    /**
     * Ensures that any interrupt from a possible cancel(true) is only
     * delivered to a task while in run or runAndReset.
     */
    private void handlePossibleCancellationInterrupt(int s) {
        // It is possible for our interrupter to stall before getting a
        // chance to interrupt us.  Let's spin-wait patiently.
        // 如果當(dāng)前正在中斷過程中，自等待，等中斷完成
        if (s == INTERRUPTING)
            while (state == INTERRUPTING)
                Thread.yield(); // wait out pending interrupt

        // assert state == INTERRUPTED;

        // We want to clear any interrupt we may have received from
        // cancel(true).  However, it is permissible to use interrupts
        // as an independent mechanism for a task to communicate with
        // its caller, and there is no way to clear only the
        // cancellation interrupt.
        //
        // Thread.interrupted();
    }

先來看下注釋：

執(zhí)行計(jì)算而不設(shè)置其結(jié)果，然后重新設(shè)置future為初始化狀態(tài)，如果執(zhí)行遇到異?；蛘呷蝿?wù)被取消，則不再執(zhí)行此操作。這樣設(shè)計(jì)的目的是：執(zhí)行多次任務(wù)。

看代碼我們知道他主要就是: 如果其他線程正在終止該任務(wù)，那么運(yùn)行該任務(wù)的線程就暫時(shí)讓出CPU時(shí)間一直到state= INTERRUPTED為止。

所以它的作用就是:

確保cancel(true) 產(chǎn)生的中斷發(fā)生在run()或者 runAndReset()方法過程中

2、get()與get(long, TimeUnit)方法

任務(wù)是由線程池提供的線程執(zhí)行，那么這時(shí)候主線程則會(huì)阻塞，直到任務(wù)線程喚醒它們。我們看看get()是怎么做的？

2.1 get()方法

代碼如下：

    /**
     * @throws CancellationException {@inheritDoc}
     */
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        // state 小于 COMPLETING 則說明任務(wù)仍然在執(zhí)行，且沒有被取消
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }

通過代碼我們知道

  • 首先 校驗(yàn)參數(shù)
  • 然后 判斷是否正常執(zhí)行且沒有被取消，如果沒有則調(diào)用awaitDone(boolean,long)方法
  • 最后 調(diào)用report(int) 方法

這里面涉及兩個(gè)方法分別是awaitDone(boolean,long)方法和report(int) 方法，那讓我們依次來看下。

2.1.1 awaitDone(boolean,long)方法

這個(gè)方法主要是等待任務(wù)執(zhí)行完畢，如果任務(wù)取消或者超時(shí)則停止
代碼如下：


    /**
     * 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 or at timeout
     */
    private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        // The code below is very delicate, to achieve these goals:
        // - call nanoTime exactly once for each call to park
        // - if nanos <= 0L, return promptly without allocation or nanoTime
        // - if nanos == Long.MIN_VALUE, don't underflow
        // - if nanos == Long.MAX_VALUE, and nanoTime is non-monotonic
        //   and we suffer a spurious wakeup, we will do no worse than
        //   to park-spin for a while
        // 起始時(shí)間
        long startTime = 0L;    // Special value 0L means not yet parked
        // 當(dāng)前等待線程的節(jié)點(diǎn)
        WaitNode q = null;
         // 是否將節(jié)點(diǎn)放在了等待列表中
        boolean queued = false;
        // 通過死循環(huán)來實(shí)現(xiàn)線程阻塞等待
        for (;;) {
            int s = state;
            if (s > COMPLETING) {
                 // 任務(wù)可能已經(jīng)完成或者被取消了
                if (q != null)
                    q.thread = null;
                return s;
            }
            else if (s == COMPLETING)
                // We may have already promised (via isDone) that we are done
                // so never return empty-handed or throw InterruptedException
                // 任務(wù)線程可能被阻塞了，讓出cpu
                Thread.yield();
            else if (Thread.interrupted()) {
                // 線程中斷則移除等待線程并拋出異常
                removeWaiter(q);
                throw new InterruptedException();
            }
            else if (q == null) {
                // 等待節(jié)點(diǎn)為空，則初始化新節(jié)點(diǎn)并關(guān)聯(lián)當(dāng)前線程
                if (timed && nanos <= 0L)
                   // 如果需要等待，并且等待時(shí)間小于0表示立即，則直接返回
                    return s;
                // 如果不需要等待，或者需要等待但是等待時(shí)間大于0。
                q = new WaitNode();
            }
            else if (!queued)
                // 等待線程入隊(duì)，因?yàn)槿绻腙?duì)成功則queued=true
                queued = U.compareAndSwapObject(this, WAITERS,
                                                q.next = waiters, q);
            else if (timed) {
                //如果有超時(shí)設(shè)置
                final long parkNanos;
                if (startTime == 0L) { // first time
                    startTime = System.nanoTime();
                    if (startTime == 0L)
                        startTime = 1L;
                    parkNanos = nanos;
                } else {
                    long elapsed = System.nanoTime() - startTime;
                    if (elapsed >= nanos) {
                        // 已經(jīng)超時(shí)，則移除等待節(jié)點(diǎn)
                        removeWaiter(q);
                        return state;
                    }
                    parkNanos = nanos - elapsed;
                }
                // nanoTime may be slow; recheck before parking
                if (state < COMPLETING)
                    // 任務(wù)還在執(zhí)行，且沒有被取消，所以繼續(xù)等待
                    LockSupport.parkNanos(this, parkNanos);
            }
            else
                LockSupport.park(this);
        }
    }

兩個(gè)入?yún)ⅲ?/p>

  • timed 為true 表示設(shè)置超時(shí)時(shí)間，false表示不設(shè)置超時(shí)間
  • nanos 表示超時(shí)的狀態(tài)

for死循環(huán)里面的邏輯如下：

  • 第一步 判斷任務(wù)是否已經(jīng)完處于完成或者取消了，如果直接返回轉(zhuǎn)狀態(tài)值，如果不是，則走第二步
  • 第二步，如果狀態(tài)值是COMPLETING，則說明當(dāng)前是在set()方法時(shí)被阻塞了，所以只需要讓出當(dāng)前線程的CPU資源。
  • 第三步，如果線程已經(jīng)中斷了，則移除線程并拋出異常
  • 第四步，如果能走到這一步，狀態(tài)值只剩下NEW了，如果狀態(tài)值是NEW，并且q==null，則說明這是第一次，所以初始化一個(gè)當(dāng)前線程的等待節(jié)點(diǎn)。
  • 第五步，此時(shí)queued=false，說明如果還沒入隊(duì)，則它是在等待入隊(duì)
  • 第六步，能走到這一步，說明queued=true，這時(shí)候判斷timed是否為true，如果為true則設(shè)置了超時(shí)時(shí)間，然后看一下startTime是否為0，如果為0，則說明是第一次，因?yàn)閟tartTime默認(rèn)值為0，如果是第一此，則設(shè)置startTime=1。保證startTime==0是第一次。如果startTime!=0，則說明不是第一次，如果不是第一次，則需要計(jì)算時(shí)差elapsed，如果elapsed大于nanos，則說明超時(shí)，如果小于則沒有超時(shí)，還有時(shí)差，然等待這個(gè)時(shí)差阻塞。
  • 第七步，如果timed為false，則說明沒有超時(shí)設(shè)置。

所以總結(jié)一下：

waitDone就是將當(dāng)前線程加入等待隊(duì)列(waitNode有當(dāng)前Thread的Thread變量)，然后用LockSupport將自己阻塞，等待超時(shí)或者被解除阻塞后，判斷是否已經(jīng)完成(state為>= COMPLETING)，如果未完成(state< COMPLETING)拋出超時(shí)異常，如果已完成則稍等或者直接返回結(jié)果。

這個(gè)方法里面調(diào)用了removeWaiter(WaitNode) 這個(gè)方法，所以我們來先看這個(gè)這個(gè)removeWaiter(WaitNode) 里面是怎么實(shí)現(xiàn)的

2.1.2 removeWaiter(WaitNode)
    /**
     * Tries to unlink a timed-out or interrupted wait node to avoid
     * accumulating garbage.  Internal nodes are simply unspliced
     * without CAS since it is harmless if they are traversed anyway
     * by releasers.  To avoid effects of unsplicing from already
     * removed nodes, the list is retraversed in case of an apparent
     * race.  This is slow when there are a lot of nodes, but we don't
     * expect lists to be long enough to outweigh higher-overhead
     * schemes.
     */
    private void removeWaiter(WaitNode node) {
        if (node != null) {
            // 將node 的thread 域置空
            node.thread = null;
            /**
             * 下面過程中會(huì)將node從等待隊(duì)列中移除，以thread為null為依據(jù)
             * 如果過程中發(fā)生了競(jìng)爭(zhēng)，重試
             */
            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 (!U.compareAndSwapObject(this, WAITERS, q, s))
                        continue retry;
                }
                break;
            }
        }
    }

首先來看下類的注釋

為了防止累積的內(nèi)存垃圾，所以需要去取消超時(shí)或者已經(jīng)被中斷的等待節(jié)點(diǎn)。內(nèi)部節(jié)點(diǎn)因?yàn)闆]有CAS所以很簡(jiǎn)單，所以他們可以被無害的釋放。為了避免已刪除節(jié)點(diǎn)的影響，如果存在競(jìng)爭(zhēng)的情況下，需要重新排列。所以當(dāng)節(jié)點(diǎn)很多是，速度會(huì)很慢，因此我們不建議列表太長(zhǎng)而導(dǎo)致效率降低。

這個(gè)方法主要就是將線程節(jié)點(diǎn)從等待隊(duì)列中移除

2.1.2 report(int) 方法
    /**
     * Returns result or throws exception for completed task.
     *
     * @param s completed state value
     */
    @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
        Object x = outcome;
         // 如果任務(wù)正常執(zhí)行完成，返回任務(wù)執(zhí)行結(jié)果
        if (s == NORMAL)
            return (V)x;

        // 如果任務(wù)被取消，拋出異常
        if (s >= CANCELLED)
            throw new CancellationException();
        
        // 其他狀態(tài) 拋出執(zhí)行異常 ExecutionException 
        throw new ExecutionException((Throwable)x);
    }
report.png

如果任務(wù)處于NEW、COMPLETING和INTERRUPTING 這三種狀態(tài)的時(shí)候是執(zhí)行不到report方法的，所以沒有對(duì)這三種狀態(tài)盡心轉(zhuǎn)換。

2.2 get(long, TimeUnit)方法

最多等待為計(jì)算完成所給的時(shí)間之后，獲取其結(jié)果

    /**
     * @throws CancellationException {@inheritDoc}
     */
    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方法源碼很簡(jiǎn)潔，首先校驗(yàn)參數(shù)，然后根據(jù)state狀態(tài)判斷是否超時(shí)，如果超時(shí)則異常，不超時(shí)則調(diào)用report去獲取最終結(jié)果。
當(dāng) s <= COMPLETING 時(shí)，表明任務(wù)仍然在執(zhí)行且沒有被取消，如果它為true，那么走到awaitDone方法。關(guān)于awaitDone方法上面已經(jīng)講解了，這里就不過闡述了。

3、cancel(boolean)方法

只能取消還沒有被執(zhí)行的任務(wù)(任務(wù)狀態(tài)為NEW的任務(wù))

    public boolean cancel(boolean mayInterruptIfRunning) {
        // 如果任務(wù)狀態(tài)不是初始化狀態(tài)，則取消任務(wù)
         //如果此時(shí)任務(wù)已經(jīng)執(zhí)行了，并且可能執(zhí)行完成，但是狀態(tài)改變還沒有來得及修改，也就是在run()方法中的set()方法還沒來得及調(diào)用
         //   繼續(xù)判斷任務(wù)的當(dāng)前狀態(tài)時(shí)否為NEW，因?yàn)榇藭r(shí)執(zhí)行任務(wù)線程可能再度獲得處理了，任務(wù)狀態(tài)可能已發(fā)生改變
        if (!(state == NEW &&
              U.compareAndSwapInt(this, STATE, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
                return false;

        // 如果任務(wù)狀態(tài)依然是NEW，也就是執(zhí)行線程沒有改變?nèi)蝿?wù)的狀態(tài)，
        // 則讓執(zhí)行線程中斷(在這個(gè)過程中執(zhí)行線程可能會(huì)改變?nèi)蝿?wù)的狀態(tài))
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    // 將任務(wù)狀態(tài)設(shè)置為中斷
                    U.putOrderedInt(this, STATE, INTERRUPTED);
                }
            }
        } finally {
            // 處理任務(wù)完成的結(jié)果
            finishCompletion();
        }
        return true;
    }

通過上述代碼，我們知道這個(gè)取消不一定起作用的。

上面的代碼邏輯如下：

  • 第一步：state不等于NEW，則表示任務(wù)即將進(jìn)入最終狀態(tài) ，則state == NEW為false，導(dǎo)致if成立直接返回false
  • 第二步：如果mayInterruptIfRunning為true在，表示中斷線程，則設(shè)置狀態(tài)為INTERRUPTING，中斷之后設(shè)置為INTERRUPTED。如果mayInterruptIfRunning為false，表示不中斷線程，把state設(shè)置為CANCELLED
  • 第三步：state狀態(tài)為NEW，任務(wù)可能已經(jīng)開始執(zhí)行，也可能還未開始，所以用Unsafe查看下，如果不是，則直接返回false
  • 第四步：移除等待線程
  • 第五步：?jiǎn)拘?/li>

所以，cancel()方法改變了futureTask的狀態(tài)為，如果傳入的是false，并且業(yè)務(wù)邏輯已經(jīng)開始執(zhí)行，當(dāng)前任務(wù)是不會(huì)被終止的，而是會(huì)繼續(xù)執(zhí)行，知道異?；蛘邎?zhí)行完畢。如果傳入的是true，會(huì)調(diào)用當(dāng)前線程的interrupt()方法，把中斷標(biāo)志位設(shè)為true。

事實(shí)上，除非線程自己停止自己的任務(wù)，或者退出JVM，是沒有其他方法完全終止一個(gè)線程任務(wù)的。mayInterruptIfRunning=true，通過希望當(dāng)前線程可以響應(yīng)中斷的方式來結(jié)束任務(wù)。當(dāng)任務(wù)被取消后，會(huì)被封裝為CancellationException拋出。

4、runAndReset() 方法

任務(wù)可以被多次執(zhí)行

    /**
     * Executes the computation without setting its result, and then
     * resets this future to initial state, failing to do so if the
     * computation encounters an exception or is cancelled.  This is
     * designed for use with tasks that intrinsically execute more
     * than once.
     *
     * @return {@code true} if successfully run and reset
     */
    protected boolean runAndReset() {
        if (state != NEW ||
            !U.compareAndSwapObject(this, RUNNER, 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;
    }

先來看下注釋：

執(zhí)行計(jì)算而不設(shè)置其結(jié)果，然后重新設(shè)置future為初始化狀態(tài)，如果執(zhí)行遇到異常或者任務(wù)被取消，則不再執(zhí)行此操作。這樣設(shè)計(jì)的目的是：執(zhí)行多次任務(wù)。

我們可以對(duì)比一下runAndReset與run方法，其實(shí)兩者相差不大，主要就是有兩點(diǎn)區(qū)別

  • 1 run()方法里面設(shè)置了result的值，而runAndReset()則移除了這段代碼

下面我們就來看下handlePossibleCancellationInterrupt(int) 這個(gè)方法

五、總結(jié)

FutureTask大部分就簡(jiǎn)單分析完了，其他的自己看下就行了。FutureTask中的任務(wù)狀態(tài)由變量state表示，任務(wù)狀態(tài)都是基于state判斷。而FutureTask的阻塞則是通過自旋+掛起線程實(shí)現(xiàn)的。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容