promise/future

Author: beston123

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

@@ -0,0 +1,97 @@
+package code.concurrency.promise;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * 〈AbstractFuture〉<p>
+ *
+ * @author zixiao
+ * @date 2019/6/11
+ */
+public abstract class AbstractFuture<R> implements Future<R> {
+
+    protected volatile R result;
+
+    protected Throwable cause;
+
+    protected boolean done = false;
+
+    protected boolean cancelled = false;
+
+    protected CountDownLatch latch = new CountDownLatch(1);
+
+    protected Lock doneLock = new ReentrantLock();
+
+    protected AtomicReference<Callback> callbackReference = new AtomicReference<>();
+
+    @Override
+    public boolean isDone(){
+       return done;
+    }
+
+    @Override
+    public boolean isSuccess(){
+        return isDone() && result != null;
+    }
+
+    @Override
+    public boolean isCancelled(){
+        return isDone() && cancelled;
+    }
+
+    @Override
+    public boolean cancel(){
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public R get() throws InterruptedException, ExecutionException {
+        if(!isDone()){
+            latch.await();
+        }
+
+        if(isSuccess()){
+            return result;
+        }
+
+        if(isCancelled()){
+            throw new ExecutionException("Task has been cancelled", null);
+        }
+
+        throw new ExecutionException(cause);
+    }
+
+    @Override
+    public R get(long timeout, TimeUnit unit)
+            throws InterruptedException, ExecutionException, TimeoutException{
+        if(!isDone()){
+            if(latch.await(timeout, unit) && !isDone()){
+                throw new TimeoutException("Get result timeout");
+            }
+        }
+
+        if(isSuccess()){
+            return result;
+        }
+
+        if(isCancelled()){
+            throw new ExecutionException("Task has been cancelled", null);
+        }
+
+        throw new ExecutionException(cause);
+    }
+
+    protected void executeCallbackOnce(){
+        Callback callback = callbackReference.getAndSet(null);
+        if(callback != null){
+            callback.call();
+        }
+    }
+
+}

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

@@ -0,0 +1,13 @@
+package code.concurrency.promise;
+
+/**
+ * 〈Callback〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/6/12
+ */
+public interface Callback {
+
+    void call();
+}

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

@@ -0,0 +1,39 @@
+package code.concurrency.promise;
+
+import java.util.function.BiConsumer;
+
+/**
+ * 〈DefaultFuture〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/6/12
+ */
+public class DefaultFuture<R> extends AbstractFuture<R>{
+
+    @Override
+    public Future<R> whenComplete(BiConsumer<? super R, ? super Throwable> callback){
+        callbackReference.compareAndSet(null, new DefaultCallback(callback, this));
+        if(isDone()){
+            executeCallbackOnce();
+        }
+        return this;
+    }
+
+    private class DefaultCallback implements Callback {
+
+        private BiConsumer<? super R, ? super Throwable> callback;
+
+        private AbstractFuture<R> future;
+
+        public DefaultCallback(BiConsumer<? super R, ? super Throwable> callback, AbstractFuture<R> future) {
+            this.callback = callback;
+            this.future = future;
+        }
+
+        public void call(){
+            callback.accept(future.result, future.cause);
+        }
+    }
+
+}

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

@@ -0,0 +1,87 @@
+package code.concurrency.promise;
+
+/**
+ * 〈DefaultPromise〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/6/11
+ */
+public class DefaultPromise<R> extends DefaultFuture<R> implements Promise<R> {
+
+    @Override
+    public Future<R> getFuture() {
+        return this;
+    }
+
+    @Override
+    public Promise<R> setSuccess(R result) throws IllegalStateException {
+        if(trySuccess(result)){
+            return this;
+        }
+        throw new IllegalStateException("Set success exception.");
+    }
+
+    @Override
+    public boolean trySuccess(R result) {
+        if(result == null){
+            throw new NullPointerException("Result can not be null");
+        }
+        if(isDone()){
+            return false;
+        }
+
+        doneLock.lock();
+        try {
+            if(!isDone()){
+                this.result = result;
+                setDone();
+                return true;
+            }
+        }finally {
+            doneLock.unlock();
+        }
+        return false;
+    }
+
+    /**
+     * 设置为完成
+     */
+    private void setDone(){
+        this.done = true;
+        latch.countDown();
+
+        executeCallbackOnce();
+    }
+
+    @Override
+    public Promise<R> setFailure(Throwable cause) {
+        if(tryFailure(cause)){
+            return this;
+        }
+        throw new IllegalStateException("Set failure exception.");
+    }
+
+    @Override
+    public boolean tryFailure(Throwable cause) {
+        if(cause == null){
+            throw new NullPointerException("Cause can not be null");
+        }
+        if(isDone()){
+            return false;
+        }
+
+        doneLock.lock();
+        try {
+            if(!isDone()){
+                this.cause = cause;
+                setDone();
+                return true;
+            }
+        }finally {
+            doneLock.unlock();
+        }
+        return false;
+    }
+
+}

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

@@ -0,0 +1,75 @@
+package code.concurrency.promise;
+
+import java.util.concurrent.*;
+
+/**
+ * 〈DefaultPromisor〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/6/3
+ */
+public class DefaultPromisor implements Promisor {
+
+    private ExecutorService executorService;
+
+    public DefaultPromisor(ExecutorService executorService) {
+        this.executorService = executorService;
+    }
+
+    /**
+     * 启动异步任务的执行，并返回用于获取异步任务执行结果的凭据对象
+     */
+    @Override
+    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));
+        } catch (RejectedExecutionException e) {
+            promise.setFailure(e);
+        }
+        return promise.getFuture();
+    }
+
+    @Override
+    public <R> R execute(Callable<R> callable) throws ExecutionException, InterruptedException {
+        Future<R> future = asyncExecute(callable);
+        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){
+        executorService.shutdown();
+        try {
+            executorService.awaitTermination(10, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+}

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

@@ -0,0 +1,61 @@
+package code.concurrency.promise;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiConsumer;
+
+/**
+ * 〈Future〉<p>
+ *                                      +---------------------------+
+ *                                      | Completed successfully    |
+ *                                      +---------------------------+
+ *                                 +---->      isDone() = true      |
+ * +--------------------------+    |    |        result = non-null  |
+ * |        Uncompleted       |    |    +===========================+
+ * +--------------------------+    |    | Completed with failure    |
+ * |      isDone() = false    |    |    +---------------------------+
+ * |        result = null     |----+---->      isDone() = true      |
+ * | isCancelled() = false    |    |    |       cause() = non-null  |
+ * |       cause() = null     |    |    +===========================+
+ * +--------------------------+    |    | Completed by cancellation |
+ *                                 |    +---------------------------+
+ *                                 +---->      isDone() = true      |
+ *                                      | isCancelled() = true      |
+ *                                      +---------------------------+
+ *
+ * @author zixiao
+ * @date 2019/6/11
+ */
+public interface Future <R> {
+
+    boolean isDone();
+
+    boolean isSuccess();
+
+    boolean isCancelled();
+
+    boolean cancel();
+
+    /**
+     * 阻塞直到取得异步操作结果
+     * @return
+     * @throws InterruptedException
+     * @throws ExecutionException
+     */
+    R get() throws InterruptedException, ExecutionException;
+
+    /**
+     * 阻塞直到超时或取得异步操作结果
+     * @param timeout
+     * @param unit
+     * @return
+     * @throws InterruptedException
+     * @throws ExecutionException
+     * @throws TimeoutException
+     */
+    R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
+
+    Future<R> whenComplete(BiConsumer<? super R, ? super Throwable> callback);
+
+}