@FunctionalInterface public interface Runnable { /** * When an object implementing interface <code>Runnable</code> is used * to create a thread, starting the thread causes the object's * <code>run</code> method to be called in that separately executing * thread. * <p> * The general contract of the method <code>run</code> is that it may * take any action whatsoever. * * @see java.lang.Thread#run() */ public abstract void run(); }
@FunctionalInterface public 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; }
RunnableFuture作为 Runnable 的 Future。成功执行 run 方法可以完成 Future 并允许访问其结果
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
/** * 作为 Runnable 的 Future。成功执行 run 方法可以完成 Future 并允许访问其结果。 * A {@link Future} that is {@link Runnable}. Successful execution of * the {@code run} method causes completion of the {@code Future} * and allows access to its results. * @see FutureTask * @see Executor * @since 1.6 * @author Doug Lea * @param <V> The result type returned by this Future's {@code get} method */ public interface RunnableFuture<V> extends Runnable, Future<V> { /** * Sets this Future to the result of its computation * unless it has been cancelled. */ void run(); }
FutureTask
我们先来看一下FutureTask的实现：
1
public class FutureTask<V> implements RunnableFuture<V>
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; } public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result); execute(ftask); return ftask; } public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }
/** * Returns a {@code RunnableFuture} for the given callable task. * * @param callable the callable task being wrapped * @param <T> the type of the callable's result * @return a {@code RunnableFuture} which, when run, will call the * underlying callable and which, as a {@code Future}, will yield * the callable's result as its result and provide for * cancellation of the underlying task * @since 1.6 */ protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return new FutureTask<T>(callable); }
/** * Returns a {@code RunnableFuture} for the given runnable and default * value. * * @param runnable the runnable task being wrapped * @param value the default value for the returned future * @param <T> the type of the given value * @return a {@code RunnableFuture} which, when run, will run the * underlying runnable and which, as a {@code Future}, will yield * the given value as its result and provide for cancellation of * the underlying task * @since 1.6 */ protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { return new FutureTask<T>(runnable, value); }
/** * 任务状态 * The run state of this task, initially NEW. The run state * transitions to a terminal state only in methods set, * setException, and cancel. During completion, state may take on * transient values of COMPLETING (while outcome is being set) or * INTERRUPTING (only while interrupting the runner to satisfy a * cancel(true)). Transitions from these intermediate to final * states use cheaper ordered/lazy writes because values are unique * and cannot be further modified. * * Possible state transitions: * NEW -> COMPLETING -> NORMAL * NEW -> COMPLETING -> EXCEPTIONAL * NEW -> CANCELLED * NEW -> INTERRUPTING -> INTERRUPTED */ 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;//任务中断
/** The underlying callable; nulled out after running */ private Callable<V> callable; //任务返回值，正常返回时是泛型指定对象，任务异常时是Throwable对象，它在state=COMPLETING阶段完成赋值操作 /** The result to return or exception to throw from get() */ private Object outcome; // non-volatile, protected by state reads/writes //当前执行任务线程，run()方法开始时进行判断和赋值，保证同一时刻只有一个线程执行FutureTask，并且FutureTask.run()只能执行一次。 /** The thread running the callable; CASed during run() */ private volatile Thread runner; //阻塞队列头节点，每个节点存储调用FutureTask.get()方法，且采用LockSupport.park()阻塞的线程。在任务对outcome完成赋值后，调用finishCompletion()唤醒所有阻塞线程。 /** Treiber stack of waiting threads */ private volatile WaitNode waiters;
/** * 将等待线程封装成节点，形成等待队列 * Simple linked list nodes to record waiting threads in a Treiber * stack. See other classes such as Phaser and SynchronousQueue * for more detailed explanation. */ static final class WaitNode { volatile Thread thread; volatile WaitNode next; WaitNode() { thread = Thread.currentThread(); } }
//UNSAFE类提供高效并且线程安全方式操作变量，直接和内存数据打交道， // 但是分配的内存需要手动释放。由于UNSAFE类只提供给JVM信任的启动类加载器所使用，这里采用反射获取UNSAFE对象。 // Unsafe mechanics private static final sun.misc.Unsafe UNSAFE; //state参数在内存中对象的偏移量 private static final long stateOffset; //runner参数在内存中对象的偏移量 private static final long runnerOffset; //waiter参数在内存中对象的偏移量 private static final long waitersOffset; static { try { //通过反射的方式来访问Unsafe类中的theUnsafe静态成员变量，该theUnsafe静态成员变量在Unsafe第一次使用时就已经初始化。 UNSAFE = sun.misc.Unsafe.getUnsafe(); Class<?> k = FutureTask.class; stateOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("state")); runnerOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("runner")); waitersOffset = UNSAFE.objectFieldOffset (k.getDeclaredField("waiters")); } catch (Exception e) { throw new Error(e); } }
isDown方法
1 2 3 4 5 6 7
/** * 因为NEW表示任务执行阶段，因此判断任务是否完成 * @return */ public boolean isDone() { return state != NEW; }
/** * 如果任务正在执行或者正在赋值，调用awaitDone()方法，阻塞当前线程，且封装成WaitNode，放入等待队列中 * 否则，调用report()，将outcome包装成返回参数类型。 * @throws CancellationException {@inheritDoc} */ public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L); return report(s); }
int s = state; if (s > COMPLETING) { if (q != null) q.thread = null; return s; } else if (s == COMPLETING) // cannot time out yet Thread.yield(); else if (q == null) q = new WaitNode(); else if (!queued) //在队列头部添加q，并将q赋值给waiters queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q); else if (timed) { nanos = deadline - System.nanoTime(); if (nanos <= 0L) { removeWaiter(q); return state; } //将线程阻塞，设置阻塞时间为nanos LockSupport.parkNanos(this, nanos); } else LockSupport.park(this); } }
/** * 功能：返回任务结果 * 描述： 由于在set(),setException() 方法中设置了otcome的值， * 可能是throwable对象，也可能是正常返回结果，所以需要 * 对outcome进行一次处理 */ @SuppressWarnings("unchecked") 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); }
/** * 定时获取结果 */ 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); }
/** * 判断：状态是否为NEW，并CAS成功将runner线程赋值为当前线程 * 线程调用run(),进行判断，运行callable.call()，再调用set()将结果设置到outcome。 * 失败调用setException()，将Throwable对象设置到outcome。 */ public void run() { 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(); ran = true; } catch (Throwable ex) { result = null; ran = false; setException(ex); } if (ran) 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); } }
/** * 设置返回结果 * Sets the result of this future to the given value unless * this future has already been set or has been cancelled. * * <p>This method is invoked internally by the {@link #run} method * upon successful completion of the computation. * * @param v the value */ protected void set(V v) { if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion(); } }
/** * 通过waiters参数，唤醒等待队列，所有因为调用get()方法而阻塞的线程 LockSupport.unpark():线程取消阻塞 * Removes and signals all waiting threads, invokes done(), and * nulls out callable. */ 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; 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 }
/** * Protected method invoked when this task transitions to state * {@code isDone} (whether normally or via cancellation). The * default implementation does nothing. Subclasses may override * this method to invoke completion callbacks or perform * bookkeeping. Note that you can query status inside the * implementation of this method to determine whether this task * has been cancelled. */ protected void done() { }