並行流parallelStream 替換默認線程池commonPool

　　java8引入了stream流和並行流，極大的簡化了多線程的操作，但是有一點要注意，parallelStream和completablefuture默認都是使用commonPool，參考源碼：ForkJoinPool.commonPool();

項目所有流操作都是共享該池，當頻繁的用於阻塞型任務（IO流：http請求等）時會導致整個項目卡頓，parallelStream只適用於cpu密集型的任務，但是我們可以將parallelStream使用自定義線程就可以避免這個問題。

 

1.使用自定義線程池作為並行流的線程池。

 

    public static void main(String[] args) {

        String[] firstRange = new String[100];
        String[] secondRange = new String[100];

　　　　　//線程池1
        ForkJoinPool forkJoinPool = new ForkJoinPool(3);
        forkJoinPool.submit(() -> {
            Stream.of(firstRange).parallel().forEach((number) -> {
                try {
                    System.out.println("任務1的線程:" + Thread.currentThread().getName());
                    // do something slow
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                }
            });
        });

　　　　　//線程池2　
        ForkJoinPool forkJoinPool2 = new ForkJoinPool(2);
        forkJoinPool2.submit(() -> {
            Stream.of(secondRange).parallel().forEach((number) -> {
                try {
                    System.out.println("任務2的線程:" + Thread.currentThread().getName());
                    // do something slow
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                }
            });
        });

        System.out.println("ok!!!!!!!!!!!!!!!!!!!");
        try {
            Thread.sleep(1000000);
        } catch (Exception e) {
        }


    }

 

 

 

執行結果

 

 

可以看到第一個線程池使用了3個線程，第二個線程池使用了2個線程，線程數量可以自己指定，commonPool的默認線程數為cpu的核心數。

這樣IO密集型的多線程操作就可以不占用commonPool線程池了。

 

 

 

2.當我們想對並行計算的結果進行處理時：

 

方式1：在ForkJoinPool池中處理，即submit方法內部處理（即上面的示例代碼）。

 

方式2：當我們想把處理結果傳遞給主線程時，由於submit方法傳遞的是Runnable對象，沒有返回值，那么我們可以用對象數組配合join方法處理，比如以下代碼：

　　　　說明：該方式和CompletableFuture區別是：

　　　　　　CompletableFuture先提交異步線程執行任務，接下來可以做其他事情，需要得到結果的時候再通過CompletableFuture的返回參數得到。

　　　　　　該方式：提交異步線程並立即等待結果（這里只是示例代碼，可以做其他調整）

 

    public static void main(String[] args) {

        final Integer[] result = new Integer[1];
        String[] firstRange = new String[100000000];
        ForkJoinPool forkJoinPool = new ForkJoinPool(16);
        forkJoinPool.submit(() -> {

            Optional<Integer> reduce = Stream.of(firstRange).parallel().map((number) -> {
                return 1;
            }).reduce((integer, integer2) -> {
                //累加總次數
                return integer + integer2;
            });

            Integer integer = reduce.get();

            result[0] = integer;

            System.out.println("count:" + integer);
        }).join();

        System.out.println("並行計算結果:" + result[0]);

        System.out.println("ok!!!!!!!!!!!!!!!!!!!");


    }

執行結果

 

 

 

 

 

 

 

參考文章：https://blog.csdn.net/qq_32331073/article/details/85116698