ExecutorServiceを介してCallableやRunnableタスクをスレッドプールに提出し、その結果を待機またはキャンセルするためのFutureオブジェクトを取得することができます。
public interface ExecutorService extends Executor {
...
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
...
}
Futureはインターフェースであり、その機能は実装クラスであるFutureTaskによって提供されます。FutureTaskはRunnableFutureインターフェースを実装しており、これはRunnableとFutureの両方を継承しています。
FutureTaskの生成過程
ExecutorServiceインターフェースのsubmitメソッドを通じて、CallableまたはRunnableをFutureTaskオブジェクトにラップします。
具体的には、ThreadPoolExecutorの抽象親クラスであるAbstractExecutorServiceのsubmitメソッドの実装を見ることができます。
public abstract class AbstractExecutorService implements ExecutorService {
...
// RunnableをFutureTaskにラップしてexecuteメソッドを呼び出す
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
// Runnableと戻り値resultをFutureTaskにラップしてexecuteメソッドを呼び出す
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;
}
// newTaskForメソッドを呼び出してRunnableとvalueからFutureTaskを作成
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value); // RunnableからFutureTaskを作成
}
// CallableオブジェクトをFutureTaskにラップする
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
// newTaskForメソッドを呼び出してCallableからFutureTaskを作成
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
...
}
ここで重要なのは、submitメソッドが受け取る引数がRunnableでもCallableでも、最初にnewTaskForメソッドを呼び出し、それらをFutureTaskオブジェクトに変換してからexecuteメソッドを呼び出していることです。
AbstractExecutorServiceではsubmitメソッドだけが実装されており、実際にタスクを実行するexecuteメソッドは子クラスのThreadPoolExecutorで実装されています。
次に、FutureTaskのコンストラクタを見てみましょう。
// Callableオブジェクトを受け取るコンストラクタ
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
// Runnableオブジェクトを受け取るコンストラクタ
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result); // RunnableをCallableに変換
this.state = NEW;
}
内部的には、RunnableもCallableも最終的にCallableに変換され、FutureTaskインスタンスが作成されます。
RunnableをCallableに変換する処理は、Executorsクラスのcallableメソッドで行われます。
// RunnableをCallableに変換するメソッド
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
// RunnableAdapterはExecutorsの内部クラスで、Callableインターフェースを実装
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
RunnableAdapterは典型的なアダプターパターンで、callメソッドが呼ばれたときには内部でRunnableのrunメソッドが実行されます。
FutureTaskの実行過程
submitメソッドの役割は、CallableまたはRunnableをFutureTaskオブジェクトに変換し、execute(Runnable task)メソッドを呼び出してタスクを実行することです。
executeメソッドは子クラスであるThreadPoolExecutorで実装されていますが、ここではFutureTaskのrunメソッドの実装に注目します。
public class FutureTask<V> implements RunnableFuture<V> {
/*
* FutureTaskの状態を表すstate変数
* 状態の遷移: NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
private volatile int state;
// 主な状態定数
private static final int NEW = 0; // FutureTaskが新しく作成された状態
private static final int COMPLETING = 1; // callメソッドの実行完了直後
private static final int NORMAL = 2; // callメソッドが正常終了
private static final int EXCEPTIONAL = 3; // callメソッドが例外で終了
// タスクのキャンセルに関する状態
private static final int CANCELLED= 4; // タスクがまだ実行されていないときにキャンセル
private static final int INTERRUPTING = 5; // runメソッドの実行中に中断信号を受け取った状態
private static final int INTERRUPTED = 6; // 中断が成功した状態
private Callable<V> callable; // 実行するタスク
private Object outcome; // タスクの結果
private volatile Thread runner; // タスクを実行するスレッド
...
// タスクのキャンセル
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try {
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally {
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
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 = null;
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
finishCompletion();
}
}
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
finishCompletion();
}
}
private void finishCompletion() {
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;
q = next;
}
break;
}
}
done();
}
protected void done() { }
private void handlePossibleCancellationInterrupt(int s) {
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield();
}
// Unsafe関連
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
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);
}
}
...
}
FutureTaskの結果取得過程
getメソッドを使用してタスクの実行結果を取得します。
/**
* タスクの結果を無期限に待つ
* タスクがキャンセルされた場合はInterruptedException、異常終了した場合はExecutionExceptionが発生
*/
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
/**
* 指定時間内にタスクの結果を待つ
* 指定時間が経過しても結果が得られなかった場合TimeoutException、キャンセルされた場合はInterruptedException、異常終了した場合はExecutionExceptionが発生
*/
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 int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING)
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
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);
}