java多线程(4)之FutureTask

FutureTask是一个执行异步任务并可以取消及获得任务结果的实现类，类关系如下:

public class FutureTask<V> implements RunnableFuture<V>{...} public interface RunnableFuture<V> extends Runnable, Future<V> { void run(); } 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; } public interface Runnable { public abstract void run(); }

FutureTask内部实现了异步执行,获取异步任务的结果,判断任务是否完成,是否取消等逻辑处理。

---

FutureTask成员变量说明

变量类型变量名称说明private volatile intstate任务当前的状态private static final intNEW=0任务开始时的初始化状态private static final intCOMPLETING=1任务已经完成,但结果还未给outcomeprivate static final intNORMAL=2任务执行完成private static final intEXCEPTIONAL=3任务执行出现异常private static final intCANCELLED=4任务执行被取消private static final intINTERRUPTING=5任务被中断中private static final intINTERRUPTED=6任务被中断private Callablecallable要执行的任务private Objectoutcome任务执行的结果private volatile Threadrunner任务执行所在的线程volatile WaitNodewaiters获取任务结果的等待线程包装类

FutureTask执行过程中任务状态的转变

NEW -> COMPLETING -> NORMAL:正常执行结束NEW -> COMPLETING -> EXCEPTIONAL:执行中出现异常NEW -> CANCELLED:任务被取消NEW -> INTERRUPTING -> INTERRUPTED:任务被中断

---

FutureTask的构造函数

public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; this.state = NEW; } public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; }

以上的构造方法中都实现了state状态的初始化,第二个构造方法调用下面的方法将runnable转换为callable,第二个形参result接收返回的结果。

public static <T> Callable<T> callable(Runnable task, T result) { if (task == null) throw new NullPointerException(); return new RunnableAdapter<T>(task, result); } private static final class RunnableAdapter<T> implements Callable<T> { private final Runnable task; private final T result; RunnableAdapter(Runnable task, T result) { this.task = task; this.result = result; } public T call() { task.run(); return result; } }

FutureTask中的实现方法

//这是实现的接口Future<V>中的方法: //返回任务是否取消 public boolean isCancelled() { return state >= CANCELLED; } //返回任务是否完成 public boolean isDone() { return state != NEW; } //取消或者中断任务(true为中断，false为取消) public boolean cancel(boolean mayInterruptIfRunning) { if (!(state == NEW && U.compareAndSwapInt(this, STATE, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; try { // in case call to interrupt throws exception if (mayInterruptIfRunning) { try { Thread t = runner; if (t != null) t.interrupt(); } finally { // final state U.putOrderedInt(this, STATE, INTERRUPTED); } } } finally { finishCompletion(); } return true; } //获取任务的执行结果,如果任务还未结束一直处于阻塞状态 public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L);//会在这里阻塞,直到任务执行结束或者出现异常 return report(s);//返回执行的结果 } // 在指定的时间内获取结果,超时会抛出TimeOutException结束,未超时会一直等下去 public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (unit == null) throw new NullPointerException(); int s = state; if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) throw new TimeoutException(); return report(s); } 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); }

FutureTask中子线程执行的run方法

public void run() { if (state != NEW || (如果不是初始化状态就必要执行,直接返回) !U.compareAndSwapObject(this, RUNNER, null, Thread.currentThread())) (CAS方式修改RUNNER线程为当前任务所在执行的线程) return; try { Callable<V> c = callable; if (c != null && state == NEW) { V result; boolean ran; try { result = c.call();//执行callable中的方法 ran = true; } catch (Throwable ex) { result = null; ran = false; setException(ex);//出现异常调用setException } if (ran) set(result);//执行成功设置state状态及结果outcome的值 } } finally { int s = state; if (s >= INTERRUPTING)//如果任务被中断了就执行handlePossibleCancellationInterrupt handlePossibleCancellationInterrupt(s); } } protected void setException(Throwable t) { //(1)首先修改state值为COMPLETING if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) { outcome = t;//(2)为outcome赋值为Throwable对象 //(3)最后修改state值为EXCEPTIONAL U.putOrderedInt(this, STATE, EXCEPTIONAL); // final state finishCompletion(); } } protected void set(V v) { //(1)首先修改state值为COMPLETING if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) { outcome = v;//(2)为outcome赋值为Throwable对象 (3)最后修改state值为EXCEPTIONAL U.putOrderedInt(this, STATE, NORMAL); finishCompletion(); } } //无论任务正常完成还是出现异常都会调用该方法，将等待列表线程唤醒,并清除任务体 private void finishCompletion() { // assert state > COMPLETING; for (WaitNode q; (q = waiters) != null;) { if (U.compareAndSwapObject(this, WAITERS, 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;// 任务体置空 }

至此任务callable的执行已经分析完毕,接下来分析获取任务的结果.

---

public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L);//调用awaitDone return report(s); } /** * @throws CancellationException {@inheritDoc} */ public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (unit == null) throw new NullPointerException(); int s = state; if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) throw new TimeoutException(); return report(s); } //等待任务的执行结果(timed: 是否有时间限制 nanos: 指定的时间) private int awaitDone(boolean timed, long nanos) throws InterruptedException { long startTime = 0L; WaitNode q = null; boolean queued = false; for (;;) { int s = state; if (s > COMPLETING) {//判断状态是否完成,若完成直接返回 if (q != null) q.thread = null; return s; } else if (s == COMPLETING)//任务已经完成,但是set()方法还未执行结束,让出cpu执行权 Thread.yield(); else if (Thread.interrupted()) {//如果线程被中断,删除队列中当前的等待线程,并且抛出InterruptedException异常 removeWaiter(q); throw new InterruptedException(); } else if (q == null) {//q==null,初始化一个当前线程的等待节点 if (timed && nanos <= 0L) return s; q = new WaitNode(); } else if (!queued)//如果当前节点没有加入等待线程列表，就加入 queued = U.compareAndSwapObject(this, WAITERS, q.next = waiters, q); else if (timed) {//如果有时间限制 final long parkNanos; if (startTime == 0L) { // first time startTime = System.nanoTime(); if (startTime == 0L) startTime = 1L; parkNanos = nanos; } else { long elapsed = System.nanoTime() - startTime; if (elapsed >= nanos) {//如果超出指定时间范围就让线程直接返回 removeWaiter(q); return state; } parkNanos = nanos - elapsed; } // nanoTime may be slow; recheck before parking if (state < COMPLETING)//时间范围内让线程阻塞，等待倒计时并被唤醒 LockSupport.parkNanos(this, parkNanos); } else LockSupport.park(this);//如果没有时间限制就让线程阻塞,直到异步任务执行结束被唤醒. } }

ok,大功告成.