Hello大家好，我是Jensen。
今天分享一篇多线程编排，纯技术干货，不讲废话。
0x1.先写个Demo
我们以商品详情页为例，如果想在一个接口内返回商详中所有的数据，大概需要以下几个步骤:
根据商品id，请求商品详细数据（接口名：spuDetail，耗时：400ms）根据商品id和位置信息，请求该商品支持的配送区域（接口名：saleArea，耗时：350ms）根据1返回的数据，获取商品对应的店铺code，请求店铺信息（接口名：findShopInfo，耗时：400ms）根据1返回的数据，请求商品规格接口（接口名：spuSpecInfo，耗时：350ms）根据1返回的信息，请求该店铺的推荐商品（接口名：appSearch，耗时：400ms）根据3返回的信息，请求商品的优惠券信息（接口名：showCouponTag，耗时：350ms）一般的情况下，很多小伙伴都会按下面的写法一把梭哈，一个个请求，等到拿到数据之后进行商详页数据的拼装并返回。
为了更直观的看到整个过程，我们使用简单直观的方法来模拟接口请求：
public class SkuDetail {
public static String spuDetail(String skuCode) {
System.out.println(“商品基础信息===》” + skuCode);
mockTime(400L);
return “商品基础信息”;
}
public static String saleArea(String skuCode) {
System.out.println(“销售区域===》” + skuCode);
mockTime(350L);
return “销售区域”;
}
public static String findShopInfo(String shopCode) {
System.out.println(“店铺信息===》” + shopCode);
mockTime(400L);
return “店铺信息”;
}
public static String spuSpecInfo(String spuCode) {
System.out.println(“规格接口===》” + spuCode);
mockTime(350L);
return “规格接口”;
}
public static String appSearch(String shopCode) {
System.out.println(“推荐===》” + shopCode);
mockTime(400L);
return “推荐”;
}
public static String showCouponTag(String shopCode) {
System.out.println(“优惠券===》” + shopCode);
mockTime(350L);
return “优惠券”;
}
/**
* 模拟耗时
* @param time
*/
private static void mockTime(long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws InterruptedException, ExecutionException {
final long begin = System.currentTimeMillis();
String skuCode = “123”;
//根据商品id，请求商品详细数据
String spuDetail = spuDetail(skuCode);
// 根据商品id和位置信息，请求该商品支持的配送区域
String saleArea = saleArea(skuCode);
// 模拟根据1返回的数据，获取商品对应的店铺code，请求店铺信息
String shopCode = skuCode;
String shopInfo = findShopInfo(shopCode);
// 模拟根据1返回的数据，通过spuCode请求商品规格接口
String spuCode = skuCode;
String spuSpecInfo = spuSpecInfo(spuCode);
// 根据3返回的信息，请求该店铺的推荐商品
String recommend = appSearch(shopCode);
// 根据3返回的信息，请求商品的优惠券信息
String couponTag = showCouponTag(skuCode);
// 组装VO返回
SkuVO skuVO = new SkuVO(spuDetail, saleArea, shopInfo, spuSpecInfo, recommend, couponTag);
// 模拟返回
System.out.println(skuVO);
final long end = System.currentTimeMillis();
System.out.println(“整个过程耗时=================》” + (end – begin));
}
}
@Data
@NoArgsConstructor
@AllArgsConstructor
class SkuVO{
private String spuDetail;
private String saleArea;
private String shopInfo;
private String spuSpecInfo;
private String recommend;
private String couponTag;
}
程序运行结果：
商品基础信息===》123
销售区域===》123
店铺信息===》123
规格接口===》123
推荐===》123
优惠券===》123
SkuVO(spuDetail=商品基础信息, saleArea=销售区域, shopInfo=店铺信息, spuSpecInfo=规格接口, recommend=推荐, couponTag=优惠券)
整个过程耗时=================》2313
这样编写代码，整个商详页的数据获取需要耗时=================2313ms
这代码是早上写的，工位是下午收拾的，人是连夜扛着火车走的。
有些同学可能觉得看起来好像还行，能用就行，要啥自行车。但是仔细想想，这个接口还有挺大的优化空间：
仔细看看，接口虽然多，但是依赖性不大，如果要加快接口响应速度，可以尝试使用：多线程（依赖性不大，可以多线程请求）、Future（获取店铺信息时，需要商品基础信息返回后才能操作）、CountDownLatch（所有接口请求完之后，才能响应前端。
但是，上面说到的这些，你写我推荐，真写我不写。
为啥？因为这光想想就觉得复杂，本来多线程就是个挺复杂的玩意，这写起来太麻烦了。
人一懒，就总能想出偷懒的办法。
这不，JDK8就提供了一个用来处理类似问题的利器：CompletableFuture
0x2.多线程编排利器
CompletableFuture是JDK8新增的，它提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法。
它实现了Future接口，还有CompletionStage接口，这让他拥有Future的功能，也拥有一个阶段完成以后可能会触发另外一个阶段的能力。
CompletableFuture 提供了四个静态方法来创建一个异步操作。
public class CompletableFuture<T> implements Future<T>, CompletionStage<T>
这里其实分两组supplyAsync(带返回值)、runAsync（不带返回值），至于这两组下面带不带线程池的区别，只是使用者是否自定义线程池供异步任务执行使用。
不指定线程池的时候，其内部也是会有默认的ForkJoin池供异步执行任务使用。
我们来看看这东西怎么使用：
public static void main(String[] args) throws InterruptedException, ExecutionException {
final long begin = System.currentTimeMillis();
String skuCode = “123”;
SkuVO result = new SkuVO();
//根据商品id，请求商品详细数据
CompletableFuture<String> spuDetail = CompletableFuture.supplyAsync(() -> {
String detail = spuDetail(skuCode);
result.setSpuDetail(detail);
return detail;
});
// 根据商品id和位置信息，请求该商品支持的配送区域
CompletableFuture<Void> saleArea = CompletableFuture.runAsync(() -> result.setSaleArea(saleArea(skuCode)));
// spuDetail执行完之后，获取spuCode继续执行其他
CompletableFuture<String> shopInfo = spuDetail.thenApplyAsync(shopCode -> {
String info = findShopInfo(shopCode);
result.setShopInfo(info);
return info;
});
// 模拟根据1返回的数据，通过spuCode请求商品规格接口
CompletableFuture<Void> spuSpecInfo = spuDetail.thenAcceptAsync(spuCode -> result.setSpuSpecInfo(spuSpecInfo(spuCode)));
// 根据1返回的信息，请求该店铺的推荐商品
CompletableFuture<Void> recommend = spuDetail.thenAcceptAsync(shopCode ->
result.setRecommend(appSearch(shopCode))
);
// 根据1返回的信息，请求商品的优惠券信息
CompletableFuture<Void> couponTag = spuDetail.thenAcceptAsync(shopCode -> result.setCouponTag(showCouponTag(shopCode)));
// 等待所有任务完成
CompletableFuture.allOf(spuDetail, saleArea, shopInfo, spuSpecInfo, recommend, couponTag).get();
// 模拟返回
System.out.println(result);
final long end = System.currentTimeMillis();
System.out.println(“整个过程耗时=================》” + (end – begin));
}
程序运行结果：
商品基础信息===》123
销售区域===》123
店铺信息===》商品基础信息
规格接口===》商品基础信息
推荐===》店铺信息
优惠券===》店铺信息
SkuVO(spuDetail=商品基础信息, saleArea=销售区域, shopInfo=店铺信息, spuSpecInfo=规格接口, recommend=推荐, couponTag=优惠券)
整个过程耗时=================》1297
相比于没有做任务编排的请求方式（2313ms），进行多线程编排之后的速度快了近1倍。
0x3.深入了解CompletableFuture
在简单的了解了CompletableFuture的应用场景之后，我们可以仔细看看CompletableFuture。我们平时的使用中，基本上就使用到这个类里面的静态方法：
方法名作用supplyAsync(Supplier supplier)异步执行对应的supplier，带返回值的supplyAsync(Supplier supplier,Excutor excutor)使用自定义的线程池异步执行对应的supplier，带返回值的runAsync(Runnable runnable)异步执行对应的runnable任务，无返回值runAsync(Runnable runnable,Excutor excutor)使用自定义的线程池异步执行对应的runnable任务，无返回值completedFuture(U value)返回值为value的CompletedFuture，这个应用场景较少allOf(CompletableFuture… cfs)返回一个新的 CompletableFuture，它在所有给定的 CompletableFuture 完成时完成anyOf(CompletableFuture… cfs)返回一个新的 CompletableFuture，它在任何给定的 CompletableFuture 完成时完成，结果相同上面的方法中，除了不太常用的completedFuture，就只有anyOf我们在上面的程序中没有使用过。
这个有什么用呢？比如dubbo中，有一种策略是对所有服务端发起调用，只要有一个有返回数据，就将数据返回。这种场景下面就可以使用到我们的anyOf。
public class AnyOfDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> first = CompletableFuture.supplyAsync(getStringSupplier(5000L));
CompletableFuture<String> second = CompletableFuture.supplyAsync(getStringSupplier(6000L));
CompletableFuture<String> three = CompletableFuture.supplyAsync(getStringSupplier(3000L));
CompletableFuture<Object> result = CompletableFuture.anyOf(first, second, three);
System.out.println(result.get());
}
private static Supplier<String> getStringSupplier(Long time) {
return () -> {
System.out.println(time + “准备执行”);
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(time + “执行完毕”);
return String.valueOf(time);
};
}
}
程序输出结果：
6000准备执行
5000准备执行
3000准备执行
3000执行完毕
3000
我们可以看到，调用的时候是三个都执行，只要第一返回之后，其余的接调用就会中断。
0x4.实例方法任务编排
上面罗列出来的几个静态方法，可以用于产生CompleteFuture实例，之后便可以调用器实例方法，进行多线程任务的编排。
我们通过下面的例子来看看它提供的强大的API。
为避免过多无用的描述，做如下规定：
实例方法中，有很多名字为xxx、xxxAsync、xxxAsync,他们功能都是类似的，带Async表示其实另起线程异步执行，两个带Async的方法，入参一个有executor,表示使用指定的线程池执行，没有话用默认的线程池执行。后续描述中，只会出现xxx，另外两个方法不做描述。
isDone()
判断是否执行完成
public static void main(String[] args) throws InterruptedException {
CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(4000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(“执行完成”);
});
System.out.println(completableFuture.isDone());
Thread.sleep(5000L);
System.out.println(completableFuture.isDone());
}
输出结果：
false
执行完成
true
get()
用于获取执行结果
public static void main(String[] args) throws InterruptedException, ExecutionException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
System.out.println(integerCompletableFuture.get());
}
输出结果：
100
get(long timeout, TimeUnit unit)
带超时时间的get，规定时间内获取执行结果，如果超时会抛异常
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
try {
// 模拟耗时操作
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 100;
});
System.out.println(integerCompletableFuture.get(1000L, TimeUnit.MILLISECONDS));
}
运行结果：
Exception in thread “main” java.util.concurrent.TimeoutException
getNow(T valueIfAbsent)
如果完成，则返回结果值（或抛出任何遇到的异常），否则将入参作为返回值返回。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 100;
});
// 如果还没执行完成，就返回1000
System.out.println(integerCompletableFuture.getNow(1000));
}
运行结果：
1000
complete(T value)
如果尚未完成，将get和相关方法返回的值设置为给定值。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 100;
});
// 如果还没执行完成，将返回值设置1000
integerCompletableFuture.complete(1000);
System.out.println(integerCompletableFuture.get());
}
运行结果：
1000
completeExceptionally(Throwable ex)
如果get的时候仍未执行完成，抛出给定的异常
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 100;
});
// 如果还没执行完成，get的时候抛出异常
integerCompletableFuture.completeExceptionally(new TimeoutException(“超时了”));
System.out.println(integerCompletableFuture.get());
}
运行结果：
Exception in thread “main” java.util.concurrent.ExecutionException: java.util.concurrent.TimeoutException: 超时了
thenApply(Function<? super T,? extends U> fn)
执行完当前任务后，将当前任务返回值传递给另一个任务（fn），并【同步】执行该任务，执行结果封装成CompletableFuture<U>进行返回。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
//将第一个的返回值进行加工，并返回新的CompletableFuture
CompletableFuture<Integer> apply = integerCompletableFuture.thenApply(value -> value * 10);
System.out.println(apply.get());
}
运行结果：
1000
thenAccept(Consumer<? super T> action)
执行完当前任务后，将当前任务返回值传递给另一个任务（action）并【同步】执行给定任务，该方法无返回值
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
//将第一个的返回值进行加工，但是不会返回
integerCompletableFuture.thenAccept(value -> System.out.println(“在方法内消耗上一个返回值” + value));
}
运行结果：
在方法内消耗上一个返回值100
thenRun(Runnable action)
执行完当前任务后，【同步】执行对应的action，该方法不关注当前方法的返回值
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
System.out.println(“执行完毕”);
return 100;
});
//上一个执行完成后触发，无返回
integerCompletableFuture.thenRun(() -> System.out.println(“上一个执行完成后触发”));
}
执行结果：
执行完毕
上一个执行完成后触发
thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
在两个任务都执行完成后，将结果交给fn进行处理，处理完的结果包装成CompletableFuture<V>。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
System.out.println(“one 执行完毕”);
return 100;
});
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> {
System.out.println(“two 执行完毕”);
return 10;
});
// 将one、two两个的结果传递给func处理，并返回新的CompletableFuture
CompletableFuture<Integer> three = one.thenCombine(two, (v1, v2) -> v1 * v2);
System.out.println(three.get());
}
运行结果:1000
thenAcceptBoth(CompletionStage other,
BiConsumer<? super T, ? super U> action)在两个任务都执行完成后，将结果交给fn进行处理，本身也无返回值。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
System.out.println(“one 执行完毕”);
return 100;
});
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> {
System.out.println(“two 执行完毕”);
return 10;
});
// 将one、two两个的结果传递给func处理，不返回
one.thenAcceptBoth(two, (v1, v2) -> System.out.println(v1 * v2));
}
运行结果：1000
runAfterBoth(CompletionStage other,
Runnable action)在两个任务都执行完成后，执行对应的action，不关心两个任务的返回值，且本身也无返回值。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(“one 执行完毕”);
return 100;
});
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> {
System.out.println(“two 执行完毕”);
return 10;
});
long start = System.currentTimeMillis();
// 将one、two两个的结果传递给func处理，不返回
one.runAfterBoth(two, () -> {
long end = System.currentTimeMillis();
System.out.println(“程序耗时：” + (end – start) + “ms”);
System.out.println(“都运行完了”);
});
System.in.read();
}
运行结果：
two 执行完毕
one 执行完毕
程序耗时：2014ms
都运行完了
applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
两个任务，谁执行的快就用谁的返回值进行下一步的操作，执行结果包装成CompletableFuture<U>进行返回。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(“one 执行完毕”);
return 100;
});
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> {
System.out.println(“two 执行完毕”);
return 10;
});
// 将one、two两个先执行完的结果传递给func加工，返回新CompletableFuture
CompletableFuture<Integer> result = one.applyToEither(two, v -> v * 10);
System.out.println(result.get());
}
执行结果：
two 执行完毕
100
acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
两个任务，谁执行的快就用谁的返回值进行下一步的操作，无执行结果。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(“one 执行完毕”);
return 100;
});
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> {
System.out.println(“two 执行完毕”);
return 10;
});
// 将one、two两个先执行完的结果传递给func加工，不返回
one.acceptEither(two, System.out::println);
}
执行结果：
two 执行完毕
10
runAfterEither(CompletionStage other,
Runnable action)两个任务，只要有一个完成，就触发下一步操作。不关心上一步返回值，本身也无返回值。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(“one 执行完毕”);
return 100;
});
CompletableFuture<Integer> two = CompletableFuture.supplyAsync(() -> {
System.out.println(“two 执行完毕”);
return 10;
});
// 将one、two两个先执行完的结果传递给func加工，不返回
one.runAfterEither(two,()->{
System.out.println(“有一个执行完了”);
});
}
运行结果：
two 执行完毕
有一个执行完了
thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
处理完第一个任务，将第一个任务的返回值传递给第二个任务，由第二个任务执行后返回新的CompletableFuture。乍一看跟thenApply差不多。两者本质的区别在于：thenApply对第一个任务的返回值进行转换，thenCompose将第一个任务的返回值传递给另一个CompletableFuture。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
return 100;
});
// 将one执行完的结果传递给另外的completeFuture处理，返回新的CompletableFuture
CompletableFuture<Integer> integerCompletableFuture = one.thenCompose(value -> CompletableFuture.supplyAsync(() -> value * 10));
System.out.println(integerCompletableFuture.get());
}
运行结果：1000
whenComplete(BiConsumer<? super T, ? super Throwable> action)
跟thenApply类似，都是将上一次的结果进行加工，同时whenComplete支持对上一次处理发生的异常进行处理。不同的是：whenComplete 的返回值是上个任务的返回值其无法对结果的栈地址进行改变，无需显式return。而thenApply可以改变前任务的结果。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
return 100;
});
// whenComplete 无法改变上一个返回值的【地址】
CompletableFuture<Integer> integerCompletableFuture = one.whenComplete((value, exception) -> {
value = value * 10;
});
System.out.println(integerCompletableFuture.get());
}
输出结果：100
如果是对于实体里某个字段的变更，这种是可以的：
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Foo> one = CompletableFuture.supplyAsync(() -> {
Foo foo = new Foo();
foo.setId(1);
foo.setName(“tomcat”);
return foo;
});
// whenComplete 无法改变上一个返回值的【地址】
CompletableFuture<Foo> integerCompletableFuture = one.whenComplete((value, exception) -> {
value.setId(10);
value.setName(“test”);
});
System.out.println(integerCompletableFuture.get());
}
@Data
static class Foo{
private Integer id;
private String name;
}
输出结果：
Test.Foo(id=10, name=test)
handle(BiFunction<? super T, Throwable, ? extends U> fn)
跟whenComplete类似，不同的是：whenComplete对上次的执行结果进行加工，不会返回新的地址。handle接受上一次的结果，自己处理后有自己的返回值。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Foo> one = CompletableFuture.supplyAsync(() -> {
Foo foo = new Foo();
foo.setId(1);
foo.setName(“tomcat”);
return foo;
});
// handle自行处理后，有自己的返回值
CompletableFuture<Foo> integerCompletableFuture = one.handle((value, exception) -> {
Foo foo = new Foo();
foo.setName(value.getName());
foo.setId(value.getId());
return foo;
});
System.out.println(integerCompletableFuture.get());
}
@Data
static class Foo {
private Integer id;
private String name;
}
exceptionally(Function<Throwable, ? extends T> fn)
用于处理异常，当thenApply执行的时候发生异常，可以由exceptionaly捕获异常，并处理
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
return 0;
}).thenApply(value -> {
// 这里会有异常
return 100 / value;
}).exceptionally(e -> 10);
System.out.println(completableFuture.get());
}
输出结果：
10
好了 关于CompletableFuture的介绍就到这了，以后就可以很愉快的进行多线程的编排了。
老规矩，一键三连，日入两千，点赞在看，年薪百万！