你使用過Java8中的parallelStream？

前言

並行編程勢不可擋，Java從1.7開始就提供了Fork/Join 支持並行處理。java1.8 進一步增強。

並行處理就是將任務拆分子任務，分發給多個處理器同時處理，以後合併。

Stream API

Java 8 引入了許多特性，Stream API是其中重要的一部分。區別 InputStream OutputStream，Stream API 是處理對象流而不是字節流。

執行原理以下，流分串行和並行兩種執行方式

// 串行執行流
stream().filter(e -> e > 10).count();
// 並行執行流
.parallelStream().filter(e -> e > 10).count()

ParallelStreams執行原理

並行執行時，java將流劃分爲多個子流，分散在不一樣CPU並行處理，而後進行合併。

並行必定比串行更快嗎？這不必定，取決於兩方面條件：

1. 處理器核心數量，並行處理核心數越多天然處理效率會更高。
2. 處理的數據量越大，優點越強。這也很好理解，好比十我的幹一我的就能完成的活兒會比它本身幹更便宜？

ParallelStreams注意事項

使用並行流時，不要使用collectors.groupingBy,collectors.toMap，替代爲

collectors.groupingByConcurrent , collectors.toConcurrentMap，或直接使用串行流。

緣由，並行流執行時，經過操做Key來合併多個map的操做比較昂貴。詳細你們能夠查看官網介紹。

https://docs.oracle.com/javas...

Map<String, List<Person>> byGender = 
  roster
      .stream()
      .collect(Collectors.groupingBy(Person::getGender));

ConcurrentMap<String, List<Person>> byGender =         
  roster
      .parallelStream()
      .collect(Collectors.groupingByConcurrent(Person::getGender));

ParallelStreams 默認使用 ForkJoinPool.commonPool()線程池。

注意：默認狀況下，你寫的 ParallelStreams 都是經過該線程池調度執行，整個應用程序都共享這個線程池。

看一個例子，咱們查詢一批新聞數據，能夠利用並行化來處理遠程新聞下載。

public List<News> queryNews(Stream<String> ids) {
     return ids.parallel()
            .map(this::getNews) // 網絡操做，新聞下載
            .collect(toList());
}

由於是網絡操做，存在不少不肯定性，假如某個任務運行時間較長，致使線程池資源佔據，阻塞其它線程，這樣就阻止了其餘的並行流任務正常進行。

若是解決這個問題的其中一種方式，進行線程池隔離。那麼如何自定義並行流的線程池呢？

ForkJoinPool 構造參數咱們默認設置爲CPU核心數。

ForkJoinPool customThreadPool = new ForkJoinPool(4);
long actualTotal = customThreadPool
  .submit(() -> roster.parallelStream().reduce(0, Integer::sum)).get();

針對 Stream API 一些侷限性，Github上有個開源庫作了補充。

https://github.com/pivovarit/...

總結

Java 1.8 提供的Stream API簡化了代碼，很好用。不過在使用過程當中應該注意以上問題。

歡迎你們留言交流，一塊兒學習分享！！！