如何將普通的Thread多線程改為Java8的parallelStream并發流

這篇文章將為大家詳細講解有關如何將普通的Thread多線程改為Java8的parallelStream并發流，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

概括總結

Java8的parallelStream并發流能達到跟多線程類似的效果，但它也不是什么善茬，為了得到跟上一版本的多線程類似的效果，一改再改，雖然最后改出來了，但是還是存在理解不了的地方。

怎樣得到一個parallelStream

理論上，你需要先有一個List<?>，任意類型的List都行，然后調用它的.parallelStream()方法就可以了。

對我這個例子來說，元素的類型不重要，因此選擇了Integer類型，核心代碼如下：

AtomicInteger atomicInteger = new AtomicInteger(0);
return Arrays.asList(new Integer[size]).parallelStream().map(i -> atomicInteger.incrementAndGet());

值得注意的是，第一行用的是AtomicInteger而不是Integer，因為Integer會存在并發問題

第二行的意思是：新建一個大小為size的數組，把數組轉成List，再把List轉成parallelStream，再把列表中的元素初始化成遞增的整數，最后返回。

為什么說parallelStream不是什么善茬

簡單來說，我認為原因是：因為它的默認值適用的場景是CPU密集型的，而一般的Web項目是IO密集型的（一般的Web項目都是需要跟數據庫打交道的，針對數據庫的操作主要就都是IO，而對CPU的消耗并不高）。

當不能使用默認值的時候，就需要開發人員額外去了解parallelStream的用法，而這些資料還不是特別好找。比如說：parallelStream默認的并發線程數是多少？怎么修改默認的線程數？

我最終找到了這篇問答：Custom thread pool in Java 8 parallel stream，供參考。

問題1： 在Java代碼中，怎樣獲取可用的CPU處理器的數量？代碼如下：(在我的機器8核上結果是：8)

Runtime.getRuntime().availableProcessors()

問題2： parallelStream默認的并發線程數是多少？代碼如下：(在我的機器上結果是：7)

ForkJoinPool.getCommonPoolParallelism()

問題3： 為什么parallelStream默認的并發線程數要比CPU處理器的數量少1個？

因為最優的策略是每個CPU處理器分配一個線程，然而主線程也算一個線程，所以要占一個名額。

問題4： 那如果電腦比較差，就只有1個CPU要怎么辦？那就不管了，默認的并發線程數就是1，總不能為零吧。

問題5： 默認的并發線程數太少了，要怎么修改？如代碼如下：（改成了20）

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");

問題6： 默認的并發線程數可以反復修改嗎？不能。因為java.util.concurrent.ForkJoinPool.common.parallelism是final類型的，整個JVM中只允許設置一次。

執行以下代碼：

int a=5;
for (;;) {
    System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "" + a++);
    System.out.println("ForkJoinPool.getCommonPoolParallelism() : " + ForkJoinPool.getCommonPoolParallelism());
    if(a>7)break;
}

/**
result:
ForkJoinPool.getCommonPoolParallelism() : 5
ForkJoinPool.getCommonPoolParallelism() : 5
ForkJoinPool.getCommonPoolParallelism() : 5
*/

問題7： 既然默認的并發線程數不能反復修改，那怎么進行不同線程數量的并發測試呢？答案是：引入ForkJoinPool。用法如下：

new ForkJoinPool(threadCount).submit(() -> {
    parallelStream.forEach(i -> {
        // 這里省略提交訂單的代碼
    });
}).get();

問題8： java.util.concurrent.ForkJoinPool.common.parallelism與new ForkJoinPool(threadCount)之間有什么關系？答案是：不知道。

這個答案很讓人失望，但是我確實沒有查出來。我這邊測試的結果是：

1.如果在new ForkJoinPool(threadCount)之前沒有設置java.util.concurrent.ForkJoinPool.common.parallelism的值，那么new ForkJoinPool(threadCount)的作用就不明顯，即就是說，改變threadCount的值對性能沒有多大影響。

2.如果在之前設置了java.util.concurrent.ForkJoinPool.common.parallelism的值，但是設置得比較小（比如32），則后續的new ForkJoinPool(threadCount)的作用也不明顯。

3.只有先把java.util.concurrent.ForkJoinPool.common.parallelism的值設置得比較大（比如10000），后續的new ForkJoinPool(threadCount)中threadCount改變之后，才對性能有明顯的影響。

問題9： 如果按問題8中的來修改，把java.util.concurrent.ForkJoinPool.common.parallelism的值設置得比較大（比如10000），就意味不再適用于CPU密集型的操作了，那應該怎么辦呢？

答案是：每次都用new ForkJoinPool(threadCount)，整體放棄使用默認的parallelStream。（那多麻煩啊

你看，隨隨便便就有這么多問題，頓時就不想用了，是不是。

并發線程數量”與“每秒能提交的訂單數量”之間的關系

這次測試的結果與上次測試的結果對比圖如下：（紅色為上一版本使用Thread測試的結果，黑色為這一版本使用parallelStream測試的結果）

可以看到差別并不大。

【備注】：不同的機器上的測試結果會不一樣，以上測試結果僅供參考。

關于如何將普通的Thread多線程改為Java8的parallelStream并發流就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。