Java並發編程原理與實戰三十一:Future&FutureTask 淺析

一、Futrue模式有什么用？
------>正所謂技術來源與生活，這里舉個栗子。在家里，我們都有煮菜的經驗。（如果沒有的話，你們還怎樣來泡女朋友呢？你懂得）。現在女票要你煮四菜一湯，這湯是雞湯，有常識的人雞湯要煲好久滴。如果你先炒四個菜，最后再煲湯，估計都已經一天的時間了。
好了。如果我們先煲湯，在煲湯的時候，同時去炒四個菜。這個模式就是Future模式啦。是不是很簡單。
------》實現Future模式的，我們直接用JDK提供給我們的FutureTask類就可以了。
-----》直接上代碼吧。

// 場景，煮飯，四菜+雞湯。事物安排就是煲雞湯
public class FutureTaskDemo {
public static void main(String[] args) throws Exception {
Callable ca = new Callable() {
@Override
public Integer call() throws Exception {
System.out.println("先煲個雞湯");
Thread.sleep(3000);
return 1;
}
};
FutureTask future = new FutureTask<>(ca);
new Thread(future).start();;
System.out.println("干其他事情先。。。");
System.out.println("雞湯煲好了嗎？     "+future.get());
}
}

FutureTask 是實現了RunnableFuture<>接口，而RunnablFuture接口繼承了Runnable和Future接口.

----》相信運行過代碼的朋友，都懂是什么回事啦。

 

二、Future設計模式實現（實現類似於JDK提供的Future）

1、Future可以讓我們提前處理一些復雜的運行。非常的方便。
這里，我們自己去實現一個Future！！！！是不是很興奮。
本人編程的思想一般都是來自於生活。從問題出發，再去找到對應的解決方法。而不是先學習大量無關的解決問題的方法，再去解決問題（當然要辯證去看這個問題）。。。不知道有無人看不懂這句話呢？
------》設計目的：設計一個方法，能在未來某一個時間點，拿出一個結果。如果在未來的那個時刻拿不出來，那就讓線程去等待，一直等到拿個結果（無論成功還是出錯的結果）才開始執行另外的事情。

=====================================================
// 場景：一天，我要去訂一個蛋糕，去到蛋糕店，下了個訂單，我就去上班了，下班之后，拿到蛋糕，回家吃啦。
// 面向對象思想，蛋糕店，使用工廠模式的對象。客人，一個對象。產品，里面有訂單的對象。
// ，去到蛋糕店，拿到蛋糕店給的訂單，使用工廠模式來模擬這個工廠，
public class MainTest {
public static void main(String[] args) {
Factory pf = new Factory(); // 創建一個蛋糕店
Future f = pf.createOrder("蛋糕"); // 蛋糕店生成什么產品，返回一個生成什么的訂單
System.out.println("我去上班了,下班回來拿蛋糕。。。");
// 根據訂單拿到做好的蛋糕，如果沒做完就要等待蛋糕店做完。
System.out.println("下班去拿蛋糕，拿到蛋糕了"+f.getProduct());
}
}

// Factory相當於蛋糕店，接受訂單，開始生產蛋糕
public class Factory {
// 提供客人一個訂單
public Future createOrder(String name) {
Future f = new Future();   // 創建一個訂單
new Thread(new Runnable() {
@Override
public void run() {
System.out.println("開始生產蛋糕");
Product product = new Product(name); // 生成產品
System.out.println("生產蛋糕結束");
f.setProduct(product);
}
}).start();
return f;
}
}

public class Product {

private int id; // 訂單編號
private String name; // 生成的產品名稱
public Product(String name) {
super();
Random random = new Random();
this.id = random.nextInt();
this.name = name;
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}

@Override
public String toString() {
return "Product [id=" + id + ", name=" + name + "]";
}
}

public class Future {
private Product product;
private boolean flag;   // true:生產完了，false：還沒生產完
public synchronized void setProduct(Product product) {
this.product=product;
if(flag) { // 如果生產完，直接返回就可以，不用再重新設置了。
return;
}
flag=true;
notifyAll(); // 叫醒所有處於wait狀態的線程
}

public synchronized Product getProduct() {
while(!flag) { // 當還沒有生產完的時候，就要等待。
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
return product;
}

@Override
public String toString() {
return "Future [product=" + product + "]";
}
}

三、Future源碼解讀

1、Future API以及應用場景
2、Future的核心實現原理
3、Future的源碼

1、怎樣使用Future API呢？
-----》如果一問到這個問題，可以馬上向起這三點。
1.1 new 一個Callable
1.2 將剛剛創建的Callable的實例交給FutureTask
1.3 將剛剛創建的FutureTask實例交給Thread

2、如果有看過源碼的朋友，都會問一個問題，就是Callable和Runnable有什么區別？
-----》Runnable的run方法是被線程調用的，在run方法時異步執行的
Callable的Call方法，不是異步執行的，是由Future的run方法調用的

接口Callable：有返回結果並且可能拋出異常的任務； 
接口Runnable：沒有返回結果 
接口Future：表示異步執行的結果； 
類FutureTask：實現Future、Runnable等接口，是一個異步執行的任務。可以直接執行，或包裝成Callable執行；

具體詳細可以參考：https://blog.csdn.net/lican19911221/article/details/78200344

四、再次理解Future模式

先上一個場景：假如你突然想做飯，但是沒有廚具，也沒有食材。網上購買廚具比較方便，食材去超市買更放心。
實現分析：在快遞員送廚具的期間，我們肯定不會閑着，可以去超市買食材。所以，在主線程里面另起一個子線程去網購廚具。
但是，子線程執行的結果是要返回廚具的，而run方法是沒有返回值的。所以，這才是難點，需要好好考慮一下。
模擬代碼1：

package test;

public class CommonCook {

    public static void main(String[] args) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        // 第一步 網購廚具
        OnlineShopping thread = new OnlineShopping();
        thread.start();
        thread.join();  // 保證廚具送到
        // 第二步 去超市購買食材
        Thread.sleep(2000);  // 模擬購買食材時間
        Shicai shicai = new Shicai();
        System.out.println("第二步：食材到位");
        // 第三步 用廚具烹飪食材
        System.out.println("第三步：開始展現廚藝");
        cook(thread.chuju, shicai);
        
        System.out.println("總共用時" + (System.currentTimeMillis() - startTime) + "ms");
    }
    
    // 網購廚具線程
    static class OnlineShopping extends Thread {
        
        private Chuju chuju;

        @Override
        public void run() {
            System.out.println("第一步：下單");
            System.out.println("第一步：等待送貨");
            try {
                Thread.sleep(5000);  // 模擬送貨時間
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("第一步：快遞送到");
            chuju = new Chuju();
        }
        
    }

    //  用廚具烹飪食材
    static void cook(Chuju chuju, Shicai shicai) {}
    
    // 廚具類
    static class Chuju {}
    
    // 食材類
    static class Shicai {}
}

運行結果：

第一步：下單
第一步：等待送貨
第一步：快遞送到
第二步：食材到位
第三步：開始展現廚藝
總共用時7013ms

可以看到，多線程已經失去了意義。在廚具送到期間，我們不能干任何事。對應代碼，就是調用join方法阻塞主線程。
有人問了，不阻塞主線程行不行？？？
不行！！！
從代碼來看的話，run方法不執行完，屬性chuju就沒有被賦值，還是null。換句話說，沒有廚具，怎么做飯。
Java現在的多線程機制，核心方法run是沒有返回值的；如果要保存run方法里面的計算結果，必須等待run方法計算完，無論計算過程多么耗時。
面對這種尷尬的處境，程序員就會想：在子線程run方法計算的期間，能不能在主線程里面繼續異步執行？？？
Where there is a will，there is a way！！！
這種想法的核心就是Future模式，下面先應用一下Java自己實現的Future模式。

模擬代碼2：

package test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class FutureCook {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        long startTime = System.currentTimeMillis();
        // 第一步 網購廚具
        Callable<Chuju> onlineShopping = new Callable<Chuju>() {

            @Override
            public Chuju call() throws Exception {
                System.out.println("第一步：下單");
                System.out.println("第一步：等待送貨");
                Thread.sleep(5000);  // 模擬送貨時間
                System.out.println("第一步：快遞送到");
                return new Chuju();
            }
            
        };
        FutureTask<Chuju> task = new FutureTask<Chuju>(onlineShopping);
        new Thread(task).start();
        // 第二步 去超市購買食材
        Thread.sleep(2000);  // 模擬購買食材時間
        Shicai shicai = new Shicai();
        System.out.println("第二步：食材到位");
        // 第三步 用廚具烹飪食材
        if (!task.isDone()) {  // 聯系快遞員，詢問是否到貨
            System.out.println("第三步：廚具還沒到，心情好就等着（心情不好就調用cancel方法取消訂單）");
        }
        Chuju chuju = task.get();
        System.out.println("第三步：廚具到位，開始展現廚藝");
        cook(chuju, shicai);
        
        System.out.println("總共用時" + (System.currentTimeMillis() - startTime) + "ms");
    }
    
    //  用廚具烹飪食材
    static void cook(Chuju chuju, Shicai shicai) {}
    
    // 廚具類
    static class Chuju {}
    
    // 食材類
    static class Shicai {}

}

運行結果：

第一步：下單
第一步：等待送貨
第二步：食材到位
第三步：廚具還沒到，心情好就等着（心情不好就調用cancel方法取消訂單）
第一步：快遞送到
第三步：廚具到位，開始展現廚藝
總共用時5005ms

可以看見，在快遞員送廚具的期間，我們沒有閑着，可以去買食材；而且我們知道廚具到沒到，甚至可以在廚具沒到的時候，取消訂單不要了。

好神奇，有沒有。

下面具體分析一下第二段代碼：

1）把耗時的網購廚具邏輯，封裝到了一個Callable的call方法里面。

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

Callable接口可以看作是Runnable接口的補充，call方法帶有返回值，並且可以拋出異常。

2）把Callable實例當作參數，生成一個FutureTask的對象，然后把這個對象當作一個Runnable，作為參數另起線程。

public class FutureTask<V> implements RunnableFuture<V>

public interface RunnableFuture<V> extends Runnable, Future<V>

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。Future的核心思想是：一個方法f，計算過程可能非常耗時，等待f返回，顯然不明智。可以在調用f的時候，立馬返回一個Future，可以通過Future這個數據結構去控制方法f的計算過程。

這里的控制包括：

get方法：獲取計算結果（如果還沒計算完，也是必須等待的）

cancel方法：還沒計算完，可以取消計算過程

isDone方法：判斷是否計算完

isCancelled方法：判斷計算是否被取消

這些接口的設計很完美，FutureTask的實現注定不會簡單，后面再說。

3）在第三步里面，調用了isDone方法查看狀態，然后直接調用task.get方法獲取廚具，不過這時還沒送到，所以還是會等待3秒。對比第一段代碼的執行結果，這里我們節省了2秒。這是因為在快遞員送貨期間，我們去超市購買食材，這兩件事在同一時間段內異步執行。

通過以上3步，我們就完成了對Java原生Future模式最基本的應用。下面具體分析下FutureTask的實現，先看JDK8的，再比較一下JDK6的實現。

既然FutureTask也是一個Runnable，那就看看它的run方法

public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable; // 這里的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); // 保存call方法拋出的異常
                }
                if (ran)
                    set(result); // 保存call方法的執行結果
            }
        } 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);
        }
    }

先看try語句塊里面的邏輯，發現run方法的主要邏輯就是運行Callable的call方法，然后將保存結果或者異常（用的一個屬性result）。這里比較難想到的是，將call方法拋出的異常也保存起來了。

這里表示狀態的屬性state是個什么鬼

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

把FutureTask看作一個Future，那么它的作用就是控制Callable的call方法的執行過程，在執行的過程中自然會有狀態的轉換：

1）一個FutureTask新建出來，state就是NEW狀態；COMPETING和INTERRUPTING用的進行時，表示瞬時狀態，存在時間極短(為什么要設立這種狀態？？？不解)；NORMAL代表順利完成；EXCEPTIONAL代表執行過程出現異常；CANCELED代表執行過程被取消；INTERRUPTED被中斷

2）執行過程順利完成：NEW -> COMPLETING -> NORMAL

3）執行過程出現異常：NEW -> COMPLETING -> EXCEPTIONAL

4）執行過程被取消：NEW -> CANCELLED

5）執行過程中，線程中斷：NEW -> INTERRUPTING -> INTERRUPTED

代碼中狀態判斷、CAS操作等細節，請讀者自己閱讀。

再看看get方法的實現：

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

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

get方法的邏輯很簡單，如果call方法的執行過程已完成，就把結果給出去；如果未完成，就將當前線程掛起等待。awaitDone方法里面死循環的邏輯，推演幾遍就能弄懂；它里面掛起線程的主要創新是定義了WaitNode類，來將多個等待線程組織成隊列，這是與JDK6的實現最大的不同。

掛起的線程何時被喚醒：

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
    }

以上就是JDK8的大體實現邏輯，像cancel、set等方法，也請讀者自己閱讀。

再來看看JDK6的實現。

JDK6的FutureTask的基本操作都是通過自己的內部類Sync來實現的，而Sync繼承自AbstractQueuedSynchronizer這個出鏡率極高的並發工具類

/** State value representing that task is running */
        private static final int RUNNING   = 1;
        /** State value representing that task ran */
        private static final int RAN       = 2;
        /** State value representing that task was cancelled */
        private static final int CANCELLED = 4;

        /** The underlying callable */
        private final Callable<V> callable;
        /** The result to return from get() */
        private V result;
        /** The exception to throw from get() */
        private Throwable exception;

里面的狀態只有基本的幾個，而且計算結果和異常是分開保存的。

 V innerGet() throws InterruptedException, ExecutionException {
            acquireSharedInterruptibly(0);
            if (getState() == CANCELLED)
                throw new CancellationException();
            if (exception != null)
                throw new ExecutionException(exception);
            return result;
        }

這個get方法里面處理等待線程隊列的方式是調用了acquireSharedInterruptibly方法，看過我之前幾篇博客文章的讀者應該非常熟悉了。其中的等待線程隊列、線程掛起和喚醒等邏輯，這里不再贅述，如果不明白，請出門左轉。

最后來看看，Future模式衍生出來的更高級的應用。

再上一個場景：我們自己寫一個簡單的數據庫連接池，能夠復用數據庫連接，並且能在高並發情況下正常工作。

實現代碼1：

package test;

import java.util.concurrent.ConcurrentHashMap;

public class ConnectionPool {

    private ConcurrentHashMap<String, Connection> pool = new ConcurrentHashMap<String, Connection>();
    
    public Connection getConnection(String key) {
        Connection conn = null;
        if (pool.containsKey(key)) {
            conn = pool.get(key);
        } else {
            conn = createConnection();
            pool.putIfAbsent(key, conn);
        }
        return conn;
    }
    
    public Connection createConnection() {
        return new Connection();
    }
    
    class Connection {}
}

我們用了ConcurrentHashMap，這樣就不必把getConnection方法置為synchronized(當然也可以用Lock)，當多個線程同時調用getConnection方法時，性能大幅提升。

貌似很完美了，但是有可能導致多余連接的創建，推演一遍：

某一時刻，同時有3個線程進入getConnection方法，調用pool.containsKey(key)都返回false，然后3個線程各自都創建了連接。雖然ConcurrentHashMap的put方法只會加入其中一個，但還是生成了2個多余的連接。如果是真正的數據庫連接，那會造成極大的資源浪費。

所以，我們現在的難點是：如何在多線程訪問getConnection方法時，只執行一次createConnection。

結合之前Future模式的實現分析：當3個線程都要創建連接的時候，如果只有一個線程執行createConnection方法創建一個連接，其它2個線程只需要用這個連接就行了。再延伸，把createConnection方法放到一個Callable的call方法里面，然后生成FutureTask。我們只需要讓一個線程執行FutureTask的run方法，其它的線程只執行get方法就好了。

上代碼：

package test;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class ConnectionPool {

    private ConcurrentHashMap<String, FutureTask<Connection>> pool = new ConcurrentHashMap<String, FutureTask<Connection>>();

    public Connection getConnection(String key) throws InterruptedException, ExecutionException {
        FutureTask<Connection> connectionTask = pool.get(key);
        if (connectionTask != null) {
            return connectionTask.get();
        } else {
            Callable<Connection> callable = new Callable<Connection>() {
                @Override
                public Connection call() throws Exception {
                    return createConnection();
                }
            };
            FutureTask<Connection> newTask = new FutureTask<Connection>(callable);
            connectionTask = pool.putIfAbsent(key, newTask);
            if (connectionTask == null) {
                connectionTask = newTask;
                connectionTask.run();
            }
            return connectionTask.get();
        }
    }

    public Connection createConnection() {
        return new Connection();
    }

    class Connection {
    }
}

 推演一遍：當3個線程同時進入else語句塊時，各自都創建了一個FutureTask，但是ConcurrentHashMap只會加入其中一個。第一個線程執行pool.putIfAbsent方法后返回null，然后connectionTask被賦值，接着就執行run方法去創建連接，最后get。后面的線程執行pool.putIfAbsent方法不會返回null，就只會執行get方法。

在並發的環境下，通過FutureTask作為中間轉換，成功實現了讓某個方法只被一個線程執行。

五、區別

Future是一個接口，FutureTask是Future的一個實現類，並實現了Runnable，因此FutureTask可以傳遞到線程對象Thread中新建一個線程執行。所以可以通過Excutor(線程池)來執行，也可傳遞給Thread對象執行。
如果在主線程中需要執行比較耗時的操作，但又不想阻塞主線程時，可以把這些作業交給Future對象在后台完成，當主線程將來需要時，就可以通過Future對象獲得后台作業的計算結果或者執行狀態。
FutureTask是為了彌補Thread的不足而設計的，它可以讓程序員准確地知道線程什么時候執行完成並獲得到線程執行完成后返回的結果（如果有需要）。
FutureTask是一種可以取消的異步的計算任務。它的計算是通過Callable實現的，它等價於可以攜帶結果的Runnable，並且有三個狀態：等待、運行和完成。完成包括所有計算以任意的方式結束，包括正常結束、取消和異常。
Executor框架利用FutureTask來完成異步任務，並可以用來進行任何潛在的耗時的計算。一般FutureTask多用於耗時的計算，主線程可以在完成自己的任務后，再去獲取結果。

 

參考文獻：

https://www.cnblogs.com/cz123/p/7693064.html

《java並發編程實戰》龍果學院