promise/future模式

Author: beston123

src/main/java/code/concurrency/promise/AbstractFuture.java

@@ -7,6 +7,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiConsumer;
 
 /**
  * 〈AbstractFuture〉<p>
@@ -31,36 +32,34 @@ public abstract class AbstractFuture<R> implements Future<R> {
     protected AtomicReference<Callback> callbackReference = new AtomicReference<>();
 
     @Override
-    public boolean isDone(){
-       return done;
+    public boolean isDone() {
+        return done;
     }
 
     @Override
-    public boolean isSuccess(){
+    public boolean isSuccess() {
         return isDone() && result != null;
     }
 
     @Override
-    public boolean isCancelled(){
+    public boolean isCancelled() {
         return isDone() && cancelled;
     }
 
     @Override
-    public boolean cancel(){
+    public boolean cancel() {
         throw new UnsupportedOperationException();
     }
 
     @Override
     public R get() throws InterruptedException, ExecutionException {
-        if(!isDone()){
+        if (!isDone()) {
             latch.await();
         }
 
-        if(isSuccess()){
+        if (isSuccess()) {
             return result;
-        }
-
-        if(isCancelled()){
+        } else if (isCancelled()) {
             throw new ExecutionException("Task has been cancelled", null);
         }
 
@@ -69,27 +68,45 @@ public R get() throws InterruptedException, ExecutionException {
 
     @Override
     public R get(long timeout, TimeUnit unit)
-            throws InterruptedException, ExecutionException, TimeoutException{
-        if(!isDone()){
-            if(latch.await(timeout, unit) && !isDone()){
+            throws InterruptedException, ExecutionException, TimeoutException {
+        if (!isDone()) {
+            if (latch.await(timeout, unit) && !isDone()) {
                 throw new TimeoutException("Get result timeout");
             }
         }
 
-        if(isSuccess()){
+        if (isSuccess()) {
             return result;
-        }
-
-        if(isCancelled()){
+        } else if (isCancelled()) {
             throw new ExecutionException("Task has been cancelled", null);
         }
 
         throw new ExecutionException(cause);
     }
 
-    protected void executeCallbackOnce(){
+    @Override
+    public Future<R> whenComplete(BiConsumer<? super R, ? super Throwable> callback) {
+        callbackReference.compareAndSet(null, () -> callback.accept(this.result, this.cause));
+
+        // 执行时间极端或线程池满的情况下，放入callback前已经完成，此处需要主动执行回调
+        if (isDone()) {
+            executeOnlyOnce();
+        }
+        return this;
+    }
+
+    @Override
+    public void setDone() {
+        this.done = true;
+        latch.countDown();
+
+        // 精确执行一次回调
+        executeOnlyOnce();
+    }
+
+    private void executeOnlyOnce() {
         Callback callback = callbackReference.getAndSet(null);
-        if(callback != null){
+        if (callback != null) {
             callback.call();
         }
     }

src/main/java/code/concurrency/promise/DefaultFuture.java

@@ -7,7 +7,7 @@
  * @author zixiao
  * @date 2019/6/11
  */
-public class DefaultPromise<R> extends DefaultFuture<R> implements Promise<R> {
+public class DefaultPromise<R> extends AbstractFuture<R> implements Promise<R> {
 
     @Override
     public Future<R> getFuture() {
@@ -16,69 +16,60 @@ public Future<R> getFuture() {
 
     @Override
     public Promise<R> setSuccess(R result) throws IllegalStateException {
-        if(trySuccess(result)){
+        if (trySuccess(result)) {
             return this;
         }
         throw new IllegalStateException("Set success exception.");
     }
 
     @Override
     public boolean trySuccess(R result) {
-        if(result == null){
+        if (result == null) {
             throw new NullPointerException("Result can not be null");
         }
-        if(isDone()){
+        if (isDone()) {
             return false;
         }
 
         doneLock.lock();
         try {
-            if(!isDone()){
+            if (!isDone()) {
                 this.result = result;
                 setDone();
                 return true;
             }
-        }finally {
+        } finally {
             doneLock.unlock();
         }
         return false;
     }
 
-    /**
-     * 设置为完成
-     */
-    private void setDone(){
-        this.done = true;
-        latch.countDown();
-
-        executeCallbackOnce();
-    }
-
     @Override
     public Promise<R> setFailure(Throwable cause) {
-        if(tryFailure(cause)){
+        if (tryFailure(cause)) {
             return this;
         }
         throw new IllegalStateException("Set failure exception.");
     }
 
     @Override
     public boolean tryFailure(Throwable cause) {
-        if(cause == null){
+        if (cause == null) {
             throw new NullPointerException("Cause can not be null");
         }
-        if(isDone()){
+
+        if (isDone()) {
             return false;
         }
 
         doneLock.lock();
         try {
-            if(!isDone()){
+            if (!isDone()) {
                 this.cause = cause;
                 setDone();
                 return true;
             }
-        }finally {
+        } finally {
             doneLock.unlock();
         }
         return false;

src/main/java/code/concurrency/promise/DefaultPromise.java

@@ -21,13 +21,20 @@ public DefaultPromisor(ExecutorService executorService) {
      * 启动异步任务的执行，并返回用于获取异步任务执行结果的凭据对象
      */
     @Override
-    public <R> Future<R> asyncExecute(Callable<R> callable){
-        if(callable == null){
+    public <R> Future<R> asyncExecute(Callable<R> callable) {
+        if (callable == null) {
             throw new NullPointerException("Callable can not be null");
         }
         Promise<R> promise = new DefaultPromise<>();
         try {
-            executorService.execute(new PromiseTask<>(promise, callable));
+            executorService.execute(() -> {
+                try {
+                    R r = callable.call();
+                    promise.trySuccess(r);
+                } catch (Exception e) {
+                    promise.tryFailure(e);
+                }
+            });
         } catch (RejectedExecutionException e) {
             promise.setFailure(e);
         }
@@ -40,35 +47,13 @@ public <R> R execute(Callable<R> callable) throws ExecutionException, Interrupte
         return future.get();
     }
 
-    private static class PromiseTask<R> implements Runnable{
-
-        private Promise<R> promise;
-
-        private Callable<R> callable;
-
-        public PromiseTask(Promise<R> promise, Callable<R> callable) {
-            this.promise = promise;
-            this.callable = callable;
-        }
-
-        @Override
-        public void run(){
-            try {
-                R r = callable.call();
-                promise.trySuccess(r);
-            } catch (Exception e) {
-                promise.tryFailure(e);
-            }
-        }
-    }
-
     @Override
-    public void shutdown(long time, TimeUnit unit){
+    public void shutdown(long time, TimeUnit unit) {
         executorService.shutdown();
         try {
             executorService.awaitTermination(10, TimeUnit.SECONDS);
         } catch (InterruptedException e) {
-            e.printStackTrace();
+            Thread.currentThread().interrupt();
         }
     }
 

src/main/java/code/concurrency/promise/DefaultPromisor.java

@@ -56,6 +56,16 @@ public interface Future <R> {
      */
     R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
 
+    /**
+     * 完成回调
+     * @param callback
+     * @return
+     */
     Future<R> whenComplete(BiConsumer<? super R, ? super Throwable> callback);
 
+    /**
+     * 设置完成
+     */
+    void setDone();
+
 }

src/main/java/code/concurrency/promise/Future.java

@@ -1,6 +1,5 @@
 package code.concurrency.promise;
 
-import code.util.ThreadPool;
 import org.junit.Test;
 
 import java.text.MessageFormat;
@@ -15,13 +14,13 @@
  */
 public class PromiseTest {
 
-    private ExecutorService executorService = ThreadPool.create(100, 100, 1024);
+    private ExecutorService executorService = new ThreadPoolExecutor(64, 64, 5, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(256));
 
     private Promisor promisor = new DefaultPromisor(executorService);
 
     @Test
-    public void testAsync(){
-        for (int i = 0; i < 100; i++) {
+    public void testAsync() {
+        for (int i = 0; i < 2048; i++) {
             promisor.asyncExecute(new Task(i)).whenComplete((s, t) -> {
                 if (t != null) {
                     System.out.println("Completed fail: " + t.getMessage());