咱们都晓得线程创立的形式有以下几种

  • Thread
  • Runnable
  • Callable
  • Executors

其中Callable是能获取到返回值或者异样的，callable接口如下

@FunctionalInterfacepublic 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必须要和线程池搭配应用，或者应用FutureTask，具体用法如下

public interface Future<V> {    //勾销工作    boolean cancel(boolean mayInterruptIfRunning);    //是否被勾销    boolean isCancelled();       //是否实现，已实现返回true    boolean isDone();      //阻塞获取返回值，如果有异样则抛出异样    V get() throws InterruptedException, ExecutionException;       //指定阻塞工夫获取返回值    V get(long timeout, TimeUnit unit)        throws InterruptedException, ExecutionException, TimeoutException;}

实现类RunnableFuture,该类实现了Runnable接口和Future接口

public interface RunnableFuture<V> extends Runnable, Future<V> {    //将此Future设置为其计算的后果，除非工作勾销    void run();}

实现类FutureTask（重点)

future接口实现个别应用线程池或者FutureTask实现线程调用

//构造方法传入callable赋值给成员变量callable,并设置state为新建public FutureTask(Callable<V> callable) {    if (callable == null)        throw new NullPointerException();    this.callable = callable;    this.state = NEW;       // ensure visibility of callable}

state有以下几个状态

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有下几个成员变量

构造方法1

//构造方法传入的callable对象private Callable<V> callable;//线程执行完的返回后果private Object outcome; //以后运行的线程private volatile Thread runner;//waiters示意如果多个线程执行一个callable对象，则会存在一个单向链表中private volatile WaitNode waiters;

构造方法2

//带返回值的构造方法public FutureTask(Runnable runnable, V result) {    this.callable = Executors.callable(runnable, result);    this.state = NEW;      }

因为FutureTask实现了Runnable接口，所以线程调度时执行FutureTask的run办法

public void run() {      //如果状态不为NEW或无奈将以后线程设置进去，为runnerOffeset，则返回    if (state != NEW ||        !UNSAFE.compareAndSwapObject(this, runnerOffset,                                     null, Thread.currentThread()))        return;    try {        Callable<V> c = callable;          //如果FutureTask内的Callable不为空且为新建状态，则执行if外部办法        if (c != null && state == NEW) {            V result;            boolean ran;            try {                  //执行callale的call办法，且设置ran=true示意执行胜利                result = c.call();                ran = true;            } catch (Throwable ex) {                result = null;                  //设置ran=false示意执行失败                ran = false;                  //如果捕捉到异样，则传入异样                setException(ex);            }              //如果是执行胜利，通过set办法设置后果            if (ran)                  //将后果设置给set办法                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);    }}

将callable的返回值通过cas设置进去，并

protected void set(V v) {    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {        outcome = v;          //将状态设置失常NORMAL        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state          //实现赋值后唤醒UnPack get的线程        finishCompletion();    }}

如果call办法出现异常，则通过cas设置胜利后进行唤醒

protected void setException(Throwable t) {    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {        outcome = t;        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state        finishCompletion();    }}

实现后进行唤醒

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;                      //唤醒get线程，且将前面所有执行的waitNode节点                    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}
//是否勾销,当state>=勾销状态,即等于CANCELLED,INTERRUPTING或者INTERRUPTED时返回truepublic boolean isCancelled() {    return state >= CANCELLED;}//是否实现,当线程不为NEW状态都视为已实现(包含失常实现,异样实现,以及中断等)public boolean isDone() {    return state != NEW;}
//阻塞获取返回后果public V get() throws InterruptedException, ExecutionException {    int s = state;     //当线程小于等于COMPLETING(NEW新建或者COMPLETING筹备实现)时,进行无工夫闲置的阻塞    if (s <= COMPLETING)        s = awaitDone(false, 0L);  //此处只有当后果返回时下面的阻塞才会被唤醒,否则始终阻塞在下面的if中    return report(s);}
//阻塞指定工夫获取返回后果,当指定工夫仍旧未获取到后果,抛出TimeoutException异样public V get(long timeout, TimeUnit unit)    throws InterruptedException, ExecutionException, TimeoutException {    if (unit == null)          //传入的工夫单位如果为空，则抛空指针异样        throw new NullPointerException();    int s = state;  //当后果未返回且等待时间大于指定的timeout工夫,抛出timeout异样    if (s <= COMPLETING &&        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)        throw new TimeoutException();    //此处只有当后果返回时下面的阻塞才会被唤醒,否则始终阻塞在下面的if中    return report(s);}

report是依据以后的state判断call办法是执行失常还是失败，失常则返回泛型的result,如果是勾销状态，则抛出勾销异样，否则抛出ExecutionException异样示意执行异样

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

阻塞办法