Java8 自定義CompletableFuture的原理解析

public class FutureInAction3 {
    public static void main(String[] args) {
        Future<String> future = invoke(() -> {
            try {
                Thread.sleep(10000L);
                return "I am Finished.";
            } catch (InterruptedException e) {
                return "I am Error";
            }
        });
        future.setCompletable(new Completable<String>() {
            @Override
            public void complete(String s) {
                System.out.println("complete called ---- " + s);
            }
            @Override
            public void exception(Throwable cause) {
                System.out.println("error");
                cause.printStackTrace();
            }
        });
        System.out.println("....do something else .....");
        System.out.println("try to get result ->" + future.get());
    }
    private static <T> Future<T> invoke(Callable<T> callable) {
        AtomicReference<T> result = new AtomicReference<>();
        AtomicBoolean finished = new AtomicBoolean(false);
        Future<T> future = new Future<T>() {
            private Completable<T> completable;
            @Override
            public T get() {
                return result.get();
            }
            @Override
            public boolean isDone() {
                return finished.get();
            }
            // 設置完成
            @Override
            public void setCompletable(Completable<T> completable) {
                this.completable = completable;
            }
            // 獲取
            @Override
            public Completable<T> getCompletable() {
                return completable;
            }
        };
        Thread t = new Thread(() -> {
            try {
                T value = callable.action();
                result.set(value);
                finished.set(true);
                if (future.getCompletable() != null)
                    future.getCompletable().complete(value);
            } catch (Throwable cause) {
                if (future.getCompletable() != null)
                    future.getCompletable().exception(cause);
            }
        });
        t.start();
        return future;
    }
    private interface Future<T> {
        T get();
        boolean isDone();
        //  1
        void setCompletable(Completable<T> completable);
        //  2
        Completable<T> getCompletable();
    }
    private interface Callable<T> {
        T action();
    }
    // 回調接口
    private interface Completable<T> {
        void complete(T t);
        void exception(Throwable cause);
    }
}

package demos;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
 * @author djh on  2019/4/22 10:23
 * @E-Mail [email protected]
 */
public class Demo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService cachePool = Executors.newCachedThreadPool();
        Future<String> future = cachePool.submit(() -> {
            Thread.sleep(3000);
            return "異步任務計算結果!";
        });
        // 提交完異步任務後, 主線程可以繼續幹一些其他的事情.
        doSomeThingElse();
        // 為瞭獲取異步計算結果, 我們可以通過 future.get 和 輪詢機制來獲取.
        String result;
        // Get 方式會導致當前線程阻塞, 這顯然違背瞭異步計算的初衷.
        // result = future.get();
        // 輪詢方式雖然不會導致當前線程阻塞, 但是會導致高額的 CPU 負載.
        long start = System.currentTimeMillis();
        while (true) {
            if (future.isDone()) {
                break;
            }
        }
        System.out.println("輪詢耗時:" + (System.currentTimeMillis() - start));        
        result = future.get();
        System.out.println("獲取到異步計算結果啦: " + result);
        cachePool.shutdown();
    }
    private static void doSomeThingElse() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("我的最重要的事情幹完瞭, 我要獲取異步計算結果來執行剩下的事情.");
    }
}

package demos;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * @author djh on  2019/5/1 20:26
 * @E-Mail [email protected]
 */
public class CompleteFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> completableFutureOne = new CompletableFuture<>();
        ExecutorService cachePool = Executors.newCachedThreadPool();
        cachePool.execute(() -> {
            try {
                Thread.sleep(3000);
                completableFutureOne.complete("異步任務執行結果");
                System.out.println(Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        // WhenComplete 方法返回的 CompletableFuture 仍然是原來的 CompletableFuture 計算結果.
        CompletableFuture<String> completableFutureTwo = completableFutureOne.whenComplete((s, throwable) -> {
            System.out.println("當異步任務執行完畢時打印異步任務的執行結果: " + s);
        });
        // ThenApply 方法返回的是一個新的 completeFuture.
        CompletableFuture<Integer> completableFutureThree = completableFutureTwo.thenApply(s -> {
            System.out.println("當異步任務執行結束時, 根據上一次的異步任務結果, 繼續開始一個新的異步任務!");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return s.length();
        });
        System.out.println("阻塞方式獲取執行結果:" + completableFutureThree.get());
        cachePool.shutdown();
    }
}