前言

在工作中, 經(jīng)常有一種需求, 就是等待所有線程執(zhí)行（并行）完成后, 再去執(zhí)行最后的某個(gè)操作. 在Java8之前用得最多的要屬CountDownLatch, 跟Java8點(diǎn)流式api相比缺點(diǎn)還是非常明顯的，本文介紹了處理該場景的3種方式。

1.經(jīng)典的CountDownLatch用法

CountDownLatch countDownLatch = new CountDownLatch(3);//線程數(shù)量
//線程里的run執(zhí)行計(jì)數(shù)
countDownLatch.countdown();
//最后
try {
    countDownLatch.await();
} catch (Exception e) {
    log.error(e.toString());
} finally {
    doSth();
}

CountDownLatch 有一個(gè)很明顯的缺點(diǎn)是如果某個(gè)線程的異常沒有捕捉，導(dǎo)致最后的語句部分不會(huì)被執(zhí)行，這在某些業(yè)務(wù)場景中不能忍受。

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("所有線程執(zhí)行完畢");
                    }
                })
                .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實(shí)現(xiàn)的并發(fā)。缺點(diǎn)也很明顯，各自的異常自行處理，否則無法往下運(yùn)行其他線程了。

3.Java8 CompletableFuture用法

@Slf4j
public class MyTest {
    //此executor線程池如果不傳,CompletableFuture經(jīng)測試默認(rèn)只啟用最多3個(gè)線程,所以最好自己指定線程數(shù)量
    ExecutorService executor = Executors.newFixedThreadPool(15);
    @Test
    public void ex() {
        long start = System.currentTimeMillis();
        //參數(shù)
        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("所有線程已執(zhí)行完[{}]",allFutures.isDone());
        allFutures.whenComplete(new BiConsumer<Void, Throwable>() {
            @Override
            public void accept(Void aVoid, Throwable throwable) {
                log.info("執(zhí)行最后一步操作");
                // doSth();
                long end = System.currentTimeMillis();
                log.info("耗時(shí):"+ (end-start)/1000L );
            }
        });
    }

    //executor 可不傳入,則默認(rèn)最多3個(gè)線程
    CompletableFuture<Void> handle(String pageLink) {
        return CompletableFuture.runAsync(() -> {
            //int i = 1/0;
            log.info("執(zhí)行任務(wù)"+pageLink);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        },executor).exceptionally(new Function<Throwable, Void>() { //捕捉異常,不會(huì)導(dǎo)致整個(gè)流程中斷
            @Override
            public Void apply(Throwable throwable) {
                log.info("線程[{}]發(fā)生了異常, 繼續(xù)執(zhí)行其他線程,錯(cuò)誤詳情[{}]",Thread.currentThread().getName(),throwable.getMessage());
                return null;
            }
        });
    }
}

此代碼運(yùn)用了Java8點(diǎn)Lambda表達(dá)式，Stream API特性，鏈?zhǔn)秸{(diào)用，極具RxJava風(fēng)格。

總結(jié)

本文介紹了就前言的業(yè)務(wù)場景，提供了3種常用的解決方法，各有所好吧！1和2兩種必須在最外層指明try catch，捕捉一切異常，否則其中一個(gè)線程有異常將終止運(yùn)行，第三種統(tǒng)一處理異常，邏輯清晰。