文章目录
- 1. 到底遇到了什么问题?
- 2. CountDownLatch搞起?
- 3. allOf里面的坑
- 4. 优化建议:
1. 到底遇到了什么问题?
最近看到组里面的同学遇到了这样的业务场景:
主线程需要异步并发调用多个接口,并且主线程需要等待线程全部执行完成之后,返回执行结果,业务流程如下:
这里面有一个注意点:一旦异步线程有一个失败,我主线程就不等待了,这种需求很常见,比如多线程拼装对象数据等等。那我们如何解决呢?
一定要坚持看完!细节往往决定成败,写代码也一样!!收获多多
2. CountDownLatch搞起?
此时有可能有可能第一时间会想到CountDownLatch,但是CountDownLatch本质内部是采用计数器,主线程一直调用await()阻塞等待,需要等待所有子线程执行完之后(不管成功或者失败)主线程才会往下执行,且主线程和子线程无法传递异常,并且需要注意异常的抛出。
这块可以参考之前写的CountDownLatch文章:
关于CountDownLatch的底层源码和闭坑指南,只看这一篇就够了!!
这里面有常见的坑一栏,可以重点看下,保你必有收获,所以CountDownLatch并不是最优方案。
这块很多同学会选用CompletableFuture来实现,因为CompletableFuture底层有丰富的任务编排和链式调用,并且此时肯定会有很多同学说CompletableFuture.allOf可以轻松实现。
此时如果你不自己去试试的话,八股文一背,此时你就被带偏了!!我们可以去看一下,在接下来会给出一个allOf的使用场景,把这个问题重现一下:
3. allOf里面的坑
场景:当有一批任务交给线程池执行,我们需要获取所有线程的返回结果。
首先定义一个并发执行器类:
public class CompletableFutureEngine {
private final static ExecutorService executorService = Executors.newFixedThreadPool(4);
/**
* 创建并行任务并执行
*
* @param list 数据源
* @param function API调用逻辑
* @param exceptionHandle 异常处理逻辑
* @return 处理结果列表
*/
public static <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> function, Consumer<Throwable> exceptionHandle) {
List<CompletableFuture<T>> completableFutures = list.stream()
.map(s -> CompletableFuture.supplyAsync(() -> function.apply(s)))
.collect(Collectors.toList());
List<T> results = new ArrayList<>();
try {
CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).join();
for (CompletableFuture<T> completableFuture : completableFutures) {
results.add(completableFuture.get());
}
} catch (Exception e) {
if (e instanceof CompletionException) {
if (e.getCause() != null) {
exceptionHandle.accept(e.getCause());
}
}
}
return results;
}
}
调用方法:
public class EngineDemo {
private static void sleep(long sleepTime) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static void currentDate(String str) {
// 创建一个Date对象,它包含了当前时间
Date now = new Date();
// 创建一个SimpleDateFormat对象,用于指定输出格式
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// 使用format方法将Date对象格式化为字符串
String currentTime = dateFormat.format(now);
// 打印当前时间
System.out.println(str + "是: " + currentTime);
}
public static void main(String[] args) {
currentDate("执行前");
List<Integer> numList = CompletableFutureEngine.parallelFutureJoin(Arrays.asList(1, 3, 5),
num -> {
sleep(num * 1000);
if (num == 1) {
throw new BusinessException("心别太大");
}
return num;
}
, e -> {
if (e instanceof BusinessException) {
System.out.println("BusinessException =" + e.getMessage());
} else {
System.out.println("Exception entrance");
}
}
);
System.out.println(numList);
currentDate("执行后");
}
}
此时,你会发现,程序的执行时间为5秒钟,感觉似乎和想的不一样,因为在代码中,是根据num决定的休眠时间,因此在第一印象中,应该是第一秒执行完就会抛出异常。尴尬了,现在从日志看程序的执行时间是5秒,那我们看下原因:走起,那我们debug下代码:
你会发现当程序抛出异常的时候,发现传进来的三个CompletableFuture,不管是成功还是失败都执行完了。
这也就说明了当抛出异常后,allOf并不会及时感知异常,而是等所有任务都执行完之后才往下继续运行,那此处会有两种情况:
- 使用完allOf之后,还要去做流程编排,不去直接get的话,这种方式没有问题。
- 需要汇聚所有的子线程执行结果返回给主线程,这种allOf可能效率会低,因为需要等待所有的子线程执行完才会去返回最终的结果,那如果遇到主要一个子线程失败,执行就失败,就会导致执行时间短的失败了,但是有个任务执行时间很长,返回时间也会变长,从而导致主线程等待的时间变长.
那针对于第二种情况应该如何优化呢!!!继续往下看!!!
4. 优化建议:
可以用CompletableFutures.stream().map(CompletableFuture::join).collect(Collectors.toList()) 来代替 allOf。这个方法只要有一个子线程出现异常,主线程就会感知到异常,不用等待其他线程执行完。
优化后的并发执行器如下:
public class CompletableFutureEngine2 {
/**
* 创建并行任务并执行
*
* @param list 数据源
* @param function API调用逻辑
* @return 处理结果列表
*/
public static <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> function, Consumer<Throwable> exceptionHandle) {
List<CompletableFuture<T>> completableFutures = list.stream()
.map(s -> CompletableFuture.supplyAsync(() -> function.apply(s)))
.collect(Collectors.toList());
List<T> results = null;
try {
results = completableFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());
} catch (Exception e) {
if (e instanceof CompletionException) {
if (e.getCause() != null) {
exceptionHandle.accept(e.getCause());
}
}
}
return results;
}
}
调用demo如下:
public class EngineDemo {
private static void sleep(long sleepTime) {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static void currentDate(String str) {
// 创建一个Date对象,它包含了当前时间
Date now = new Date();
// 创建一个SimpleDateFormat对象,用于指定输出格式
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
// 使用format方法将Date对象格式化为字符串
String currentTime = dateFormat.format(now);
// 打印当前时间
System.out.println(str + "是: " + currentTime);
}
public static void main(String[] args) {
currentDate("执行前");
List<Integer> numList = CompletableFutureEngine2.parallelFutureJoin(Arrays.asList(1, 3, 5),
num -> {
sleep(num * 1000);
if (num == 1) {
throw new BusinessException("心别太大");
}
return num;
}
, e -> {
if (e instanceof BusinessException) {
System.out.println("BusinessException =" + e.getMessage());
} else {
System.out.println("Exception entrance");
}
}
);
System.out.println(numList);
currentDate("执行后");
}
}
运行完之后,我们再去看下执行日志
你会发现,因为子线程的睡眠时间为传进来的num值,当num=1时,触发告警,因此主线程在等待一秒中就会感知到异常,郑如日志打印的时间间隔为1秒钟。
此时再去想想:那此时抛出异常时,num为3,5执行状态时怎么样的,再去debug一下触发异常这块代码:
你会发现当num=1抛出异常的时候,其他两个线程的执行状态还是未完成,这就是和allOf这个方法最大的区别。
是否感觉只有自己试验过才会有更大的收获,今天就先分享到这,后面干货多多!!
感觉对您有所启发的话,记得帮忙点赞,收藏加关注,并且分享给需要的小伙伴!!加油!!