同時17:Futureの詳細
9353 ワード
Future
FutureはJ.U.Cのインタフェースで、非同期実行結果を表しています.
Futureは、スレッドが実行時に呼び出し元に残された1つのルートと見なすことができ、これによりスレッド実行状態の表示(isDone)、実行の取り消し(cancel)、実行の完了待ちをブロックして結果を返す(get)、非同期実行コールバック関数(callback)などの操作を行うことができる.
public interface Future {
/** ，mayInterruptIfRunning-false: **/
boolean cancel(boolean mayInterruptIfRunning);
/** **/
boolean isCancelled();
/** **/
boolean isDone();
/** **/
V get() throws InterruptedException, ExecutionException;
/** ， **/
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
ReentrantLockでFutureを実現
小さな栗:
public class ResponseFuture implements Future {
private final ResponseCallback callback;
private String responsed;
private final ReentrantLock lock = new ReentrantLock();
private final Condition condition = lock.newCondition();
public ResponseFuture(ResponseCallback callback) {
this.callback = callback;
}
public boolean isDone() {
return null != this.responsed;
}
public String get() throws InterruptedException, ExecutionException {
if (!isDone()) {
try {
this.lock.lock();
while (!this.isDone()) {
condition.await();
if (this.isDone()) break;
}
} finally {
this.lock.unlock();
}
}
return this.responsed;
}
//
public void done(String responsed) throws Exception{
this.responsed = responsed;
try {
this.lock.lock();
this.condition.signal();
if(null != this.callback) this.callback.call(this.responsed);
} finally {
this.lock.unlock();
}
}
public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
throw new UnsupportedOperationException();
}
public boolean cancel(boolean mayInterruptIfRunning) {
throw new UnsupportedOperationException();
}
public boolean isCancelled() {
throw new UnsupportedOperationException();
}
}
このFutureを使用:
public class ResponseCallback {
public String call(String o) {
System.out.println("ResponseCallback： ， ："+o);
return o;
}
}
public class FutureTest {
public static void main(String[] args) {
final ResponseFuture responseFuture = new ResponseFuture(new ResponseCallback());
new Thread(new Runnable() {//
public void run() {
System.out.println(" ");
// System.out.println(responseFuture.get()); ，
System.out.println(" , ResponseCallback ");
}
}).start();
new Thread(new Runnable() {//
public void run() {
try {
Thread.sleep(10000);//
responseFuture.done("ok");//
}catch (Exception e) {
e.printStackTrace();
}
}).start();
}
}
このFutureはConditionのawaitとsignalメソッドを使用しています
同期アクション:getが呼び出されたときに結果が返されていないことが判明した場合、スレッドは結果が返されるまでFutureに待機中のスレッドを起動するように通知するまで待機をブロックします.
非同期操作:呼び出し線形は待つ必要はなく、結果が戻ったときにsignal()を呼び出して待機スレッドがないことを発見し、Callbackを直接呼び出します.
FutureTask
FutureTaskはJ.U.CのFutureの実装であり、Runnableインタフェースも実装されており、ExecutorServiceでこのFutureの実行を直接コミットすることができます.
小さな栗:
public class FutureTaskTest {
public static class CallableImpl implements Callable{
public String call() throws Exception {
String tName = Thread.currentThread().getName();
System.out.println(tName+" : ");
Thread.sleep(6000);
return tName+" ok";
}
}
public static void main(String[] args) throws Exception {
ExecutorService es = Executors.newFixedThreadPool(1);
// // ， 6000
// FutureTask future = new FutureTask(new CallableImpl());
// es.submit(future);
// System.out.println(future.get());
//
FutureTask future = new FutureTask(new CallableImpl()) {
protected void done() {
try {
System.out.println(" ："+this.get());
} catch (Exception e) {
e.printStackTrace();
}
}
};
es.submit(future);
System.out.println(" ");
es.shutdown();
}
}
FutureTask実装
一方向チェーンテーブル
WaitNodeを使用して待ち行列スレッドを表し、volatile stateを現在のスレッドの実行状態を表し、状態変換は以下の通りである.NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED
awaitDone
ブロック待ち
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();// CPU
else if (q == null)// ，
q = new WaitNode();
else if (!queued)// waiters q
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);//
}
}
run
スレッド実行
// s1
public void run() {
// state!=NEW, ， 。
// CAS ， runner ， ， ， 。
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable 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);// s2
}
if (ran)
set(result);// s3
}
} finally {
runner = null;// runner
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
s 1:stateがNEWでないか、他のスレッドが実行されている場合runメソッドは実行されません.実行中異常転入s 2実行終了転入s 3
// s2
protected void setException(Throwable t) {
// NEW-COMPLETING-EXCEPTIONAL
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
s 2:このステップ完了状態:NEW-COMPLETING-EXCEPTIONAL置換異常付与結果outcome転入s 4
// s3
protected void set(V v) {
// NEW-COMPLETING-NORMAL
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
s 3:このステップ完了状態:NEW-COMPLETING-NORMAL置換実行結果をoutcomeに割り当ててs 4に移行する
// s4
private void finishCompletion() {
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {//waiters
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
}
s4:for(;;)の異常起動待ちスレッド呼び出しコールバックフックdone()が起動したスレッドはawaitDoneでCPUを譲り、ループを終了します.
小結
1、Futureは1つの非同期実行結果2、getは同期操作で、現在のスレッドは実行完了を待つことをブロックして、結果を返す3、非同期操作はフックを使ってコールバックして、現在のスレッドをブロックしない