分享一篇多线程编排,纯技术干货,不讲废话。
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 CompletableFutureT> implements FutureT>, CompletionStageT>
这里其实分两组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 spuDetail = CompletableFuture.supplyAsync(() -> {
String detail = spuDetail(skuCode);
result.setSpuDetail(detail);
return detail;
});
// 根据商品id和位置信息,请求该商品支持的配送区域
CompletableFuture saleArea = CompletableFuture.runAsync(() -> result.setSaleArea(saleArea(skuCode)));
// spuDetail执行完之后,获取spuCode继续执行其他
CompletableFuture shopInfo = spuDetail.thenApplyAsync(shopCode -> {
String info = findShopInfo(shopCode);
result.setShopInfo(info);
return info;
});
// 模拟根据1返回的数据,通过spuCode请求商品规格接口
CompletableFuture spuSpecInfo = spuDetail.thenAcceptAsync(spuCode -> result.setSpuSpecInfo(spuSpecInfo(spuCode)));
// 根据1返回的信息,请求该店铺的推荐商品
CompletableFuture recommend = spuDetail.thenAcceptAsync(shopCode ->
result.setRecommend(appSearch(shopCode))
);
// 根据1返回的信息,请求商品的优惠券信息
CompletableFuture 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 first = CompletableFuture.supplyAsync(getStringSupplier(5000L));
CompletableFuture second = CompletableFuture.supplyAsync(getStringSupplier(6000L));
CompletableFuture three = CompletableFuture.supplyAsync(getStringSupplier(3000L));
CompletableFuture
程序输出结果:
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 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 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 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 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 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 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进行返回。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
//将第一个的返回值进行加工,并返回新的CompletableFuture
CompletableFuture 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 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 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
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture one = CompletableFuture.supplyAsync(() -> {
System.out.println("one 执行完毕");
return 100;
});
CompletableFuture two = CompletableFuture.supplyAsync(() -> {
System.out.println("two 执行完毕");
return 10;
});
// 将one、two两个的结果传递给func处理,并返回新的CompletableFuture
CompletableFuture three = one.thenCombine(two, (v1, v2) -> v1 * v2);
System.out.println(three.get());
}
运行结果:1000
thenAcceptBoth(CompletionStage extends U> other,
BiConsumer super T, ? super U> action)
在两个任务都执行完成后,将结果交给fn进行处理,本身也无返回值。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
CompletableFuture one = CompletableFuture.supplyAsync(() -> {
System.out.println("one 执行完毕");
return 100;
});
CompletableFuture 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 one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("one 执行完毕");
return 100;
});
CompletableFuture 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进行返回。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("one 执行完毕");
return 100;
});
CompletableFuture two = CompletableFuture.supplyAsync(() -> {
System.out.println("two 执行完毕");
return 10;
});
// 将one、two两个先执行完的结果传递给func加工,返回新CompletableFuture
CompletableFuture 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 one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("one 执行完毕");
return 100;
});
CompletableFuture 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 one = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("one 执行完毕");
return 100;
});
CompletableFuture two = CompletableFuture.supplyAsync(() -> {
System.out.println("two 执行完毕");
return 10;
});
// 将one、two两个先执行完的结果传递给func加工,不返回
one.runAfterEither(two,()->{
System.out.println("有一个执行完了");
});
}
运行结果:
two 执行完毕
有一个执行完了
thenCompose(Function super T, ? extends CompletionStageU>> fn)
处理完第一个任务,将第一个任务的返回值传递给第二个任务,由第二个任务执行后返回新的CompletableFuture。乍一看跟thenApply差不多。两者本质的区别在于:thenApply对第一个任务的返回值进行转换,thenCompose将第一个任务的返回值传递给另一个CompletableFuture。
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture one = CompletableFuture.supplyAsync(() -> {
return 100;
});
// 将one执行完的结果传递给另外的completeFuture处理,返回新的CompletableFuture
CompletableFuture 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 one = CompletableFuture.supplyAsync(() -> {
return 100;
});
// whenComplete 无法改变上一个返回值的【地址】
CompletableFuture 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 one = CompletableFuture.supplyAsync(() -> {
Foo foo = new Foo();
foo.setId(1);
foo.setName("tomcat");
return foo;
});
// whenComplete 无法改变上一个返回值的【地址】
CompletableFuture 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 one = CompletableFuture.supplyAsync(() -> {
Foo foo = new Foo();
foo.setId(1);
foo.setName("tomcat");
return foo;
});
// handle自行处理后,有自己的返回值
CompletableFuture 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
, ? extends T> fn)
用于处理异常,当thenApply执行的时候发生异常,可以由exceptionaly捕获异常,并处理
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
return 0;
}).thenApply(value -> {
// 这里会有异常
return 100 / value;
}).exceptionally(e -> 10);
System.out.println(completableFuture.get());
}
输出结果:
10
好了 关于CompletableFuture的介绍就到这了,以后就可以很愉快的进行多线程的编排了。
0 条评论