import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class AddTask implements Callable<Integer> {

 private int a,b;
 
 public AddTask(int a, int b) {
 this.a = a;
 this.b = b;
 }
 
 @Override
 public Integer call throws Exception {
 Integer result = a + b;
 return result;
 }
 
 public static void main(String[] args) throws InterruptedException, ExecutionException {
 ExecutorService executor = Executors.newSingleThreadExecutor;
 //JDK目前為止返回的都是FutureTask的實(shí)例
 Future<Integer> future = executor.submit(new AddTask(1, 2));
 Integer result = future.get;// 只有當(dāng)future的狀態(tài)是已完成時(shí)(future.isDone = true),get方法才會(huì)返回
 }
} 

 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;
} 

 package future;

import java.util.concurrent.CancellationException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * The result of an asynchronous operation.
 * 
 * @author lixiaohui
 * @param <V> 執(zhí)行結(jié)果的類型參數(shù)
 */
public interface IFuture<V> extends Future<V> { 
 boolean isSuccess; // 是否成功 
 V getNow; //立即返回結(jié)果(不管Future是否處于完成狀態(tài))
 Throwable cause; //若執(zhí)行失敗時(shí)的原因
    boolean isCancellable; //是否可以取消
 IFuture<V> await throws InterruptedException; //等待future的完成
 boolean await(long timeoutMillis) throws InterruptedException; // 超時(shí)等待future的完成
 boolean await(long timeout, TimeUnit timeunit) throws InterruptedException;
    IFuture<V> awaitUninterruptibly; //等待future的完成，不響應(yīng)中斷
    boolean awaitUninterruptibly(long timeoutMillis);//超時(shí)等待future的完成，不響應(yīng)中斷
 boolean awaitUninterruptibly(long timeout, TimeUnit timeunit);
 IFuture<V> addListener(IFutureListener<V> l); //當(dāng)future完成時(shí)，會(huì)通知這些加進(jìn)來的監(jiān)聽器
 IFuture<V> removeListener(IFutureListener<V> l);
 
} 

 public class Object {
  /**
   * Causes the current thread to wait until another thread invokes the
   * {@link java.lang.Object#notify} method or the
   * {@link java.lang.Object#notifyAll} method for this object.
   * In other words, this method behaves exactly as if it simply
   * performs the call {@code wait(0)}.
   * 調(diào)用該方法后，當(dāng)前線程會(huì)釋放對(duì)象監(jiān)視器鎖，并讓出CPU使用權(quán)。直到別的線程調(diào)用notify/notifyAll
   */
  public final void wait throws InterruptedException {
    wait(0);
  }

  /**
   * Wakes up all threads that are waiting on this object's monitor. A
   * thread waits on an object's monitor by calling one of the
   * {@code wait} methods.
   * <p>
   * The awakened threads will not be able to proceed until the current
   * thread relinquishes the lock on this object. The awakened threads
   * will compete in the usual manner with any other threads that might
   * be actively competing to synchronize on this object; for example,
   * the awakened threads enjoy no reliable privilege or disadvantage in
   * being the next thread to lock this object.
   */
  public final native void notifyAll;
} 

 package future;

import java.util.Collection;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * <pre>
 * 正常結(jié)束時(shí), 若執(zhí)行的結(jié)果不為null, 則result為執(zhí)行結(jié)果; 若執(zhí)行結(jié)果為null, 則result = {@link AbstractFuture#SUCCESS_SIGNAL}
 * 異常結(jié)束時(shí), result為 {@link CauseHolder} 的實(shí)例;若是被取消而導(dǎo)致的異常結(jié)束, 則result為 {@link CancellationException} 的實(shí)例, 否則為其它異常的實(shí)例
 * 以下情況會(huì)使異步操作由未完成狀態(tài)轉(zhuǎn)至已完成狀態(tài), 也就是在以下情況發(fā)生時(shí)調(diào)用notifyAll方法:
 * <ul>
 * <li>異步操作被取消時(shí)(cancel方法)</li>
 * <li>異步操作正常結(jié)束時(shí)(setSuccess方法)</li>
 * <li>異步操作異常結(jié)束時(shí)(setFailure方法)</li>
 * </ul>
 * </pre>
 * 
 * @author lixiaohui
 *
 * @param <V>
 * 異步執(zhí)行結(jié)果的類型
 */
public class AbstractFuture<V> implements IFuture<V> {

 protected volatile Object result; // 需要保證其可見性
    /**
     * 監(jiān)聽器集
     */
 protected Collection<IFutureListener<V>> listeners = new CopyOnWriteArrayList<IFutureListener<V>>;

 /**
 * 當(dāng)任務(wù)正常執(zhí)行結(jié)果為null時(shí), 即客戶端調(diào)用{@link AbstractFuture#setSuccess(null)}時(shí), 
 * result引用該對(duì)象
 */
 private static final SuccessSignal SUCCESS_SIGNAL = new SuccessSignal;

 @Override
 public boolean cancel(boolean mayInterruptIfRunning) {
 if (isDone) { // 已完成了不能取消
  return false;
 }

 synchronized (this) {
  if (isDone) { // double check
  return false;
  }
  result = new CauseHolder(new CancellationException);
  notifyAll; // isDone = true, 通知等待在該對(duì)象的wait的線程
 }
 notifyListeners; // 通知監(jiān)聽器該異步操作已完成
 return true;
 }
 
 @Override
 public boolean isCancellable {
 return result == null;
 }
 
 @Override
 public boolean isCancelled {
 return result != null && result instanceof CauseHolder && ((CauseHolder) result).cause instanceof CancellationException;
 }

 @Override
 public boolean isDone {
 return result != null;
 }

 @Override
 public V get throws InterruptedException, ExecutionException {
 await; // 等待執(zhí)行結(jié)果

 Throwable cause = cause;
 if (cause == null) { // 沒有發(fā)生異常，異步操作正常結(jié)束
  return getNow;
 }
 if (cause instanceof CancellationException) { // 異步操作被取消了
  throw (CancellationException) cause;
 }
 throw new ExecutionException(cause); // 其他異常
 }

 @Override
 public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
 if (await(timeout, unit)) {// 超時(shí)等待執(zhí)行結(jié)果
  Throwable cause = cause;
  if (cause == null) {// 沒有發(fā)生異常，異步操作正常結(jié)束
  return getNow;
  }
  if (cause instanceof CancellationException) {// 異步操作被取消了
  throw (CancellationException) cause;
  }
  throw new ExecutionException(cause);// 其他異常
 }
 // 時(shí)間到了異步操作還沒有結(jié)束, 拋出超時(shí)異常
 throw new TimeoutException;
 }

 @Override
 public boolean isSuccess {
 return result == null ? false : !(result instanceof CauseHolder);
 }

 @SuppressWarnings("unchecked")
 @Override
 public V getNow {
 return (V) (result == SUCCESS_SIGNAL ? null : result);
 }

 @Override
 public Throwable cause {
 if (result != null && result instanceof CauseHolder) {
  return ((CauseHolder) result).cause;
 }
 return null;
 }

 @Override
 public IFuture<V> addListener(IFutureListener<V> listener) {
 if (listener == null) {
  throw new NullPointerException("listener");
 }
 if (isDone) { // 若已完成直接通知該監(jiān)聽器
  notifyListener(listener);
  return this;
 }
 synchronized (this) {
  if (!isDone) {
  listeners.add(listener);
  return this;
  }
 }
 notifyListener(listener);
 return this;
 }

 @Override
 public IFuture<V> removeListener(IFutureListener<V> listener) {
 if (listener == null) {
  throw new NullPointerException("listener");
 }

 if (!isDone) {
  listeners.remove(listener);
 }

 return this;
 }

 @Override
 public IFuture<V> await throws InterruptedException {
 return await0(true);
 }

 
 private IFuture<V> await0(boolean interruptable) throws InterruptedException {
 if (!isDone) { // 若已完成就直接返回了
  // 若允許終端且被中斷了則拋出中斷異常
  if (interruptable && Thread.interrupted) {
  throw new InterruptedException("thread " + Thread.currentThread.getName + " has been interrupted.");
  }

  boolean interrupted = false;
  synchronized (this) {
  while (!isDone) {
   try {
   wait; // 釋放鎖進(jìn)入waiting狀態(tài)，等待其它線程調(diào)用本對(duì)象的notify/notifyAll方法
   } catch (InterruptedException e) {
   if (interruptable) {
    throw e;
   } else {
    interrupted = true;
   }
   }
  }
  }
  if (interrupted) {
  // 為什么這里要設(shè)中斷標(biāo)志位？因?yàn)閺膚ait方法返回后, 中斷標(biāo)志是被clear了的, 
  // 這里重新設(shè)置以便讓其它代碼知道這里被中斷了。
  Thread.currentThread.interrupt;
  }
 }
 return this;
 }
 
 @Override
 public boolean await(long timeoutMillis) throws InterruptedException {
 return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), true);
 }
 
 @Override
 public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
 return await0(unit.toNanos(timeout), true);
 }

 private boolean await0(long timeoutNanos, boolean interruptable) throws InterruptedException {
 if (isDone) {
  return true;
 }

 if (timeoutNanos <= 0) {
  return isDone;
 }

 if (interruptable && Thread.interrupted) {
  throw new InterruptedException(toString);
 }

 long startTime = timeoutNanos <= 0 ? 0 : System.nanoTime;
 long waitTime = timeoutNanos;
 boolean interrupted = false;

 try {
  synchronized (this) {
  if (isDone) {
   return true;
  }

  if (waitTime <= 0) {
   return isDone;
  }

  for (;;) {
   try {
   wait(waitTime / 1000000, (int) (waitTime % 1000000));
   } catch (InterruptedException e) {
   if (interruptable) {
    throw e;
   } else {
    interrupted = true;
   }
   }

   if (isDone) {
   return true;
   } else {
   waitTime = timeoutNanos - (System.nanoTime - startTime);
   if (waitTime <= 0) {
    return isDone;
   }
   }
  }
  }
 } finally {
  if (interrupted) {
  Thread.currentThread.interrupt;
  }
 }
 }

 @Override
 public IFuture<V> awaitUninterruptibly {
 try {
  return await0(false);
 } catch (InterruptedException e) { // 這里若拋異常了就無法處理了
  throw new java.lang.InternalError;
 }
 }
 
 @Override
 public boolean awaitUninterruptibly(long timeoutMillis) {
 try {
  return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), false);
 } catch (InterruptedException e) {
  throw new java.lang.InternalError;
 }
 }

 @Override
 public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {
 try {
  return await0(unit.toNanos(timeout), false);
 } catch (InterruptedException e) {
  throw new java.lang.InternalError;
 }
 }

 protected IFuture<V> setFailure(Throwable cause) {
 if (setFailure0(cause)) {
  notifyListeners;
  return this;
 }
 throw new IllegalStateException("complete already: " + this);
 }

 private boolean setFailure0(Throwable cause) {
 if (isDone) {
  return false;
 }

 synchronized (this) {
  if (isDone) {
  return false;
  }
  result = new CauseHolder(cause);
  notifyAll;
 }

 return true;
 }

 protected IFuture<V> setSuccess(Object result) {
 if (setSuccess0(result)) { // 設(shè)置成功后通知監(jiān)聽器
  notifyListeners;
  return this;
 }
 throw new IllegalStateException("complete already: " + this);
 }

 private boolean setSuccess0(Object result) {
 if (isDone) {
  return false;
 }

 synchronized (this) {
  if (isDone) {
  return false;
  }
  if (result == null) { // 異步操作正常執(zhí)行完畢的結(jié)果是null
  this.result = SUCCESS_SIGNAL;
  } else {
  this.result = result;
  }
  notifyAll;
 }
 return true;
 }

 private void notifyListeners {
 for (IFutureListener<V> l : listeners) {
  notifyListener(l);
 }
 }

 private void notifyListener(IFutureListener<V> l) {
 try {
  l.operationCompleted(this);
 } catch (Exception e) {
  e.printStackTrace;
 }
 }

 private static class SuccessSignal {

 }

 private static final class CauseHolder {
 final Throwable cause;

 CauseHolder(Throwable cause) {
  this.cause = cause;
 }
 }
} 

 package future.test;

import future.IFuture;
import future.IFutureListener;

/**
 * 延時(shí)加法
 * @author lixiaohui
 *
 */
public class DelayAdder {
 
 public static void main(String[] args) {
 new DelayAdder.add(3 * 1000, 1, 2).addListener(new IFutureListener<Integer> {
  
  @Override
  public void operationCompleted(IFuture<Integer> future) throws Exception {
  System.out.println(future.getNow);
  }
  
 });
 }
 /**
 * 延遲加
 * @param delay 延時(shí)時(shí)長(zhǎng) milliseconds
 * @param a 加數(shù)
 * @param b 加數(shù)
 * @return 異步結(jié)果
 */
 public DelayAdditionFuture add(long delay, int a, int b) {
 DelayAdditionFuture future = new DelayAdditionFuture; 
 new Thread(new DelayAdditionTask(delay, a, b, future)).start;
 return future;
 }
 
 private class DelayAdditionTask implements Runnable {

 private long delay;
 
 private int a, b;
 
 private DelayAdditionFuture future;
 
 public DelayAdditionTask(long delay, int a, int b, DelayAdditionFuture future) {
  super;
  this.delay = delay;
  this.a = a;
  this.b = b;
  this.future = future;
 }

 @Override
 public void run {
  try {
  Thread.sleep(delay);
  Integer i = a + b;
  // TODO 這里設(shè)置future為完成狀態(tài)(正常執(zhí)行完畢)
  future.setSuccess(i);
  } catch (InterruptedException e) {
  // TODO 這里設(shè)置future為完成狀態(tài)(異常執(zhí)行完畢)
  future.setFailure(e.getCause);
  }
 }
 
 }
} package future.test;

import future.AbstractFuture;
import future.IFuture;
//只是把兩個(gè)方法對(duì)外暴露
public class DelayAdditionFuture extends AbstractFuture<Integer> {
 
 @Override
 public IFuture<Integer> setSuccess(Object result) {
 return super.setSuccess(result);
 }
 
 @Override
 public IFuture<Integer> setFailure(Throwable cause) {
 return super.setFailure(cause);
 }
 
}