Java教程

聊聊Java中CompletableFuture的使用

本文主要是介绍聊聊Java中CompletableFuture的使用，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

目录

多任务并行协作

任务串行执行 

结果组合运算 

 thenCombine和thenCompose

 thenAcceptBoth和runAfterBoth

acceptEither、runAfterEither和applyToEither 

java9的改进 

总结 

CompletableFuture是java8引入的一个异步类，它最大的优势是可以在创建的对象中传入一个回调对象，在任务结束后(done或throw exception)，自动调用回调对象的回调方法，而不用让主线程阻塞。

多任务并行协作

假如我们要做咖啡，有3个子任务可以并行执行：洗杯子、磨咖啡、烧水，这3步完成后，我们开始泡咖啡。这种需求我们一般怎么实现呢？

下面我们看一下，使用Future是怎么完成这个功能的。
先定义一个线程池：

public class MyThreadPoolExecutor {
    public static ExecutorService getThreadPoolExecutor(){
        return ThreadPoolExecutorFactory.THREAD_POOL;
    }

    private static class ThreadPoolExecutorFactory{
        private static int PROCESSOR_NUM = Runtime.getRuntime().availableProcessors();
        private static final ExecutorService THREAD_POOL = new ThreadPoolExecutor(PROCESSOR_NUM, PROCESSOR_NUM + 1, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(100));
    }
}

我们再来看看使用Future制作咖啡的过程，我们把每个步骤都用一个线程来执行，必须等待前面三步都执行完成后，我们才能开始泡咖啡

public class MakeTea {
    public static void main(String[] args){
        ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
        List<Future> list = new ArrayList<>(3);
        list.add(executor.submit(() -> washCup()));
        list.add(executor.submit(() -> hotWater()));
        list.add(executor.submit(() -> grindCoffee()));
        while (true){
            int i = 0;
            for (Future future : list){
                if (future.isDone()){
                    i ++;
                }
            }
            if (i == list.size()){
                break;
            }
        }
        executor.submit(() -> System.out.println("泡咖啡"));
		System.out.println("我是主线程");
    }

    private static String washCup(){
        System.out.println("洗杯子");
        return "洗杯子";
    }

    private static String hotWater(){
        System.out.println("烧水");
        return "烧水";
    }

    private static String grindCoffee(){
        System.out.println("磨咖啡");
        return "磨咖啡";
    }
}

上面的代码创建3个线程后，因为等待执行结果，阻塞了主线程，等3个步骤都执行完成，主线程才能执行。输出如下：

洗杯子
烧水
磨咖啡
我是主线程
泡咖啡

如果我们使用CompletableFuture来写，要怎么实现呢？代码如下：

public static void main(String[] args){
    ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
    CompletableFuture future1 = CompletableFuture.runAsync(() -> {
        try {
            washCup();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, executor);

    CompletableFuture future2 = CompletableFuture.runAsync(() -> {
        try {
            hotWater();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, executor);

    CompletableFuture future3 = CompletableFuture.runAsync(() -> {
        try {
            grindCoffee();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, executor);

    CompletableFuture.allOf(future1, future2, future3).thenAccept(
            r -> {
                System.out.println("泡咖啡");
            }
    );
    System.out.println("我是主线程");
}

上面代码输出结果如下，可以看到主线程并没有被阻塞

我是主线程
洗杯子
烧水
磨咖啡
泡咖啡

 上面的示例是多个任务之间的调度，最后一个任务必须等之前的3个任务都完成后(allOf)，才能执行。如果前面3个任务只有一个完成最后一个任务就可以执行,那就用anyOf方法，把上面代码中allOf改成anyOf，其他代码不变，执行结果如下：

我是主线程
洗杯子
泡咖啡
烧水
磨咖啡

注意：
1.anyOf方法返回的是Object对象而不是Void，这是跟allOf的一个很大的区别，我们要配置异常情况的回调对象，在allOf创建的CompletableFuture中是不可以的。看下面代码

CompletableFuture<Object> future4 = CompletableFuture.anyOf(future1, future2, future3);
    future4.thenApply(
            r -> {
                System.out.println("泡咖啡");
                return null;
            }
    );
    future4.exceptionally(e -> {
        e.printStackTrace();
        return null;
});

2.supplyAsync和runAsync区别是前者创建的是CompletableFuture<U>,后者创建的是CompletableFuture<Void>

任务串行执行 

CompletableFuture也支持任务串行执行，后面的任务依赖前面任务的执行结果。我们再举一个做果汁的例子，我们分串行三步: 洗水果 —> 切水果 -> 榨汁，看如下代码

public static void main(String[] args) throws InterruptedException {
    ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> washFruit(), executor);
    CompletableFuture<String> future2 = future1.thenApplyAsync(r -> StringUtils.isBlank(r) ? null : cutFruit());
    future2.thenApplyAsync(r -> StringUtils.isBlank(r) ? null : juicing());
    //为了主线程不立刻退出，以便查看结果
    Thread.sleep(100);
}

private static String washFruit() {
    System.out.println(Thread.currentThread().getName());
    System.out.println("洗水果");
    return "洗水果";
}

private static String cutFruit() {
    System.out.println(Thread.currentThread().getName());
    System.out.println("切水果");
    return "切水果";
}

private static String juicing() {
    System.out.println(Thread.currentThread().getName());
    System.out.println("榨汁");
    return "榨汁";
}

上面main函数执行后输出：

pool-1-thread-1
洗水果
ForkJoinPool.commonPool-worker-1
切水果
ForkJoinPool.commonPool-worker-1
榨汁

上面的代码就是一个串行执行的任务，这儿除了thenApply，还有thenAccept(Consumer<? super T> action)和thenRun(Runnable action)，这2个方法都不返回执行结果。

注意：上面方法中，thenApply、thenAccept、thenRun都有一个对应的Async方法，区别在于Async方法会从线程池中拿线程执行，而不带Async的方法在当前线程执行。所以如果上面代码中thenApplyAsync换成thenApply，执行结果如下：

pool-1-thread-1
洗水果
pool-1-thread-1
切水果
pool-1-thread-1
榨汁

结果组合运算 

 thenCombine和thenCompose

thenCombine用于组合2个CompletableFuture，对结果进行运算

thenCompose把第一个CompletableFuture的结果放到第二个CompletableFuture中进行运算

下面2段代码都是输出110

ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10, executor);
CompletableFuture<Integer> future2 = future1.thenCompose(r -> CompletableFuture.supplyAsync(() -> r + 100));
System.out.println(future2.get());

ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10, executor);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 100, executor);
CompletableFuture<Integer> future3 = future1.thenCombineAsync(future2, (x, y) -> x + y);
System.out.println(future3.get());

上面的组合方法其实用上一节讲的串行执行也可以完成，见如下代码

ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10, executor);
CompletableFuture<Integer> future2 = future1.thenApplyAsync(r -> r + 100);
System.out.println(future2.get());

 thenAcceptBoth和runAfterBoth

thenAcceptBoth用于对前面2个线程的结果进行组合运算，下面代码输出110

ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10, executor);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 100, executor);
CompletableFuture<Void> future3 = future1.thenAcceptBoth(future2, (x, y) -> System.out.println(x + y));

runAfterBoth用于等待前面2个线程之后执行第三个线程

ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> System.out.println("线程1"), executor);
CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> System.out.println("线程2"), executor);
CompletableFuture<Void> future3 = future1.runAfterBothAsync(future2, () -> System.out.println("线程3"));

上面代码输出：

线程1
线程2
线程3

acceptEither、runAfterEither和applyToEither 

这三个方法只取组合线程中执行最快的一个结果，看下面代码：

public static void main(String[] args) {
    CompletableFuture<String> f1 = CompletableFuture.supplyAsync(()-> getTask1());
    CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> getTask2());
    CompletableFuture<String> f3 = f1.applyToEither(f2,s -> s);
    System.out.println(f3.join());//输出task2

    CompletableFuture<String> f4 = CompletableFuture.supplyAsync(()-> getTask1());
    CompletableFuture<String> f5 = CompletableFuture.supplyAsync(() -> getTask2());
    CompletableFuture<Void> f6 = f4.runAfterEither(f5, () -> System.out.println("task3"));
    f6.join();//输出task3

    CompletableFuture<String> f7 = CompletableFuture.supplyAsync(()-> getTask1());
    CompletableFuture<String> f8 = CompletableFuture.supplyAsync(() -> getTask2());
    CompletableFuture<Void> f9 = f8.acceptEither(f7, s -> System.out.println(s));
    f9.join();//输出task2
	
}

private static String getTask1(){
    try {
        Thread.currentThread().sleep(2000);
        return "task1";
    } catch (InterruptedException e) {
        e.printStackTrace();
        return null;
    }
}

private static String getTask2(){
    try {
        Thread.currentThread().sleep(1000);
        return "task2";
    } catch (InterruptedException e) {
        e.printStackTrace();
        return null;
    }
}

同样注意:上面7个方法都存在对应的Async方法，会从线程池中取线程来执行。

java9的改进 

1.可以设置超时时间，超时后给一个默认值，比如下面代码输出100

public static void main(String[] args){
    ExecutorService executor = MyThreadPoolExecutor.getThreadPoolExecutor();
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> getNum(),     executor);
    future1.completeOnTimeout(100, 3000, TimeUnit.MILLISECONDS);
    System.out.println(future1.get());
}


private static Integer getNum() {
        try {
            Thread.currentThread().sleep(50000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 10;
    }

2.增加public static <U> CompletableFuture<U> failedFuture(Throwable ex)，跟public static <U> CompletableFuture<U> completedFuture(U value)配对，前者创建一个指定给定值的CompletableFuture，后者创建一个指定异常的CompletableFuture
3.增加了public static <U> CompletionStage<U> completedStage(U value)和public static <U> CompletionStage<U> failedStage(Throwable ex)，这2个方法返回CompletableFuture的继承类MinimalStage
4.增加了public Executor defaultExecutor()和public <U> CompletableFuture<U> newIncompleteFuture()，可以让子类自己去实现

总结 

CompletableFuture类对多线程调度的支持还是挺强大的，本文主要介绍了一些常用的方法，对于其他方法，大家可以查看api或者CompletionStage接口中定义的方法选择使用。

 

微信公众号，欢迎关注，共同学习进步

 

这篇关于聊聊Java中CompletableFuture的使用的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
栏目导航