并行流parallel，CompletableFuture与Executors线程池的使用与区别

清疚 2021-06-22 15:38 442阅读 0赞

list并行操作在项目开发可以极大提高代码效率与性能，java8对其进行了很好的封装，简单使用研究一下：

1. 先自己创建一个list：

// list在实际使用中要注意线程安全，Collections.synchronizedList写操作性能高，CopyOnWriteArrayList读操作性能较好
    List<String> list = Arrays.asList(new String[10000]);

2. parallel并行流使用：

list.stream().parallel().forEach(a -> {
                // 操作代码.....
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

并行流特点：

基于服务器内核的限制，如果你是八核，每次线程只能起八个，不能自定义线程池;
    适用于对list密集计算操作充分利用CPU资源，如果需要调用远端服务不建议使用；

3. CompletableFuture使用

3.1 未使用自定义线程池：

// supplyAsync需要有返回值，runAsync不需要有返回值
    list.stream().map(a -> CompletableFuture.supplyAsync(() -> {
                 // 操作代码.....
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return a;
            })).collect(Collectors.toList()).stream().map(CompletableFuture::join).collect(Collectors.toList());

划重点：

未自定义线程池时默认线程池跟并行流一样，都是根据服务器内核数创建线程数量。

3.2 使用自定义线程池：

ExecutorService executor = Executors.newFixedThreadPool(Math.min(list.size(), 100));
    
    list.stream().map(a -> CompletableFuture.supplyAsync(() -> {
                // 操作代码.....
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return a;
            }, executor)).collect(Collectors.toList()).stream().map(CompletableFuture::join).collect(Collectors.toList());

补充：

1. 线程数量的计算公式：

T(线程数) = N(服务器内核数) * u(期望cpu利用率) * （1 + E(等待时间)/C(计算时间));

2. 获取服务器内核数：

int count = Runtime.getRuntime().availableProcessors();

3.划重点:

此处join方法和CompletableFuture的get()方法类似，都是阻塞线程，等待结果，但是join方法不抛异常，不需要处理异常，让你代码更方便，get方法抛异常。

4. Executors使用（有多种线程池）

list.forEach(a ->
                executor.submit(() -> {
                    // 操作代码.....
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                })
            );
            executor.shutdown();
            while (true) {
                if (executor.isTerminated()) {
                    System.out.println("线程执行完毕!!!");
                    break;
                }
                Thread.sleep(10);
            }

5. 简单总结：

可以将代码粘贴到idea中运行把运行时间打出来看看效果，最后发现：

*  parallel与未定义自线程池的CompletableFuture效果差别不大，原因是底层都使用的默认的线程池;
 *  CompletableFuture自定义线程池与Executors的运行效果差别不大，但CompletableFuture有很多组合式的异步编程方法：  
    `runAsync：异步执行没有返回值； supplyAsync：异步执行有返回值； thenApply：继续执行当前线程future完成的函数，不需要阻塞等待其处理完成； thenApplyAsync：在不同线程池异步地应用参数中的函数； thenCompose：用于多个彼此依赖的futrue进行串联起来 thenCombine：并联起两个独立的future，注意，这些future都是在长时间计算都完成以后`