前言
在工作中, 经常有一种需求, 就是等待所有线程执行（并行）完成后, 再去执行最后的某个操作. 在Java8之前用得最多的要属CountDownLatch, 跟Java8点流式api相比缺点还是非常明显的，本文介绍了处理该场景的3种方式。
1.经典的CountDownLatch用法
CountDownLatch countDownLatch = new CountDownLatch(3);//线程数量
//线程里的run执行计数
countDownLatch.countdown();
//最后
try {
countDownLatch.await();
} catch (Exception e) {
log.error(e.toString());
} finally {
doSth();
}
CountDownLatch 有一个很明显的缺点是如果某个线程的异常没有捕捉，导致最后的语句部分不会被执行，这在某些业务场景中不能忍受。
2.RxJava用法
@Slf4j
public class RxJavaTest {
@Test
public void ex() {
//final ExecutorService executor = Executors.newFixedThreadPool(10);
//final Scheduler scheduler = Schedulers.from(executor);
Observable.range(1, 10)
.flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
return Observable.just(integer)
.subscribeOn(Schedulers.io()) //可指定scheduler
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return integer.toString();
}
});
}
})
.doFinally(new Action() {
@Override
public void run() throws Exception {
//executor.shutdown();
log.info("所有线程执行完毕");
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String str) throws Exception {
try {
int i = 1 / 0;
log.info(str);
} catch (Exception e) {
log.error("异常:{}", e.getMessage(), e);
}
}
});
//Junit是单线程
try {
Thread.sleep(30 * 1000);
} catch (Exception e) {
}
}
}
使用flatMap实现的并发。缺点也很明显，各自的异常自行处理，否则无法往下运行其他线程了。
3.Java8 CompletableFuture用法
@Slf4j
public class MyTest {
//此executor线程池如果不传,CompletableFuture经测试默认只启用最多3个线程,所以最好自己指定线程数量
ExecutorService executor = Executors.newFixedThreadPool(15);
@Test
public void ex() {
long start = System.currentTimeMillis();
//参数
List<String> webPageLinks = Arrays.asList("A", "B", "C","D","E","F","G","H");
List<CompletableFuture<Void>> pageContentFutures = webPageLinks.stream()
.map(webPageLink -> handle(webPageLink))
.collect(Collectors.toList());
CompletableFuture<Void> allFutures = CompletableFuture.allOf(
pageContentFutures.toArray(new CompletableFuture[pageContentFutures.size()])
);
allFutures.join();
log.info("所有线程已执行完[{}]",allFutures.isDone());
allFutures.whenComplete(new BiConsumer<Void, Throwable>() {
@Override
public void accept(Void aVoid, Throwable throwable) {
log.info("执行最后一步操作");
// doSth();
long end = System.currentTimeMillis();
log.info("耗时:"+ (end-start)/1000L );
}
});
}
//executor 可不传入,则默认最多3个线程
CompletableFuture<Void> handle(String pageLink) {
return CompletableFuture.runAsync(() -> {
//int i = 1/0;
log.info("执行任务"+pageLink);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
},executor).exceptionally(new Function<Throwable, Void>() { //捕捉异常,不会导致整个流程中断
@Override
public Void apply(Throwable throwable) {
log.info("线程[{}]发生了异常, 继续执行其他线程,错误详情[{}]",Thread.currentThread().getName(),throwable.getMessage());
return null;
}
});
}
}
此代码运用了Java8点Lambda表达式，Stream API特性，链式调用，极具RxJava风格。
总结
本文介绍了就前言的业务场景，提供了3种常用的解决方法，各有所好吧！1和2两种必须在最外层指明try catch，捕捉一切异常，否则其中一个线程有异常将终止运行，第三种统一处理异常，逻辑清晰。
