CompletableFuture引入

一、Future介绍　　　　

Future
以前我们如果有个方法action执行，比如去数据库中查询数据、上传一个文件等，这个方法执行10分钟，调用者就需要等10分钟。
基于此，调用者可以先执行action，返回一个票据future，然后可以继续做其他的事情，这样就不阻塞了，完全异步化。然后可以根据future拿到action的结果。

自己怎么实现一个Future呢？

代码例子：

package com.cy.java8;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class FutureInAction {
    public static void main(String[] args) throws InterruptedException {
        Future<String> future = invoke(() -> {
            try {
                Thread.sleep(10000);
                return "I am finished";
            } catch (InterruptedException e) {
                return "error";
            }
        });

        System.out.println(future.get());
        System.out.println(future.get());
        System.out.println(future.get());
        //do other thing...

        while (!future.isDone()) {
            Thread.sleep(10);
        }
        System.out.println(future.get());
    }

    /**
     * 以前阻塞式的方式
     *
     * @param callable
     * @param <T>
     * @return
     */
    private static <T> T block(Callable<T> callable) {
        return callable.action();
    }

    /**
     * Future异步方式
     *
     * @param callable
     * @param <T>
     * @return
     */
    private static <T> Future<T> invoke(Callable<T> callable) {
        AtomicReference<T> result = new AtomicReference<>();
        AtomicBoolean finished = new AtomicBoolean(false);
        Thread t = new Thread(() -> {
            T value = callable.action();
            result.set(value);
            finished.set(true);
        });
        t.start();

        Future<T> future = new Future<T>() {
            @Override
            public T get() {
                return result.get();
            }

            @Override
            public boolean isDone() {
                return finished.get();
            }
        };
        return future;
    }

    private interface Future<T> {
        T get();

        boolean isDone();
    }

    private interface Callable<T> {
        T action();
    }
}

console打印：

null
null
null
I am finished

二、jdk自带的Future介绍

package com.cy.java8;

import java.util.concurrent.*;

public class FutureInAction2 {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        /**
         * 创建了一个线程池，但这个线程池里面只有1根线程，单线程
         * 这个线程是可以复用的，它执行完之后不会消失，会始终保持等待另外一个任务去执行
         * 所以main函数执行结束之后，它不会退出的，所以还要shutdown
         */
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future<String> future = executorService.submit(() -> {
            try {
                Thread.sleep(10000);
                return "I am finished.";
            } catch (InterruptedException e) {
                return "I am error.";
            }
        });

        //do other thing...

        /**
         * jdk的future和我们自己写的不一样，它get的时候会阻塞住，如果没有结果它就阻塞住了，一直等结果
         */
        //String value = future.get();
        //System.out.println(value);

        //最多等10秒，如果没有拿到结果就拉倒，抛TimeoutException
        //String value2 = future.get(10000, TimeUnit.MILLISECONDS);
        //System.out.println(value2);

        while(!future.isDone()){
            Thread.sleep(10);
        }
        System.out.println(future.get());

        executorService.shutdown();
    }
}

三、CompletableFuture引入

能不能future的任务执行完成之后，你来回调我呢？异步回调，你完成了之后来回调我。
就像ajax的异步回调，post请求完成之后回调函数，来执行其他的东西。
CompletableFuture最大的作用就是干这个的。

在讲CompletableFuture之前，先说下它的原理，自己模拟实现一个CompletableFuture.

模拟代码例子：

package com.cy.java8;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class FutureInAction3 {
    public static void main(String[] args) {
        invoke(() -> {
            try {
                Thread.sleep(3000);
                return "I am finished.";
            } catch (InterruptedException e) {
                return "I am error.";
            }
        }).setCompletable(new Completable<String>() {
            @Override
            public void complete(String s) {
                System.out.println("complete. value = " + s);
            }

            @Override
            public void exception(Throwable e) {
                System.out.println("error...");
                e.printStackTrace();
            }
        });

        //do other thing...
        System.out.println("do other thing...");
    }

    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();

        void setCompletable(Completable<T> completable);

        Completable<T> getCompletable();
    }

    private interface Completable<T> {
        void complete(T t);

        void exception(Throwable cause);
    }

    private interface Callable<T> {
        T action();
    }
}

console打印：

do other thing...
complete. value = I am finished.　　

----