Future

Future是J.U.C中的一個(gè)接口，它代表著一個(gè)異步執(zhí)行結(jié)果。

Future可以看成線程在執(zhí)行時(shí)留給調(diào)用者的一個(gè)存根，通過(guò)這個(gè)存在可以進(jìn)行查看線程執(zhí)行狀態(tài)(isDone)、取消執(zhí)行(cancel)、阻塞等待執(zhí)行完成并返回結(jié)果(get)、異步執(zhí)行回調(diào)函數(shù)(callback)等操作。

public interface Future<V> {
    /** 取消，mayInterruptIfRunning-false:不允許在線程運(yùn)行時(shí)中斷 **/
    boolean cancel(boolean mayInterruptIfRunning);
    /** 是否取消**/
    boolean isCancelled();
    /** 是否完成 **/
    boolean isDone();
    /** 同步獲取結(jié)果 **/
    V get() throws InterruptedException, ExecutionException;
    /** 同步獲取結(jié)果，響應(yīng)超時(shí) **/
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

用ReentrantLock實(shí)現(xiàn)Future

一個(gè)小栗子：

public class ResponseFuture implements Future<String> {
    private final ResponseCallback callback;
    private String responsed;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();

    public ResponseFuture(ResponseCallback callback) {
        this.callback = callback;
    }
    public boolean isDone() {
        return null != this.responsed;
    }
    public String get() throws InterruptedException, ExecutionException {
        if (!isDone()) {
            try {
                this.lock.lock();
                while (!this.isDone()) {
                    condition.await();
                    if (this.isDone()) break;
                }
            } finally {
                this.lock.unlock();
            }
        }
        return this.responsed;
    }
    // 返回完成
    public void done(String responsed) throws Exception{
        this.responsed = responsed;
        try { 
            this.lock.lock();
            this.condition.signal();
            if(null != this.callback) this.callback.call(this.responsed);
        } finally { 
            this.lock.unlock();
        }
    }
    
    public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        throw new UnsupportedOperationException();
    }
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }
}

使用這個(gè)Future：

public class ResponseCallback {
    public String call(String o) {
        System.out.println("ResponseCallback：處理完成，返回："+o);
        return o;
    }
}

public class FutureTest {
    public static void main(String[] args) {
        final ResponseFuture responseFuture = new ResponseFuture(new ResponseCallback());
        new Thread(new Runnable() {// 請(qǐng)求線程
            public void run() {
                System.out.println("發(fā)送一個(gè)同步請(qǐng)求");
                // System.out.println(responseFuture.get()); 放開這句，就是同步
                System.out.println("接著處理其他事情,過(guò)一會(huì)ResponseCallback會(huì)打印結(jié)果");
            }
        }).start();
        new Thread(new Runnable() {// 處理線程
            public void run() {
                try {
                    Thread.sleep(10000);// 模擬處理一會(huì)
                    responseFuture.done("ok");// 處理完成
                }catch (Exception e) {
                    e.printStackTrace();
                }
        
        }).start();
    }
}

這個(gè)Future使用了Condition的await和signal方法

同步操作：調(diào)用get時(shí)如果發(fā)現(xiàn)未有結(jié)果返回，線程阻塞等待，直到有結(jié)果返回時(shí)通知Future喚醒等待的線程。

異步操作：調(diào)用線性不用等待，有結(jié)果返回時(shí)調(diào)用signal()發(fā)現(xiàn)沒(méi)有等待線程，直接調(diào)用Callback。

FutureTask

FutureTask是J.U.C中Future的實(shí)現(xiàn)，它同時(shí)也實(shí)現(xiàn)了Runnable接口，可以直接在ExecutorService中提交執(zhí)行這個(gè)Future。

一個(gè)小栗子：

public class FutureTaskTest {

    public static class CallableImpl implements Callable<String>{
        public String call() throws Exception {
            String tName = Thread.currentThread().getName();
            System.out.println(tName+" :開始執(zhí)行");
            Thread.sleep(6000);
            return tName+" ok";
        }
    }
    public static void main(String[] args) throws Exception {
        ExecutorService es = Executors.newFixedThreadPool(1);
        
//      // 同步執(zhí)行，等待6000毫秒拿到結(jié)果
//      FutureTask<String> future = new FutureTask<String>(new CallableImpl());
//      es.submit(future);
//      System.out.println(future.get());
        
        // 異步執(zhí)行
        FutureTask<String> future = new FutureTask<String>(new CallableImpl()) {
            protected void done()  { 
                try {
                    System.out.println("異步返回的結(jié)果："+this.get());
                } catch (Exception e) {
                    e.printStackTrace();
                } 
            }
        };
        es.submit(future);
        System.out.println("執(zhí)行其他操作");

        es.shutdown();
    }

}

FutureTask實(shí)現(xiàn)

使用單向鏈表WaitNode表示排隊(duì)等待的線程， volatile state表示當(dāng)前線程的執(zhí)行狀態(tài)，狀態(tài)轉(zhuǎn)換如下：

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

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()) {// 當(dāng)前線程中斷
                removeWaiter(q);// 移除等待的節(jié)點(diǎn)
                throw new InterruptedException();
            }
            int s = state;// 當(dāng)前狀態(tài)
            if (s > COMPLETING) {//已經(jīng)執(zhí)行完成
                if (q != null)
                    q.thread = null;// 節(jié)點(diǎn)綁定的線程置空
                return s;
            }
            else if (s == COMPLETING) //完成中
                Thread.yield();// 出讓CPU
            else if (q == null)// 等待節(jié)點(diǎn)為空，構(gòu)建節(jié)點(diǎn)
                q = new WaitNode();
            else if (!queued)// waiters設(shè)置為q
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            else if (timed) {
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {// 超時(shí)，摘除節(jié)點(diǎn)
                    removeWaiter(q);
                    return state;
                }
                LockSupport.parkNanos(this, nanos);//掛起線程，響應(yīng)超時(shí)
            }
            else
                LockSupport.park(this);// 掛起線程
        }
    }

run

線程執(zhí)行

    // s1
    public void run() {
        // state!=NEW,說(shuō)明沒(méi)必要執(zhí)行，返回。
        // CAS操作，將runner設(shè)置為當(dāng)前線程，如果失敗，說(shuō)明有其他線程在執(zhí)行，返回。
        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();//執(zhí)行
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);// 異常 s2
                }
                if (ran)
                    set(result);// 完成 s3
            }
        } finally {
            runner = null;// 置空runner
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

s1:state不為NEW或者有其他線程在執(zhí)行時(shí)，都不會(huì)執(zhí)行run方法。
執(zhí)行中出現(xiàn)異常轉(zhuǎn)入s2
執(zhí)行結(jié)束轉(zhuǎn)入s3

    // s2
    protected void setException(Throwable t) {
        // 完成 NEW-COMPLETING-EXCEPTIONAL 狀態(tài)置換
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

s2:這步完成狀態(tài)：NEW-COMPLETING-EXCEPTIONAL 置換
將異常賦給結(jié)果outcome
轉(zhuǎn)入s4

    // s3
    protected void set(V v) {
        // 完成NEW-COMPLETING-NORMAL狀態(tài)置換
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

s3:這步完成狀態(tài)：NEW-COMPLETING-NORMAL 置換
將執(zhí)行結(jié)果賦給outcome
轉(zhuǎn)入s4

    // s4
    private void finishCompletion() {
        for (WaitNode q; (q = waiters) != null;) {
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {//waiters置空
                for (;;) {// 依次喚醒所有等待的節(jié)點(diǎn)
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;// 置空
                        LockSupport.unpark(t);// 喚醒
                    }
                    WaitNode next = q.next;//向后探測(cè)
                    if (next == null)
                        break;
                    q.next = null; // unlink to help gc
                    q = next;
                }
                break;
            }
        }
        done();// 回調(diào)鉤子
        callable = null;        // to reduce footprint
    }

s4:for(;;)中異常喚醒等待線程
調(diào)用回調(diào)鉤子done()
被喚醒的線程會(huì)在awaitDone中讓出CPU，退出循環(huán)。

小結(jié)

1、Future代表著一個(gè)異步執(zhí)行結(jié)果
2、get是同步操作，當(dāng)前線程會(huì)阻塞等待執(zhí)行完畢，返回結(jié)果
3、異步操作使用鉤子進(jìn)行回調(diào)，不阻塞當(dāng)前線程