Java FutureTask 源码分析 Android上的实现

FutureTask类提供了可取消的异步计算,并且可以利用开始和取消计算的方法、查询计算是否完成的方法和获取计算结果的方法。

首先看一下继承关系

 

public class FutureTask<V> implements RunnableFuture<V>

 

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run(); }

FutureTask -> RunnableFuture -> Runnable,Future  Runnable我们都知道就不说了,主要看一下Future接口。

public interface Future<V> { //试图取消对此任务的执行。
    boolean cancel(boolean mayInterruptIfRunning); //如果在任务正常完成前将其取消,则返回 true。
    boolean isCancelled(); //如果任务已完成,则返回 true。
    boolean isDone(); //如有必要,等待计算完成,然后获取其结果。
    V get() throws InterruptedException, ExecutionException; //如有必要,最多等待为使计算完成所给定的时间之后,获取其结果(如果结果可用)。
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }

 

也就是说当FutureTask的实例被一个线程执行以后,我们可以对它进行控制,比如终止任务,查询任务是否完成,获取结果等。

 63 public class FutureTask<V> implements RunnableFuture<V> {  64 
 65     /**
 66  * 这个类可能的几种过度状态  67  * NEW -> COMPLETING -> NORMAL 新建-正在完成-正常完成  68  * NEW -> COMPLETING -> EXCEPTIONAL 新建-正在完成-异常  69  * NEW -> CANCELLED 新建-取消  70  * NEW -> INTERRUPTING -> INTERRUPTED 新建-正在中断线程-已经中断  71      */
 72     private volatile int state; //当前状态 注意volatile,因为这个类是为多线程使用而设计的
 73     private static final int NEW          = 0;  74     private static final int COMPLETING   = 1;  75     private static final int NORMAL       = 2;  76     private static final int EXCEPTIONAL  = 3;  77     private static final int CANCELLED    = 4;  78     private static final int INTERRUPTING = 5;  79     private static final int INTERRUPTED  = 6;  80 
 81     /** 用于执行任务并且返回结果 */
 82     private Callable<V> callable;  83     /** 用于存放返回结果 另外如果出现异常那么就存放异常的引用 */
 84     private Object outcome; // non-volatile, protected by state reads/writes
 85     /** 当前执行该类的线程 */
 86     private volatile Thread runner;  87     /** 等待锁队节点 */
 88     private volatile WaitNode waiters;  89 
 90     /**
 91  * 根据state的值,判断任务是否正常完成。  92  *  93  * @param s 当前的状态值  94      */
 95     @SuppressWarnings("unchecked")  96     private V report(int s) throws ExecutionException {  97         Object x = outcome;  98         //如果状态为NORMAL 则正常返回结果。
 99         if (s == NORMAL) 100             return (V)x; 101         //如果状态>=CANCELLED 则抛出一个异常CancellationException 告诉用户任务已经终止。
102         if (s >= CANCELLED) 103             throw new CancellationException(); 104         //否则就是任务执行过程中出现异常,包装一下直接抛出。
105         throw new ExecutionException((Throwable)x); 106  } 107 
108     /**
109  * 构造参数 传入callable对象,并且将当前状态state设置为null。 110      */
111     public FutureTask(Callable<V> callable) { 112         if (callable == null) 113             throw new NullPointerException(); 114         this.callable = callable; 115         this.state = NEW;       // ensure visibility of callable
116  } 117 
118     /**
119  * 传入runnable对象和result 120  * 通过Executors包装一下,还是返回一个callable的实例 121  * 在执行完runnable的任务后直接返回result 122      */
123     public FutureTask(Runnable runnable, V result) { 124         this.callable = Executors.callable(runnable, result); 125         this.state = NEW;       // ensure visibility of callable
126  } 127 
128     /*
129  * 判断任务是否取消 130      */
131     public boolean isCancelled() { 132         return state >= CANCELLED; 133  } 134 
135     /*
136  * 如果任务已完成,则返回 true。 137  * 可能由于正常终止、异常或取消而完成,在所有这些情况中,此方法都将返回 true。 138      */
139     public boolean isDone() { 140         return state != NEW; 141  } 142 
143     /**
144  * 取消任务 145  * mayInterruptIfRunning 如果应该中断执行此任务的线程,则为 true;否则允许正在运行的任务运行完成 146      */
147     public boolean cancel(boolean mayInterruptIfRunning) { 148         //如果当前状态不是新创建时的状态 那么返回false 149         //这种情况说明任务已经执行,那么当前状态的值就有俩种情况。 150         //一种是任务正常完成 那么当前状态将的值会按照这个方向执行 NEW -> COMPLETING -> NORMAL 151         //一种任务是执行中出现异常 那么当前状态的值会按照这个方向执行 NEW -> COMPLETING -> EXCEPTIONAL
152         if (state != NEW) 153             return false; 154         
155         if (mayInterruptIfRunning) { 156             //设置当前状态为INTERRUPTING(正在设置线程的中断状态)
157             if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING)) 158                 return false; 159             Thread t = runner; 160             if (t != null) 161                 //设置线程的中断状态
162  t.interrupt(); 163             //设置当前状态为INTERRUPTED
164             UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
165  } 166         //如果当前状态为NEW 那么会把当前状态设置为 CANCELLED 然后调用finishCompletion,然后返回成功
167         else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED)) 168             return false; 169         //这个方法下面解释
170  finishCompletion(); 171         return true; 172  } 173 
174     /**
175  * 获取任务的结果,如果任务未完成则线程进入等待状态。 176      */
177     public V get() throws InterruptedException, ExecutionException { 178         int s = state; 179         //如果任务未完成
180         if (s <= COMPLETING) 181             //等待完成 里面的是实现是等待队列
182             s = awaitDone(false, 0L); 183         //调用report方法来根据情况返回结果
184         return report(s); 185  } 186 
187     /**
188  * 最多等待为使计算完成所给定的时间之后,获取其结果。 189      */
190     public V get(long timeout, TimeUnit unit) 191         throws InterruptedException, ExecutionException, TimeoutException { 192         if (unit == null) 193             throw new NullPointerException(); 194         int s = state; 195         //如果超时任务未能完成就抛出TimeoutException
196         if (s <= COMPLETING &&
197             (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) 198             throw new TimeoutException(); 199         //调用report方法来根据情况返回结果
200         return report(s); 201  } 202 
203     /**
204  * 给子类扩展的方法,state状态发生改变时被调用。 205      */
206     protected void done() { } 207 
208     protected void set(V v) { 209         //将状态先设置为COMPLETING正在完成,然后再设置为正常NORMAL
210         if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 211             outcome = v; 212             UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state 213             //然后调用完成方法
214  finishCompletion(); 215  } 216  } 217 
218     protected void setException(Throwable t) { 219         //将状态先设置为COMPLETING正在完成,然后再设置为异常EXCEPTIONAL
220         if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 221             outcome = t; 222             UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state 223             //然后调用完成方法
224  finishCompletion(); 225  } 226  } 227 
228     //任务开始方法
229     public void run() { 230         //任务不能多个线程同时运行。 231         //如果第一次运行将runner设置为当前执行的线程。
232         if (state != NEW ||
233             !UNSAFE.compareAndSwapObject(this, runnerOffset, 234                                          null, Thread.currentThread())) 235             return; 236         try { 237             Callable<V> c = callable; 238             if (c != null && state == NEW) { 239  V result; 240                 boolean ran; 241                 try { 242                     result = c.call(); 243                     //任务正常完成 也就是没有出现异常,那么ran=true
244                     ran = true; 245                 } catch (Throwable ex) { 246                     //出现异常就调用setException方法,将状态state设置为EXCEPTIONAL
247                     result = null; 248                     ran = false; 249  setException(ex); 250  } 251                 //如果上面任务正常完成则调用set方法,那么将状态state设置为NORMAL
252                 if (ran) 253  set(result); 254  } 255         } finally { 256             runner = null; 257             int s = state; 258             //这里要判断一下,任务是否被取消过。
259             if (s >= INTERRUPTING) 260  handlePossibleCancellationInterrupt(s); 261  } 262  } 263 
264     /**
265  * Executes the computation without setting its result, and then 266  * resets this future to initial state, failing to do so if the 267  * computation encounters an exception or is cancelled. This is 268  * designed for use with tasks that intrinsically execute more 269  * than once. 270  * 271  * @return true if successfully run and reset 272      */
273     protected boolean runAndReset() { 274         if (state != NEW ||
275             !UNSAFE.compareAndSwapObject(this, runnerOffset, 276                                          null, Thread.currentThread())) 277             return false; 278         boolean ran = false; 279         int s = state; 280         try { 281             Callable<V> c = callable; 282             if (c != null && s == NEW) { 283                 try { 284                     c.call(); // don't set result
285                     ran = true; 286                 } catch (Throwable ex) { 287  setException(ex); 288  } 289  } 290         } finally { 291             // runner must be non-null until state is settled to 292             // prevent concurrent calls to run()
293             runner = null; 294             // state must be re-read after nulling runner to prevent 295             // leaked interrupts
296             s = state; 297             if (s >= INTERRUPTING) 298  handlePossibleCancellationInterrupt(s); 299  } 300         return ran && s == NEW; 301  } 302 
303     /**
304  * 如果其他线程正在中午该任务,那么运行该任务的线程就暂时让出CPU时间 305  * 一直到state=INTERRUPTED为止。 306      */
307     private void handlePossibleCancellationInterrupt(int s) { 308 
309         if (s == INTERRUPTING) 310             while (state == INTERRUPTING) 311                 Thread.yield(); // wait out pending interrupt
312  } 313 
314     /**
315  * Simple linked list nodes to record waiting threads in a Treiber 316  * stack. See other classes such as Phaser and SynchronousQueue 317  * for more detailed explanation. 318      */
319     static final class WaitNode { 320         volatile Thread thread; 321         volatile WaitNode next; 322         WaitNode() { thread = Thread.currentThread(); } 323  } 324 
325     /**
326  * 解锁所有等待队列当中的线程,让他们调用done方法。 327      */
328     private void finishCompletion() { 329         // assert state > COMPLETING;
330         for (WaitNode q; (q = waiters) != null;) { 331             if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) { 332                 for (;;) { 333                     Thread t = q.thread; 334                     if (t != null) { 335                         q.thread = null; 336  LockSupport.unpark(t); 337  } 338                     WaitNode next = q.next; 339                     if (next == null) 340                         break; 341                     q.next = null; // unlink to help gc
342                     q = next; 343  } 344                 break; 345  } 346  } 347 
348  done(); 349 
350         callable = null;        // to reduce footprint
351  } 352 
353     /**
354  * 等待任务完成,在get方法中最后会调用到这里 355      */
356     private int awaitDone(boolean timed, long nanos) 357         throws InterruptedException { 358         //计算一下需要等待的时间,有可能为0,为0的话就无限期等待。
359         final long deadline = timed ? System.nanoTime() + nanos : 0L; 360         //一个等待节点
361         WaitNode q = null; 362         //是否加入队列
363         boolean queued = false; 364         for (;;) { 365             //如果当且调用get方法的线程被interrupt 那么就把当前线程从等待队列remove 366             //然后抛出异常
367             if (Thread.interrupted()) { 368  removeWaiter(q); 369                 throw new InterruptedException(); 370  } 371             int s = state; 372             //如果任务已经完成 不管是被暂停了 还是出现异常了 只要状态大于COMPLETING就返回。
373             if (s > COMPLETING) { 374                 if (q != null) 375                     q.thread = null; 376                 return s; 377  } 378             else if (s == COMPLETING) // cannot time out yet 379             //如果任务正在完成中(注意是ING进行时) 那么让出一会CPU时间在继续执行
380  Thread.yield(); 381             else if (q == null) 382                 //创建等待节点
383                 q = new WaitNode(); 384             else if (!queued) 385                 //第一次创建的等待节点需要加入等待队列,这里加入队列等待。
386                 queued = UNSAFE.compareAndSwapObject(this, waitersOffset, 387                                                      q.next = waiters, q); 388             else if (timed) { 389                 //如果设置了等待 那么就得锁住线程等待,如果时间到了就返回状态。 390                 //方法 public V get(long timeout, TimeUnit unit) 这里会根据状态做后续处理。
391                 nanos = deadline - System.nanoTime(); 392                 if (nanos <= 0L) { 393  removeWaiter(q); 394                     return state; 395  } 396                 LockSupport.parkNanos(this, nanos); 397  } 398             else
399                 //否则锁住线程等待其他线程解锁。
400                 LockSupport.park(this); 401  } 402  } 403 
404     /**
405  * 这里就是循环线程队列,将当前等待节点remove掉 406      */
407     private void removeWaiter(WaitNode node) { 408         if (node != null) { 409             node.thread = null; 410  retry: 411             for (;;) {          // restart on removeWaiter race
412                 for (WaitNode pred = null, q = waiters, s; q != null; q = s) { 413                     s = q.next; 414                     if (q.thread != null) 415                         pred = q; 416                     else if (pred != null) { 417                         pred.next = s; 418                         if (pred.thread == null) // check for race
419                             continue retry; 420  } 421                     else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, 422  q, s)) 423                         continue retry; 424  } 425                 break; 426  } 427  } 428  } 429 
430     // Unsafe mechanics
431     private static final sun.misc.Unsafe UNSAFE; 432     private static final long stateOffset; 433     private static final long runnerOffset; 434     private static final long waitersOffset; 435     static { 436         try { 437             UNSAFE = sun.misc.Unsafe.getUnsafe(); 438             Class<?> k = FutureTask.class; 439             stateOffset = UNSAFE.objectFieldOffset 440                 (k.getDeclaredField("state")); 441             runnerOffset = UNSAFE.objectFieldOffset 442                 (k.getDeclaredField("runner")); 443             waitersOffset = UNSAFE.objectFieldOffset 444                 (k.getDeclaredField("waiters")); 445         } catch (Exception e) { 446             throw new Error(e); 447  } 448  } 449 
450 }

 

 

 

由此可见此类的设计还是非常复杂的,不得不佩服老外们的逻辑思维能力,相当的缜密,上面一些关于状态的转换过程还是比较复杂的,

因为要考虑到多线程的情况,还要考虑到状态转换时的线程安全问题等等。具体奥妙还的自己分析源码慢慢体会。

 

    原文作者:大新博客
    原文地址: http://www.cnblogs.com/daxin/p/3802392.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞