背景
如果你想要在同一个CPU上执行几个松耦合的任务，同时防止因某个任务等待过长而阻塞线程的执行，那么你需要做的是充分利用CPU的核，让其足够忙碌，最大化程序的吞吐量从而实现并发。
并行与并发的区别：
并行：在同一个核上同时执行多个任务，任务不互相阻塞
并发：多个任务分发给多个核去执行
在java5中，已经引入了Future接口方便开发人员进行异步编程。由于其使用繁琐，代码复杂，不足以让我们编写简介并发代码，因此java8引入了CompletableFuture接口。
使用CompletableFuture构建异步应用
查询商品价格的例子，假设获取价格是一个远程服务，我们使用sleep 1秒来模拟此行为。
public class Shop {
private String name;
public Double getPrice(String product){
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return new Random().nextDouble() * product.charAt(0) + product.charAt(1);
}
}
List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
new Shop("LetsSaveBig"),
new Shop("MyFavoriteShop"),
new Shop("MyFavoriteShop2"),
new Shop("MyFavoriteShop3"),
new Shop("MyFavoriteShop4"),
new Shop("MyFavoriteShop5"),
new Shop("MyFavoriteShop6"),
new Shop("MyFavoriteShop7"),
new Shop("BuyItAll"));
-
情景一
试想此场景，我们需要根据某个商品名称去查询商品的价格，发货地等等一系列操作，我们可能会写出如下伪代码：
操作A
shop.getPrice(product)
操作B
操作C
...
这些操作直接没有什么关联性，上述代码中靠后的操作需要等待前面的操作执行完之后才能执行，造成了阻塞。
那么我们其实可以使用CompletableFuture来实现异步执行，下面的代码中每个操作都不需要等待前面的操作便能执行。
操作A
CompletableFuture<Double> completableFuture
= CompletableFuture.supplyAsync(() -> shop.getPrice(product));
操作B
操作C
...
Double d = completableFuture.get();
-
情景二
如果给定一个product和一个List<Shop>,想要获取所有shop中对此product的定价，该如何实现？
我们已经知道流的使用，按照常规思路，写出下列代码应该不难
List<Double> list = shops.stream() .map( (Shop s) -> s.getPrice(product) ) .collect(Collectors.toList());但是我们可不可以把map中获取价格的代码实现异步执行呢？答案当然是可以的。
其中一种操作是将流转为并行流，这里我们使用另一种方式：
List<CompletableFuture<Double>> list = shops.stream() .map( (Shop s) -> CompletableFuture.supplyAsync( () -> s.getPrice(product) ) ) .collect(Collectors.toList()); List<Double> list2 =list.stream() .map(CompletableFuture::join).collect(Collectors.toList()); -
情景三
如果你试过情景三种的实现方式后，你会发现其执行速度并没有多少提升。那么有没有方法能够让他更快点呢？我们可以通过调整线程池的大小，确保整体的计算不会因为线程都在等待I/O而发生阻塞。
List中有10个shop，我们可以调整线程池大小为10个。
final Executor executor = Executors.newFixedThreadPool(11, new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setDaemon(true); return t; } }); // CompletableFuture.supplyAsync()方法可以设置第二个参数 CompletableFuture.supplyAsync( () -> s.getPrice(product),executor )-
情景四
对两个异步操作进行流水线，第一个操作完成时，将其 结果作为参数传递给第二个操作。使用thenCompose连接。
List<CompletableFuture<String>> list = shops.stream() .map( (Shop s) -> CompletableFuture.supplyAsync( () -> s.getPrice(product), executor ) ) .map(c -> c.thenCompose( (Double d) -> CompletableFuture.supplyAsync( () ->d.doubleValue() + "---" + (d.doubleValue() - new Random().nextInt(10))) )) .collect(Collectors.toList()); -
情景五
将两个完全不相干的CompletableFuture对象的结果整合起来，而且你也不希望等到第一个任务完全结束才开始第二项任务 。使用thenCombine连接
List<CompletableFuture<String>> list = shops.stream() .map(s -> CompletableFuture.supplyAsync(() -> s.getPrice(product), executor) .thenCombine( CompletableFuture.supplyAsync( () -> new Random().nextInt(10) ), (d, c) -> d + "-----" + c )) .collect(Collectors.toList());
-
-
情景六
响应CompletableFuture的completion事件
一旦CompletableFuture计算得到结果，就得到一个相应。那么可以使用thenAccept
Stream<CompletableFuture<String>> list = shops.stream() .map((Shop s) -> CompletableFuture.supplyAsync(() -> s.getPrice(product), executor)) .map(c -> c.thenCompose( (Double d) -> CompletableFuture.supplyAsync(() -> d.doubleValue() + "---" + (d.doubleValue() - new Random().nextInt(10)),executor) )); list.map(c->c.thenAccept(System.out::println));
