使用Java8 新特性处理多线程并行

前言

在工作中, 经常有一种需求, 就是等待所有线程执行（并行）完成后, 再去执行最后的某个操作. 在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，捕捉一切异常，否则其中一个线程有异常将终止运行，第三种统一处理异常，逻辑清晰。