并发工具类：异步神器CompletableFuture

并发工具类：异步神器CompletableFuture

为什么要搞一个CompletableFuture工具类

正好上次分享了函数式接口和Stream的使用，这次就分享一下CompletableFuture，里面也用到了大量的函数式接口

想方便的异步执行任务，就必须放到单独的线程中。继承Thread类，实现Runnable都不能拿到任务的执行结果，这时就不得不提创建线程的另一种方式了，实现Callable接口。

对于简单的场景使用Future并没有什么不方便。但是一些复杂的场景就很麻烦，
如2个异步任务，其中一个有结果就直接返回。Future用起来就不方便，因为想获取结果时，要么执行()方法，但是这样会阻塞线程，变成同步操作，要么轮询isDone()方法，但是比较耗费CPU资源。

Netty和Google guava为了解决这个问题，在Future的基础上引入了观察者模式（即在Future上addListener），当计算结果完成时通知监听者。

Java8新增的CompletableFuture则借鉴了Netty等对Future的改造，简化了异步编程的复杂性，并且提供了函数式编程的能力

串行关系

And 汇聚关系

Or汇聚关系

创建CompletableFuture对象

@FunctionalInterface
public interface Supplier<T> {T get();
}

Supplier在《强大的Stream》中已经介绍过了，是一个能获取返回值的函数式接口

CompletableFuture<Integer> intFuture = CompletableFuturepletedFuture(100);
// 100
System.out.());CompletableFuture<Void> voidFuture = CompletableFuture.runAsync(() -> System.out.println("hello"));
// null
System.out.());CompletableFuture<String> stringFuture = CompletableFuture.supplyAsync(() -> "hello");
// hello
System.out.());

计算结果完成时

因为入参是BiConsumer<? super T,? super Throwable>函数式接口，所以可以处理正常和异常的计算结果

whenComplete和whenCompleteAsync的区别如下

1. whenComplete：执行完当前任务的线程继续执行whenComplete的任务
2. whenCompleteAsync：把whenCompleteAsync这个任务提交给线程池来执行

CompletableFuture的所有方法的定义和whenComplete都很类似

1. 方法不以Async结尾意味着使用相同的线程执行
2. 方法以Async结尾意味着将任务提交到线程池来执行
3. 方法以Async结尾时可以用ForkJoinPoolmonPool()作为线程池，也可以使用自己的线程池

后续介绍的所有方法都只写一种case

CompletableFuture future = CompletableFuture.supplyAsync(() -> {return "hello";
}).whenComplete((v, e) -> {// helloSystem.out.println(v);
});
// hello
System.out.());

转换，消费，执行

CompletableFuture.supplyAsync(() -> {return "hello ";
}).thenAccept(str -> {// hello worldSystem.out.println(str + "world");
}).thenRun(() -> {// task finishSystem.out.println("task finish");
});

组合（两个任务都完成）

CompletableFuture future = CompletableFuture.supplyAsync(() -> {return "欢迎关注 ";
}).thenApply(t -> {return t + "微信公众号 ";
}).thenCombine(CompletableFuturepletedFuture("Java识堂"), (t, u) -> {return t + u;
}).whenComplete((t, e) -> {// 欢迎关注 微信公众号 Java识堂System.out.println(t);
});

组合（只需要一个任务完成）

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "欢迎关注微信公众号";
});
CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "Java识堂";
});
CompletableFuture future = future1.applyToEither(future2, str -> str);
// 欢迎关注微信公众号 Java识堂 随机输出
System.out.());

sleepRandom()为我写的一个随机暂停的函数

多任务组合

allOf的使用

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "欢迎关注";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "微信公众号";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "Java识堂";
});
// 欢迎关注 微信公众号 Java识堂
CompletableFuture.allOf(future1, future2, future3).thenApply(v ->Stream.of(future1, future2, future3).map(CompletableFuture::join).collect(Collectors.joining(" "))).thenAccept(System.out::print);

anyOf的使用

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "欢迎关注";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "微信公众号";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {sleepRandom();return "Java识堂";
});
CompletableFuture<Object> resultFuture = CompletableFuture.anyOf(future1, future2, future3);
// 欢迎关注 微信公众号 Java识堂 随机输出
System.out.());

异常处理

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {return 100 / 0;
}).thenApply(num -> {return num + 10;
}).exceptionally(throwable -> {return 0;
});
// 0
System.out.());

当然有一些接口能捕获异常

CompletableFuture future = CompletableFuture.supplyAsync(() -> {String str = null;return str.length();
}).whenComplete((v, e) -> {if (e == null) {System.out.println("正常结果为" + v);} else {// 发生异常了urrent.CompletionException: java.lang.NullPointerExceptionSystem.out.println("发生异常了" + e.toString());}
});

CompletableFuture future = new CompletableFuture();
futurepleteExceptionally(new RuntimeException());
// 抛出异常 urrent.ExecutionException: java.lang.RuntimeException
Object object = ();

参考博客

[0]/
[1]/
[2]/
好的系列文章
[1]
[2]
[3]
系列文章
[1]
[2]
[3]
好文
[4].html