JDK future框架，提供了一種非同步程式設計模式，基於執行緒池的。將任務runnable/callable提交到執行緒池executor，返回一個Future物件。通過future.get()獲取執行結果，這裡提交到執行緒池，後面的操作不會阻塞。future.get()獲取結果會阻塞，其實也是用多線執行緒執行任務。

future.get()這裡會阻塞，google的guava提供了一個calllback解決辦法，這也是我準備看的

下面是一個future的demo

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @Author: <[email protected]>
 * @Description: jdk future demo
 * @Date: Created in : 2018/11/30 6:01 PM
 **/
public class TestJdkFuture {

  public static void main(String[] args) throws
 
 ExecutionException, InterruptedException {
    testJdkFuture();
  }


  public static void testJdkFuture() throws ExecutionException, InterruptedException {
    Callable<Integer> callable = () ->{
      System.out.println("callable do some compute");
      return 1;
    } ;
    Runnable runnable 
 
= () -> {
      System.out.println("runable do some compute");
    };
    ExecutorService executorService = Executors.newCachedThreadPool();
    Future<Integer> future = executorService.submit(callable);
    Future runableFuture = executorService.submit(runnable);
    Object runableRes = runableFuture.get();
    int res = future.get();
    executorService.shutdown();
    System.out.println("callable res: " + res);
    System.out.println("runnable res: " + runableRes);

  }
}

demo裡面提交了一個Callable和一個Runnable到執行緒池，通過future獲取計算結果

executorService.submit(callable)原始碼

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

這裡主要是封裝了一個RunnableFuture和提交任務到執行緒池

我們看下RunnableFuture裡面的run方法，因為執行緒池執行任務，是執行run方法。看FutureTask裡面的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);
        }
    }

首先，設定runner為執行緒池中的當前執行緒，後面執行call()方法，計算出結果，set(result)。跟進set(result)

/**
     * 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) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

這裡用了一個cas設定FutureTask的state欄位為COMPLETING，完成中的一個狀態。接著設定outcom為計算結果，我們跟進finishCompletion()

/**
     * Removes and signals all waiting threads, invokes done(), and
     * nulls out callable.
     */
    private void finishCompletion() {
        // assert state > COMPLETING;
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (;;) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }

        done();

        callable = null;        // to reduce footprint
    }

WaitNode主要是一個阻塞執行緒連結串列，即呼叫future.get()方法的執行緒連結串列。這裡主要作用是注意喚醒這些執行緒，通過LockSupport.unpark(t)喚醒。這裡用阻塞執行緒連結串列，主要是考慮到可能有多個執行緒會呼叫future.get()阻塞

ok,到這裡執行任務，把計算結果放到future中，並喚醒阻塞執行緒已經理清楚了

我們再來看下，future.get()是如何實現阻塞，和獲取到計算結果的

進入future.get()

/**
     * Waits if necessary for the computation to complete, and then
     * retrieves its result.
     *
     * @return the computed result
     * @throws CancellationException if the computation was cancelled
     * @throws ExecutionException if the computation threw an
     * exception
     * @throws InterruptedException if the current thread was interrupted
     * while waiting
     */
    V get() throws InterruptedException, ExecutionException;

Future.java是一個介面，這裡看註釋可以看出，會阻塞等待結果計算完成。我們看下一個實現FutureTask.java的get()方法

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

awaitDone(false, 0l)主要是阻塞執行緒，進入方法

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

這裡，主要是阻塞執行緒，把當前執行緒放到阻塞執行緒連結串列中，通過LockSupport.park(this)阻塞當前執行緒，等待執行緒池裡面的執行緒喚醒。喚醒之後，回到get()方法，看report()方法

/**
     * 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;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }

這裡主要是返回計算結果給阻塞執行緒

到這裡，基本理清楚了，future的阻塞實現，以及獲取計算結果的步驟。

future框架 = 執行緒池 + futrue