Java并发编程之异步Future机制的原理和实现

项目中经常有些任务需要异步(提交到线程池中)去执行，而主线程往往需要知道异步执行产生的结果，这时我们要怎么做呢？用runnable是无法实现的，我们需要用callable看下面的代码：

Java代码  收藏代码
  1. import java.util.concurrent.Callable;  
  2. import java.util.concurrent.ExecutionException;  
  3. import java.util.concurrent.ExecutorService;  
  4. import java.util.concurrent.Executors;  
  5. import java.util.concurrent.Future;  
  6.   
  7. public class AddTask implements Callable<Integer> {  
  8.   
  9.     private int a,b;  
  10.       
  11.     public AddTask(int a, int b) {  
  12.         this.a = a;  
  13.         this.b = b;  
  14.     }  
  15.       
  16.     @Override  
  17.     public Integer call() throws Exception {  
  18.         Integer result = a + b;  
  19.         return result;  
  20.     }  
  21.       
  22.     public static void main(String[] args) throws InterruptedException, ExecutionException {  
  23.         ExecutorService executor = Executors.newSingleThreadExecutor();  
  24.         //JDK目前为止返回的都是FutureTask的实例  
  25.         Future<Integer> future = executor.submit(new AddTask(1, 2));  
  26.         Integer result = future.get();// 只有当future的状态是已完成时(future.isDone() = true),get()方法才会返回  
  27.     }  
  28. }  

  虽然可以实现获取异步执行结果的需求，但是我们发现这个Future其实很不好用，因为它没有提供通知的机制，也就是说我们不知道future什么时候完成（如果我们需要轮询isDone()来判断的话感觉就没有用这个的必要了）。看下java.util.concurrent.future.Future 的接口方法：

Java代码  收藏代码
  1. public interface Future<V> {  
  2.     boolean cancel(boolean mayInterruptIfRunning);  
  3.     boolean isCancelled();  
  4.     boolean isDone();  
  5.     V get() throws InterruptedException, ExecutionException;  
  6.     V get(long timeout, TimeUnit unit)  
  7.         throws InterruptedException, ExecutionException, TimeoutException;  
  8. }  

 由此可见JDK的Future机制其实并不好用，如果能给这个future加个监听器，让它在完成时通知监听器的话就比较好用了，就像下面这个IFuture：

 

Java代码  收藏代码
  1. package future;  
  2.   
  3. import java.util.concurrent.CancellationException;  
  4. import java.util.concurrent.Future;  
  5. import java.util.concurrent.TimeUnit;  
  6.   
  7. /** 
  8.  * The result of an asynchronous operation. 
  9.  *  
  10.  * @author lixiaohui 
  11.  * @param <V> 执行结果的类型参数 
  12.  */  
  13. public interface IFuture<V> extends Future<V> {   
  14.     boolean isSuccess(); // 是否成功      
  15.     V getNow(); //立即返回结果(不管Future是否处于完成状态)  
  16.     Throwable cause();  //若执行失败时的原因  
  17.         boolean isCancellable(); //是否可以取消  
  18.     IFuture<V> await() throws InterruptedException; //等待future的完成  
  19.     boolean await(long timeoutMillis) throws InterruptedException; // 超时等待future的完成  
  20.     boolean await(long timeout, TimeUnit timeunit) throws InterruptedException;  
  21.         IFuture<V> awaitUninterruptibly(); //<span style="line-height: 1.5;">等待future的完成，不响应中断</span>  
  22.         boolean awaitUninterruptibly(long timeoutMillis);<span style="line-height: 1.5;">//超时</span><span style="line-height: 1.5;">等待future的完成，不响应中断</span>  
  23.     boolean awaitUninterruptibly(long timeout, TimeUnit timeunit);  
  24.     IFuture<V> addListener(IFutureListener<V> l); //当future完成时，会通知这些加进来的监听器  
  25.     IFuture<V> removeListener(IFutureListener<V> l);  
  26.       
  27. }  

 

 

接下来就一起来实现这个IFuture，在这之前要说明下Object.wait(),Object.notifyAll()方法，因为整个Future实现的原理的核心就是这两个方法.看看JDK里面的解释:

Java代码  收藏代码
  1. public class Object {  
  2.     /** 
  3.      * Causes the current thread to wait until another thread invokes the 
  4.      * {@link java.lang.Object#notify()} method or the 
  5.      * {@link java.lang.Object#notifyAll()} method for this object. 
  6.      * In other words, this method behaves exactly as if it simply 
  7.      * performs the call {@code wait(0)}. 
  8.      * 调用该方法后，当前线程会释放对象监视器锁，并让出CPU使用权。直到别的线程调用notify()/notifyAll() 
  9.      */  
  10.     public final void wait() throws InterruptedException {  
  11.         wait(0);  
  12.     }  
  13.   
  14.     /** 
  15.      * Wakes up all threads that are waiting on this object's monitor. A 
  16.      * thread waits on an object's monitor by calling one of the 
  17.      * {@code wait} methods. 
  18.      * <p> 
  19.      * The awakened threads will not be able to proceed until the current 
  20.      * thread relinquishes the lock on this object. The awakened threads 
  21.      * will compete in the usual manner with any other threads that might 
  22.      * be actively competing to synchronize on this object; for example, 
  23.      * the awakened threads enjoy no reliable privilege or disadvantage in 
  24.      * being the next thread to lock this object. 
  25.      */  
  26.     public final native void notifyAll();  
  27. }  

 知道这个后，我们要自己实现Future也就有了思路，当线程调用了IFuture.await()等一系列的方法时，如果Future还未完成，那么就调用future.wait() 方法使线程进入WAITING状态。而当别的线程设置Future为完成状态（注意这里的完成状态包括正常结束和异常结束）时，就需要调用future.notifyAll()方法来唤醒之前因为调用过wait()方法而处于WAITING状态的那些线程。完整的实现如下(代码应该没有很难理解的地方，我是参考netty的Future机制的。有兴趣的可以去看看netty的源码)：

Java代码  收藏代码
  1. package future;  
  2.   
  3. import java.util.Collection;  
  4. import java.util.concurrent.CancellationException;  
  5. import java.util.concurrent.CopyOnWriteArrayList;  
  6. import java.util.concurrent.ExecutionException;  
  7. import java.util.concurrent.TimeUnit;  
  8. import java.util.concurrent.TimeoutException;  
  9.   
  10. /** 
  11.  * <pre> 
  12.  * 正常结束时, 若执行的结果不为null, 则result为执行结果; 若执行结果为null, 则result = {@link AbstractFuture#SUCCESS_SIGNAL} 
  13.  * 异常结束时, result为 {@link CauseHolder} 的实例;若是被取消而导致的异常结束, 则result为 {@link CancellationException} 的实例, 否则为其它异常的实例 
  14.  * 以下情况会使异步操作由未完成状态转至已完成状态, 也就是在以下情况发生时调用notifyAll()方法: 
  15.  * <ul> 
  16.  * <li>异步操作被取消时(cancel方法)</li> 
  17.  * <li>异步操作正常结束时(setSuccess方法)</li> 
  18.  * <li>异步操作异常结束时(setFailure方法)</li> 
  19.  * </ul> 
  20.  * </pre> 
  21.  *  
  22.  * @author lixiaohui 
  23.  * 
  24.  * @param <V> 
  25.  *            异步执行结果的类型 
  26.  */  
  27. public class AbstractFuture<V> implements IFuture<V> {  
  28.   
  29.     protected volatile Object result; // 需要保证其可见性  
  30.         /** 
  31.          * 监听器集 
  32.          */  
  33.     protected Collection<IFutureListener<V>> listeners = new CopyOnWriteArrayList<IFutureListener<V>>();  
  34.   
  35.     /** 
  36.      * 当任务正常执行结果为null时, 即客户端调用{@link AbstractFuture#setSuccess(null)}时,  
  37.      * result引用该对象 
  38.      */  
  39.     private static final SuccessSignal SUCCESS_SIGNAL = new SuccessSignal();  
  40.   
  41.     @Override  
  42.     public boolean cancel(boolean mayInterruptIfRunning) {  
  43.         if (isDone()) { // 已完成了不能取消  
  44.             return false;  
  45.         }  
  46.   
  47.         synchronized (this) {  
  48.             if (isDone()) { // double check  
  49.                 return false;  
  50.             }  
  51.             result = new CauseHolder(new CancellationException());  
  52.             notifyAll(); // isDone = true, 通知等待在该对象的wait()的线程  
  53.         }  
  54.         notifyListeners(); // 通知监听器该异步操作已完成  
  55.         return true;  
  56.     }  
  57.       
  58.     @Override  
  59.     public boolean isCancellable() {  
  60.         return result == null;  
  61.     }  
  62.       
  63.     @Override  
  64.     public boolean isCancelled() {  
  65.         return result != null && result instanceof CauseHolder && ((CauseHolder) result).cause instanceof CancellationException;  
  66.     }  
  67.   
  68.     @Override  
  69.     public boolean isDone() {  
  70.         return result != null;  
  71.     }  
  72.   
  73.     @Override  
  74.     public V get() throws InterruptedException, ExecutionException {  
  75.         await(); // 等待执行结果  
  76.   
  77.         Throwable cause = cause();  
  78.         if (cause == null) { // 没有发生异常，异步操作正常结束  
  79.             return getNow();  
  80.         }  
  81.         if (cause instanceof CancellationException) { // 异步操作被取消了  
  82.             throw (CancellationException) cause;  
  83.         }  
  84.         throw new ExecutionException(cause); // 其他异常  
  85.     }  
  86.   
  87.     @Override  
  88.     public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {  
  89.         if (await(timeout, unit)) {// 超时等待执行结果  
  90.             Throwable cause = cause();  
  91.             if (cause == null) {// 没有发生异常，异步操作正常结束  
  92.                 return getNow();  
  93.             }  
  94.             if (cause instanceof CancellationException) {// 异步操作被取消了  
  95.                 throw (CancellationException) cause;  
  96.             }  
  97.             throw new ExecutionException(cause);// 其他异常  
  98.         }  
  99.         // 时间到了异步操作还没有结束, 抛出超时异常  
  100.         throw new TimeoutException();  
  101.     }  
  102.   
  103.     @Override  
  104.     public boolean isSuccess() {  
  105.         return result == null ? false : !(result instanceof CauseHolder);  
  106.     }  
  107.   
  108.     @SuppressWarnings("unchecked")  
  109.     @Override  
  110.     public V getNow() {  
  111.         return (V) (result == SUCCESS_SIGNAL ? null : result);  
  112.     }  
  113.   
  114.     @Override  
  115.     public Throwable cause() {  
  116.         if (result != null && result instanceof CauseHolder) {  
  117.             return ((CauseHolder) result).cause;  
  118.         }  
  119.         return null;  
  120.     }  
  121.   
  122.     @Override  
  123.     public IFuture<V> addListener(IFutureListener<V> listener) {  
  124.         if (listener == null) {  
  125.             throw new NullPointerException("listener");  
  126.         }  
  127.         if (isDone()) { // 若已完成直接通知该监听器  
  128.             notifyListener(listener);  
  129.             return this;  
  130.         }  
  131.         synchronized (this) {  
  132.             if (!isDone()) {  
  133.                 listeners.add(listener);  
  134.                 return this;  
  135.             }  
  136.         }  
  137.         notifyListener(listener);  
  138.         return this;  
  139.     }  
  140.   
  141.     @Override  
  142.     public IFuture<V> removeListener(IFutureListener<V> listener) {  
  143.         if (listener == null) {  
  144.             throw new NullPointerException("listener");  
  145.         }  
  146.   
  147.         if (!isDone()) {  
  148.             listeners.remove(listener);  
  149.         }  
  150.   
  151.         return this;  
  152.     }  
  153.   
  154.     @Override  
  155.     public IFuture<V> await() throws InterruptedException {  
  156.         return await0(true);  
  157.     }  
  158.   
  159.       
  160.     private IFuture<V> await0(boolean interruptable) throws InterruptedException {  
  161.         if (!isDone()) { // 若已完成就直接返回了  
  162.             // 若允许终端且被中断了则抛出中断异常  
  163.             if (interruptable && Thread.interrupted()) {  
  164.                 throw new InterruptedException("thread " + Thread.currentThread().getName() + " has been interrupted.");  
  165.             }  
  166.   
  167.             boolean interrupted = false;  
  168.             synchronized (this) {  
  169.                 while (!isDone()) {  
  170.                     try {  
  171.                         wait(); // 释放锁进入waiting状态，等待其它线程调用本对象的notify()/notifyAll()方法  
  172.                     } catch (InterruptedException e) {  
  173.                         if (interruptable) {  
  174.                             throw e;  
  175.                         } else {  
  176.                             interrupted = true;  
  177.                         }  
  178.                     }  
  179.                 }  
  180.             }  
  181.             if (interrupted) {  
  182.                 // 为什么这里要设中断标志位？因为从wait方法返回后, 中断标志是被clear了的,   
  183.                 // 这里重新设置以便让其它代码知道这里被中断了。  
  184.                 Thread.currentThread().interrupt();  
  185.             }  
  186.         }  
  187.         return this;  
  188.     }  
  189.       
  190.     @Override  
  191.     public boolean await(long timeoutMillis) throws InterruptedException {  
  192.         return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), true);  
  193.     }  
  194.       
  195.     @Override  
  196.     public boolean await(long timeout, TimeUnit unit) throws InterruptedException {  
  197.         return await0(unit.toNanos(timeout), true);  
  198.     }  
  199.   
  200.     private boolean await0(long timeoutNanos, boolean interruptable) throws InterruptedException {  
  201.         if (isDone()) {  
  202.             return true;  
  203.         }  
  204.   
  205.         if (timeoutNanos <= 0) {  
  206.             return isDone();  
  207.         }  
  208.   
  209.         if (interruptable && Thread.interrupted()) {  
  210.             throw new InterruptedException(toString());  
  211.         }  
  212.   
  213.         long startTime = timeoutNanos <= 0 ? 0 : System.nanoTime();  
  214.         long waitTime = timeoutNanos;  
  215.         boolean interrupted = false;  
  216.   
  217.         try {  
  218.             synchronized (this) {  
  219.                 if (isDone()) {  
  220.                     return true;  
  221.                 }  
  222.   
  223.                 if (waitTime <= 0) {  
  224.                     return isDone();  
  225.                 }  
  226.   
  227.                 for (;;) {  
  228.                     try {  
  229.                         wait(waitTime / 1000000, (int) (waitTime % 1000000));  
  230.                     } catch (InterruptedException e) {  
  231.                         if (interruptable) {  
  232.                             throw e;  
  233.                         } else {  
  234.                             interrupted = true;  
  235.                         }  
  236.                     }  
  237.   
  238.                     if (isDone()) {  
  239.                         return true;  
  240.                     } else {  
  241.                         waitTime = timeoutNanos - (System.nanoTime() - startTime);  
  242.                         if (waitTime <= 0) {  
  243.                             return isDone();  
  244.                         }  
  245.                     }  
  246.                 }  
  247.             }  
  248.         } finally {  
  249.             if (interrupted) {  
  250.                 Thread.currentThread().interrupt();  
  251.             }  
  252.         }  
  253.     }  
  254.   
  255.     @Override  
  256.     public IFuture<V> awaitUninterruptibly() {  
  257.         try {  
  258.             return await0(false);  
  259.         } catch (InterruptedException e) { // 这里若抛异常了就无法处理了  
  260.             throw new java.lang.InternalError();  
  261.         }  
  262.     }  
  263.       
  264.     @Override  
  265.     public boolean awaitUninterruptibly(long timeoutMillis) {  
  266.         try {  
  267.             return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), false);  
  268.         } catch (InterruptedException e) {  
  269.             throw new java.lang.InternalError();  
  270.         }  
  271.     }  
  272.   
  273.     @Override  
  274.     public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {  
  275.         try {  
  276.             return await0(unit.toNanos(timeout), false);  
  277.         } catch (InterruptedException e) {  
  278.             throw new java.lang.InternalError();  
  279.         }  
  280.     }  
  281.   
  282.     protected IFuture<V> setFailure(Throwable cause) {  
  283.         if (setFailure0(cause)) {  
  284.             notifyListeners();  
  285.             return this;  
  286.         }  
  287.         throw new IllegalStateException("complete already: " + this);  
  288.     }  
  289.   
  290.     private boolean setFailure0(Throwable cause) {  
  291.         if (isDone()) {  
  292.             return false;  
  293.         }  
  294.   
  295.         synchronized (this) {  
  296.             if (isDone()) {  
  297.                 return false;  
  298.             }  
  299.             result = new CauseHolder(cause);  
  300.             notifyAll();  
  301.         }  
  302.   
  303.         return true;  
  304.     }  
  305.   
  306.     protected IFuture<V> setSuccess(Object result) {  
  307.         if (setSuccess0(result)) { // 设置成功后通知监听器  
  308.             notifyListeners();  
  309.             return this;  
  310.         }  
  311.         throw new IllegalStateException("complete already: " + this);  
  312.     }  
  313.   
  314.     private boolean setSuccess0(Object result) {  
  315.         if (isDone()) {  
  316.             return false;  
  317.         }  
  318.   
  319.         synchronized (this) {  
  320.             if (isDone()) {  
  321.                 return false;  
  322.             }  
  323.             if (result == null) { // 异步操作正常执行完毕的结果是null  
  324.                 this.result = SUCCESS_SIGNAL;  
  325.             } else {  
  326.                 this.result = result;  
  327.             }  
  328.             notifyAll();  
  329.         }  
  330.         return true;  
  331.     }  
  332.   
  333.     private void notifyListeners() {  
  334.         for (IFutureListener<V> l : listeners) {  
  335.             notifyListener(l);  
  336.         }  
  337.     }  
  338.   
  339.     private void notifyListener(IFutureListener<V> l) {  
  340.         try {  
  341.             l.operationCompleted(this);  
  342.         } catch (Exception e) {  
  343.             e.printStackTrace();  
  344.         }  
  345.     }  
  346.   
  347.     private static class SuccessSignal {  
  348.   
  349.     }  
  350.   
  351.     private static final class CauseHolder {  
  352.         final Throwable cause;  
  353.   
  354.         CauseHolder(Throwable cause) {  
  355.             this.cause = cause;  
  356.         }  
  357.     }  
  358. }  

 监听接口

public interface IFutureListener<V> {
    void operationCompleted(IFuture<V> future) throws Exception;
}

那么要怎么使用这个呢，有了上面的骨架实现，我们就可以定制各种各样的异步结果了。下面模拟一下一个延时的任务：

Java代码  收藏代码
  1. package future.test;  
  2.   
  3. import future.IFuture;  
  4. import future.IFutureListener;  
  5.   
  6. /** 
  7.  * 延时加法 
  8.  * @author lixiaohui 
  9.  * 
  10.  */  
  11. public class DelayAdder {  
  12.       
  13.     public static void main(String[] args) {  
  14.         new DelayAdder().add(3 * 1000, 1, 2).addListener(new IFutureListener<Integer>() {  
  15.               
  16.             @Override  
  17.             public void operationCompleted(IFuture<Integer> future) throws Exception {  
  18.                 System.out.println(future.getNow());  
  19.             }  
  20.               
  21.         });  
  22.     }  
  23.     /** 
  24.      * 延迟加 
  25.      * @param delay 延时时长 milliseconds 
  26.      * @param a 加数 
  27.      * @param b 加数 
  28.      * @return 异步结果 
  29.      */  
  30.     public DelayAdditionFuture add(long delay, int a, int b) {  
  31.         DelayAdditionFuture future = new DelayAdditionFuture();   
  32.         new Thread(new DelayAdditionTask(delay, a, b, future)).start();  
  33.         return future;  
  34.     }  
  35.       
  36.     private class DelayAdditionTask implements Runnable {  
  37.   
  38.         private long delay;  
  39.           
  40.         private int a, b;  
  41.           
  42.         private DelayAdditionFuture future;  
  43.           
  44.         public DelayAdditionTask(long delay, int a, int b, DelayAdditionFuture future) {  
  45.             super();  
  46.             this.delay = delay;  
  47.             this.a = a;  
  48.             this.b = b;  
  49.             this.future = future;  
  50.         }  
  51.   
  52.         @Override  
  53.         public void run() {  
  54.             try {  
  55.                 Thread.sleep(delay);  
  56.                 Integer i = a + b;  
  57.                 // TODO 这里设置future为完成状态(正常执行完毕)  
  58.                 future.setSuccess(i);  
  59.             } catch (InterruptedException e) {  
  60.                 // TODO 这里设置future为完成状态(异常执行完毕)  
  61.                 future.setFailure(e.getCause());  
  62.             }  
  63.         }  
  64.           
  65.     }  
  66. }  

 

Java代码  收藏代码
  1. package future.test;  
  2.   
  3. import future.AbstractFuture;  
  4. import future.IFuture;  
  5. //只是把两个方法对外暴露  
  6. public class DelayAdditionFuture extends AbstractFuture<Integer> {  
  7.       
  8.     @Override  
  9.     public IFuture<Integer> setSuccess(Object result) {  
  10.         return super.setSuccess(result);  
  11.     }  
  12.       
  13.     @Override  
  14.     public IFuture<Integer> setFailure(Throwable cause) {  
  15.         return super.setFailure(cause);  
  16.     }  
  17.       
  18. }  

  可以看到客户端不用主动去询问future是否完成，而是future完成时自动回调operationcompleted方法，客户端只需在回调里实现逻辑即可。

        项目中经常有些任务需要异步(提交到线程池中)去执行，而主线程往往需要知道异步执行产生的结果，这时我们要怎么做呢？用runnable是无法实现的，我们需要用callable看下面的代码：

Java代码  收藏代码
  1. import java.util.concurrent.Callable;  
  2. import java.util.concurrent.ExecutionException;  
  3. import java.util.concurrent.ExecutorService;  
  4. import java.util.concurrent.Executors;  
  5. import java.util.concurrent.Future;  
  6.   
  7. public class AddTask implements Callable<Integer> {  
  8.   
  9.     private int a,b;  
  10.       
  11.     public AddTask(int a, int b) {  
  12.         this.a = a;  
  13.         this.b = b;  
  14.     }  
  15.       
  16.     @Override  
  17.     public Integer call() throws Exception {  
  18.         Integer result = a + b;  
  19.         return result;  
  20.     }  
  21.       
  22.     public static void main(String[] args) throws InterruptedException, ExecutionException {  
  23.         ExecutorService executor = Executors.newSingleThreadExecutor();  
  24.         //JDK目前为止返回的都是FutureTask的实例  
  25.         Future<Integer> future = executor.submit(new AddTask(1, 2));  
  26.         Integer result = future.get();// 只有当future的状态是已完成时(future.isDone() = true),get()方法才会返回  
  27.     }  
  28. }  

  虽然可以实现获取异步执行结果的需求，但是我们发现这个Future其实很不好用，因为它没有提供通知的机制，也就是说我们不知道future什么时候完成（如果我们需要轮询isDone()来判断的话感觉就没有用这个的必要了）。看下java.util.concurrent.future.Future 的接口方法：

Java代码  收藏代码
  1. public interface Future<V> {  
  2.     boolean cancel(boolean mayInterruptIfRunning);  
  3.     boolean isCancelled();  
  4.     boolean isDone();  
  5.     V get() throws InterruptedException, ExecutionException;  
  6.     V get(long timeout, TimeUnit unit)  
  7.         throws InterruptedException, ExecutionException, TimeoutException;  
  8. }  

 由此可见JDK的Future机制其实并不好用，如果能给这个future加个监听器，让它在完成时通知监听器的话就比较好用了，就像下面这个IFuture：

 

Java代码  收藏代码
  1. package future;  
  2.   
  3. import java.util.concurrent.CancellationException;  
  4. import java.util.concurrent.Future;  
  5. import java.util.concurrent.TimeUnit;  
  6.   
  7. /** 
  8.  * The result of an asynchronous operation. 
  9.  *  
  10.  * @author lixiaohui 
  11.  * @param <V> 执行结果的类型参数 
  12.  */  
  13. public interface IFuture<V> extends Future<V> {   
  14.     boolean isSuccess(); // 是否成功      
  15.     V getNow(); //立即返回结果(不管Future是否处于完成状态)  
  16.     Throwable cause();  //若执行失败时的原因  
  17.         boolean isCancellable(); //是否可以取消  
  18.     IFuture<V> await() throws InterruptedException; //等待future的完成  
  19.     boolean await(long timeoutMillis) throws InterruptedException; // 超时等待future的完成  
  20.     boolean await(long timeout, TimeUnit timeunit) throws InterruptedException;  
  21.         IFuture<V> awaitUninterruptibly(); //<span style="line-height: 1.5;">等待future的完成，不响应中断</span>  
  22.         boolean awaitUninterruptibly(long timeoutMillis);<span style="line-height: 1.5;">//超时</span><span style="line-height: 1.5;">等待future的完成，不响应中断</span>  
  23.     boolean awaitUninterruptibly(long timeout, TimeUnit timeunit);  
  24.     IFuture<V> addListener(IFutureListener<V> l); //当future完成时，会通知这些加进来的监听器  
  25.     IFuture<V> removeListener(IFutureListener<V> l);  
  26.       
  27. }  

 

 

接下来就一起来实现这个IFuture，在这之前要说明下Object.wait(),Object.notifyAll()方法，因为整个Future实现的原理的核心就是这两个方法.看看JDK里面的解释:

Java代码  收藏代码
  1. public class Object {  
  2.     /** 
  3.      * Causes the current thread to wait until another thread invokes the 
  4.      * {@link java.lang.Object#notify()} method or the 
  5.      * {@link java.lang.Object#notifyAll()} method for this object. 
  6.      * In other words, this method behaves exactly as if it simply 
  7.      * performs the call {@code wait(0)}. 
  8.      * 调用该方法后，当前线程会释放对象监视器锁，并让出CPU使用权。直到别的线程调用notify()/notifyAll() 
  9.      */  
  10.     public final void wait() throws InterruptedException {  
  11.         wait(0);  
  12.     }  
  13.   
  14.     /** 
  15.      * Wakes up all threads that are waiting on this object's monitor. A 
  16.      * thread waits on an object's monitor by calling one of the 
  17.      * {@code wait} methods. 
  18.      * <p> 
  19.      * The awakened threads will not be able to proceed until the current 
  20.      * thread relinquishes the lock on this object. The awakened threads 
  21.      * will compete in the usual manner with any other threads that might 
  22.      * be actively competing to synchronize on this object; for example, 
  23.      * the awakened threads enjoy no reliable privilege or disadvantage in 
  24.      * being the next thread to lock this object. 
  25.      */  
  26.     public final native void notifyAll();  
  27. }  

 知道这个后，我们要自己实现Future也就有了思路，当线程调用了IFuture.await()等一系列的方法时，如果Future还未完成，那么就调用future.wait() 方法使线程进入WAITING状态。而当别的线程设置Future为完成状态（注意这里的完成状态包括正常结束和异常结束）时，就需要调用future.notifyAll()方法来唤醒之前因为调用过wait()方法而处于WAITING状态的那些线程。完整的实现如下(代码应该没有很难理解的地方，我是参考netty的Future机制的。有兴趣的可以去看看netty的源码)：

Java代码  收藏代码
  1. package future;  
  2.   
  3. import java.util.Collection;  
  4. import java.util.concurrent.CancellationException;  
  5. import java.util.concurrent.CopyOnWriteArrayList;  
  6. import java.util.concurrent.ExecutionException;  
  7. import java.util.concurrent.TimeUnit;  
  8. import java.util.concurrent.TimeoutException;  
  9.   
  10. /** 
  11.  * <pre> 
  12.  * 正常结束时, 若执行的结果不为null, 则result为执行结果; 若执行结果为null, 则result = {@link AbstractFuture#SUCCESS_SIGNAL} 
  13.  * 异常结束时, result为 {@link CauseHolder} 的实例;若是被取消而导致的异常结束, 则result为 {@link CancellationException} 的实例, 否则为其它异常的实例 
  14.  * 以下情况会使异步操作由未完成状态转至已完成状态, 也就是在以下情况发生时调用notifyAll()方法: 
  15.  * <ul> 
  16.  * <li>异步操作被取消时(cancel方法)</li> 
  17.  * <li>异步操作正常结束时(setSuccess方法)</li> 
  18.  * <li>异步操作异常结束时(setFailure方法)</li> 
  19.  * </ul> 
  20.  * </pre> 
  21.  *  
  22.  * @author lixiaohui 
  23.  * 
  24.  * @param <V> 
  25.  *            异步执行结果的类型 
  26.  */  
  27. public class AbstractFuture<V> implements IFuture<V> {  
  28.   
  29.     protected volatile Object result; // 需要保证其可见性  
  30.         /** 
  31.          * 监听器集 
  32.          */  
  33.     protected Collection<IFutureListener<V>> listeners = new CopyOnWriteArrayList<IFutureListener<V>>();  
  34.   
  35.     /** 
  36.      * 当任务正常执行结果为null时, 即客户端调用{@link AbstractFuture#setSuccess(null)}时,  
  37.      * result引用该对象 
  38.      */  
  39.     private static final SuccessSignal SUCCESS_SIGNAL = new SuccessSignal();  
  40.   
  41.     @Override  
  42.     public boolean cancel(boolean mayInterruptIfRunning) {  
  43.         if (isDone()) { // 已完成了不能取消  
  44.             return false;  
  45.         }  
  46.   
  47.         synchronized (this) {  
  48.             if (isDone()) { // double check  
  49.                 return false;  
  50.             }  
  51.             result = new CauseHolder(new CancellationException());  
  52.             notifyAll(); // isDone = true, 通知等待在该对象的wait()的线程  
  53.         }  
  54.         notifyListeners(); // 通知监听器该异步操作已完成  
  55.         return true;  
  56.     }  
  57.       
  58.     @Override  
  59.     public boolean isCancellable() {  
  60.         return result == null;  
  61.     }  
  62.       
  63.     @Override  
  64.     public boolean isCancelled() {  
  65.         return result != null && result instanceof CauseHolder && ((CauseHolder) result).cause instanceof CancellationException;  
  66.     }  
  67.   
  68.     @Override  
  69.     public boolean isDone() {  
  70.         return result != null;  
  71.     }  
  72.   
  73.     @Override  
  74.     public V get() throws InterruptedException, ExecutionException {  
  75.         await(); // 等待执行结果  
  76.   
  77.         Throwable cause = cause();  
  78.         if (cause == null) { // 没有发生异常，异步操作正常结束  
  79.             return getNow();  
  80.         }  
  81.         if (cause instanceof CancellationException) { // 异步操作被取消了  
  82.             throw (CancellationException) cause;  
  83.         }  
  84.         throw new ExecutionException(cause); // 其他异常  
  85.     }  
  86.   
  87.     @Override  
  88.     public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {  
  89.         if (await(timeout, unit)) {// 超时等待执行结果  
  90.             Throwable cause = cause();  
  91.             if (cause == null) {// 没有发生异常，异步操作正常结束  
  92.                 return getNow();  
  93.             }  
  94.             if (cause instanceof CancellationException) {// 异步操作被取消了  
  95.                 throw (CancellationException) cause;  
  96.             }  
  97.             throw new ExecutionException(cause);// 其他异常  
  98.         }  
  99.         // 时间到了异步操作还没有结束, 抛出超时异常  
  100.         throw new TimeoutException();  
  101.     }  
  102.   
  103.     @Override  
  104.     public boolean isSuccess() {  
  105.         return result == null ? false : !(result instanceof CauseHolder);  
  106.     }  
  107.   
  108.     @SuppressWarnings("unchecked")  
  109.     @Override  
  110.     public V getNow() {  
  111.         return (V) (result == SUCCESS_SIGNAL ? null : result);  
  112.     }  
  113.   
  114.     @Override  
  115.     public Throwable cause() {  
  116.         if (result != null && result instanceof CauseHolder) {  
  117.             return ((CauseHolder) result).cause;  
  118.         }  
  119.         return null;  
  120.     }  
  121.   
  122.     @Override  
  123.     public IFuture<V> addListener(IFutureListener<V> listener) {  
  124.         if (listener == null) {  
  125.             throw new NullPointerException("listener");  
  126.         }  
  127.         if (isDone()) { // 若已完成直接通知该监听器  
  128.             notifyListener(listener);  
  129.             return this;  
  130.         }  
  131.         synchronized (this) {  
  132.             if (!isDone()) {  
  133.                 listeners.add(listener);  
  134.                 return this;  
  135.             }  
  136.         }  
  137.         notifyListener(listener);  
  138.         return this;  
  139.     }  
  140.   
  141.     @Override  
  142.     public IFuture<V> removeListener(IFutureListener<V> listener) {  
  143.         if (listener == null) {  
  144.             throw new NullPointerException("listener");  
  145.         }  
  146.   
  147.         if (!isDone()) {  
  148.             listeners.remove(listener);  
  149.         }  
  150.   
  151.         return this;  
  152.     }  
  153.   
  154.     @Override  
  155.     public IFuture<V> await() throws InterruptedException {  
  156.         return await0(true);  
  157.     }  
  158.   
  159.       
  160.     private IFuture<V> await0(boolean interruptable) throws InterruptedException {  
  161.         if (!isDone()) { // 若已完成就直接返回了  
  162.             // 若允许终端且被中断了则抛出中断异常  
  163.             if (interruptable && Thread.interrupted()) {  
  164.                 throw new InterruptedException("thread " + Thread.currentThread().getName() + " has been interrupted.");  
  165.             }  
  166.   
  167.             boolean interrupted = false;  
  168.             synchronized (this) {  
  169.                 while (!isDone()) {  
  170.                     try {  
  171.                         wait(); // 释放锁进入waiting状态，等待其它线程调用本对象的notify()/notifyAll()方法  
  172.                     } catch (InterruptedException e) {  
  173.                         if (interruptable) {  
  174.                             throw e;  
  175.                         } else {  
  176.                             interrupted = true;  
  177.                         }  
  178.                     }  
  179.                 }  
  180.             }  
  181.             if (interrupted) {  
  182.                 // 为什么这里要设中断标志位？因为从wait方法返回后, 中断标志是被clear了的,   
  183.                 // 这里重新设置以便让其它代码知道这里被中断了。  
  184.                 Thread.currentThread().interrupt();  
  185.             }  
  186.         }  
  187.         return this;  
  188.     }  
  189.       
  190.     @Override  
  191.     public boolean await(long timeoutMillis) throws InterruptedException {  
  192.         return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), true);  
  193.     }  
  194.       
  195.     @Override  
  196.     public boolean await(long timeout, TimeUnit unit) throws InterruptedException {  
  197.         return await0(unit.toNanos(timeout), true);  
  198.     }  
  199.   
  200.     private boolean await0(long timeoutNanos, boolean interruptable) throws InterruptedException {  
  201.         if (isDone()) {  
  202.             return true;  
  203.         }  
  204.   
  205.         if (timeoutNanos <= 0) {  
  206.             return isDone();  
  207.         }  
  208.   
  209.         if (interruptable && Thread.interrupted()) {  
  210.             throw new InterruptedException(toString());  
  211.         }  
  212.   
  213.         long startTime = timeoutNanos <= 0 ? 0 : System.nanoTime();  
  214.         long waitTime = timeoutNanos;  
  215.         boolean interrupted = false;  
  216.   
  217.         try {  
  218.             synchronized (this) {  
  219.                 if (isDone()) {  
  220.                     return true;  
  221.                 }  
  222.   
  223.                 if (waitTime <= 0) {  
  224.                     return isDone();  
  225.                 }  
  226.   
  227.                 for (;;) {  
  228.                     try {  
  229.                         wait(waitTime / 1000000, (int) (waitTime % 1000000));  
  230.                     } catch (InterruptedException e) {  
  231.                         if (interruptable) {  
  232.                             throw e;  
  233.                         } else {  
  234.                             interrupted = true;  
  235.                         }  
  236.                     }  
  237.   
  238.                     if (isDone()) {  
  239.                         return true;  
  240.                     } else {  
  241.                         waitTime = timeoutNanos - (System.nanoTime() - startTime);  
  242.                         if (waitTime <= 0) {  
  243.                             return isDone();  
  244.                         }  
  245.                     }  
  246.                 }  
  247.             }  
  248.         } finally {  
  249.             if (interrupted) {  
  250.                 Thread.currentThread().interrupt();  
  251.             }  
  252.         }  
  253.     }  
  254.   
  255.     @Override  
  256.     public IFuture<V> awaitUninterruptibly() {  
  257.         try {  
  258.             return await0(false);  
  259.         } catch (InterruptedException e) { // 这里若抛异常了就无法处理了  
  260.             throw new java.lang.InternalError();  
  261.         }  
  262.     }  
  263.       
  264.     @Override  
  265.     public boolean awaitUninterruptibly(long timeoutMillis) {  
  266.         try {  
  267.             return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), false);  
  268.         } catch (InterruptedException e) {  
  269.             throw new java.lang.InternalError();  
  270.         }  
  271.     }  
  272.   
  273.     @Override  
  274.     public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {  
  275.         try {  
  276.             return await0(unit.toNanos(timeout), false);  
  277.         } catch (InterruptedException e) {  
  278.             throw new java.lang.InternalError();  
  279.         }  
  280.     }  
  281.   
  282.     protected IFuture<V> setFailure(Throwable cause) {  
  283.         if (setFailure0(cause)) {  
  284.             notifyListeners();  
  285.             return this;  
  286.         }  
  287.         throw new IllegalStateException("complete already: " + this);  
  288.     }  
  289.   
  290.     private boolean setFailure0(Throwable cause) {  
  291.         if (isDone()) {  
  292.             return false;  
  293.         }  
  294.   
  295.         synchronized (this) {  
  296.             if (isDone()) {  
  297.                 return false;  
  298.             }  
  299.             result = new CauseHolder(cause);  
  300.             notifyAll();  
  301.         }  
  302.   
  303.         return true;  
  304.     }  
  305.   
  306.     protected IFuture<V> setSuccess(Object result) {  
  307.         if (setSuccess0(result)) { // 设置成功后通知监听器  
  308.             notifyListeners();  
  309.             return this;  
  310.         }  
  311.         throw new IllegalStateException("complete already: " + this);  
  312.     }  
  313.   
  314.     private boolean setSuccess0(Object result) {  
  315.         if (isDone()) {  
  316.             return false;  
  317.         }  
  318.   
  319.         synchronized (this) {  
  320.             if (isDone()) {  
  321.                 return false;  
  322.             }  
  323.             if (result == null) { // 异步操作正常执行完毕的结果是null  
  324.                 this.result = SUCCESS_SIGNAL;  
  325.             } else {  
  326.                 this.result = result;  
  327.             }  
  328.             notifyAll();  
  329.         }  
  330.         return true;  
  331.     }  
  332.   
  333.     private void notifyListeners() {  
  334.         for (IFutureListener<V> l : listeners) {  
  335.             notifyListener(l);  
  336.         }  
  337.     }  
  338.   
  339.     private void notifyListener(IFutureListener<V> l) {  
  340.         try {  
  341.             l.operationCompleted(this);  
  342.         } catch (Exception e) {  
  343.             e.printStackTrace();  
  344.         }  
  345.     }  
  346.   
  347.     private static class SuccessSignal {  
  348.   
  349.     }  
  350.   
  351.     private static final class CauseHolder {  
  352.         final Throwable cause;  
  353.   
  354.         CauseHolder(Throwable cause) {  
  355.             this.cause = cause;  
  356.         }  
  357.     }  
  358. }  

 

那么要怎么使用这个呢，有了上面的骨架实现，我们就可以定制各种各样的异步结果了。下面模拟一下一个延时的任务：

Java代码  收藏代码
  1. package future.test;  
  2.   
  3. import future.IFuture;  
  4. import future.IFutureListener;  
  5.   
  6. /** 
  7.  * 延时加法 
  8.  * @author lixiaohui 
  9.  * 
  10.  */  
  11. public class DelayAdder {  
  12.       
  13.     public static void main(String[] args) {  
  14.         new DelayAdder().add(3 * 1000, 1, 2).addListener(new IFutureListener<Integer>() {  
  15.               
  16.             @Override  
  17.             public void operationCompleted(IFuture<Integer> future) throws Exception {  
  18.                 System.out.println(future.getNow());  
  19.             }  
  20.               
  21.         });  
  22.     }  
  23.     /** 
  24.      * 延迟加 
  25.      * @param delay 延时时长 milliseconds 
  26.      * @param a 加数 
  27.      * @param b 加数 
  28.      * @return 异步结果 
  29.      */  
  30.     public DelayAdditionFuture add(long delay, int a, int b) {  
  31.         DelayAdditionFuture future = new DelayAdditionFuture();   
  32.         new Thread(new DelayAdditionTask(delay, a, b, future)).start();  
  33.         return future;  
  34.     }  
  35.       
  36.     private class DelayAdditionTask implements Runnable {  
  37.   
  38.         private long delay;  
  39.           
  40.         private int a, b;  
  41.           
  42.         private DelayAdditionFuture future;  
  43.           
  44.         public DelayAdditionTask(long delay, int a, int b, DelayAdditionFuture future) {  
  45.             super();  
  46.             this.delay = delay;  
  47.             this.a = a;  
  48.             this.b = b;  
  49.             this.future = future;  
  50.         }  
  51.   
  52.         @Override  
  53.         public void run() {  
  54.             try {  
  55.                 Thread.sleep(delay);  
  56.                 Integer i = a + b;  
  57.                 // TODO 这里设置future为完成状态(正常执行完毕)  
  58.                 future.setSuccess(i);  
  59.             } catch (InterruptedException e) {  
  60.                 // TODO 这里设置future为完成状态(异常执行完毕)  
  61.                 future.setFailure(e.getCause());  
  62.             }  
  63.         }  
  64.           
  65.     }  
  66. }  

 

Java代码  收藏代码
  1. package future.test;  
  2.   
  3. import future.AbstractFuture;  
  4. import future.IFuture;  
  5. //只是把两个方法对外暴露  
  6. public class DelayAdditionFuture extends AbstractFuture<Integer> {  
  7.       
  8.     @Override  
  9.     public IFuture<Integer> setSuccess(Object result) {  
  10.         return super.setSuccess(result);  
  11.     }  
  12.       
  13.     @Override  
  14.     public IFuture<Integer> setFailure(Throwable cause) {  
  15.         return super.setFailure(cause);  
  16.     }  
  17.       
  18. }  

  可以看到客户端不用主动去询问future是否完成，而是future完成时自动回调operationcompleted方法，客户端只需在回调里实现逻辑即可。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/cxhfuujust/p/12622039.html