一、Callable和Future

對比Callable和Runnable：

Runnable介面：

public interface Runnable {
    public abstract void run();
}

Callable介面:

public interface Callable<V> {
    V call() throws Exception;
}

兩者的不同在於:

1、Runnable介面的run()方法沒有返回值，而Callable介面的call()方法是帶有泛型的返回值。

2、Runnable方法的run()方法的異常只能在內部處理，而不能向上拋，而Callable的call()方法允許丟擲異常。

Future:表示非同步計算的結果，它提供了檢查計算是否完成的方法，以等待計算的完成，並接收計算的結果。Future的cancel()方法可以取消任務的執行，它有一布林引數，引數為 true 表示立即中斷任務的執行，引數為 false 表示允許正在執行的任務執行完成。Future的 get() 方法等待計算完成，獲取計算結果。

Callable的Demo:

public class TestCallable {
    public static void main(String[] args) {
        ThreadDemo td = new ThreadDemo();
        FutureTask task = new FutureTask(td);
        new Thread(task).start();
        try {
Thread.sleep(1000);//執行其他操作
            System.out.println(task.get());//等待計算結果
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
class ThreadDemo implements Callable<Integer>{
    @Override
    public Integer call() throws Exception {
        int sum = 0;
        for(int i = 0;i<=100;i++){
            sum += i;
        }
        return sum;
    }
}
 

二、FutrueTask原始碼分析:

UML圖:

FutrueTask的構造方法：

public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        sync = new Sync(callable);
}

public FutureTask(Runnable runnable, V result) {  //把Runnable適配成Callable
        sync = new Sync(Executors.callable(runnable, result));
}
 

//下面是任務的四個狀態值,使用AQS的state來表示，預設為0
private static final int READY     = 0;   //任務準備執行
        private static final int RUNNING   = 1;  //任務正在執行
        private static final int RAN       = 2;   //已經執行完畢
        private static final int CANCELLED = 4;    //任務被取消

        private final Callable<V> callable;    
//get()方法得到的結果
        private V result;
        //get()方法丟擲的異常
private Throwable exception;

        //當前任務執行的執行緒物件
        private volatile Thread runner;

FutureTask的get()方法:

public V get() throws InterruptedException, ExecutionException {
        return sync.innerGet();
}

V innerGet() throws InterruptedException, ExecutionException {
            acquireSharedInterruptibly(0);   //AQS共享模式的可中斷的獲取資源的方法，不過引數為0
            if (getState() == CANCELLED)  //任務被取消，丟擲異常
                throw new CancellationException();
            if (exception != null)   //call方法有異常丟擲
                throw new ExecutionException(exception);
            return result;
        }

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)  //任務是否完成
            doAcquireSharedInterruptibly(arg);//任務沒完成，排隊等待
}

protected int tryAcquireShared(int ignore) {
            return innerIsDone() ? 1 : -1;
        }

boolean innerIsDone() {
            return ranOrCancelled(getState()) && runner == null;//狀態為RAN或CANCELLED，且當前沒有要執行的執行緒物件
        }

public void run() {
        sync.innerRun();
}

void innerRun() {
            if (!compareAndSetState(READY, RUNNING)) //CAS設定狀態，預期值READY，更新值RUNING，設定失敗函式直接返回
                return;

            runner = Thread.currentThread(); //拿到當前執行緒
            if (getState() == RUNNING) { // 上面把state設定成了RUNNING，這裡重新檢查
                V result;
                try {
                    result = callable.call(); //呼叫Callable的call方法取到返回值
                } catch (Throwable ex) {
                    setException(ex);
                    return;
                }
                set(result);   //將返回值設定給result，並設定狀態
            } else {
                releaseShared(0); //這邊是CANCELLED狀態
            }
        }

protected void set(V v) {
        sync.innerSet(v);
    }

void innerSet(V v) {
            for (;;) {   //自旋
                int s = getState(); //拿到任務狀態
                if (s == RAN)
                    return;
                if (s == CANCELLED) {
                    releaseShared(0);
                    return;
                }
                if (compareAndSetState(s, RAN)) {//CAS設定狀態，預期值RUNNING，更新值RAN
                    result = v;
                    releaseShared(0);
                    done(); //done方法為空方法可以重寫，相當於一個回撥函式
                    return;
                }
            }
        }

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

protected boolean tryReleaseShared(int ignore) {
            runner = null;  //將當前任務執行的執行緒物件設定為null
            return true;
        }