前言：

并行編程勢(shì)不可擋，Java從1.7開(kāi)始就提供了Fork/Join 支持并行處理。java1.8 進(jìn)一步加強(qiáng)。

并行處理就是將任務(wù)拆分子任務(wù)，分發(fā)給多個(gè)處理器同時(shí)處理，之后合并。

1、Stream API

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

執(zhí)行原理如下，流分串行和并行兩種執(zhí)行方式

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

2、ParallelStreams執(zhí)行原理

并行執(zhí)行時(shí)，java將流劃分為多個(gè)子流，分散在不同CPU并行處理，然后進(jìn)行合并。

并行一定比串行更快嗎？這不一定，取決于兩方面條件：

• 處理器核心數(shù)量，并行處理核心數(shù)越多自然處理效率會(huì)更高。
• 處理的數(shù)據(jù)量越大，優(yōu)勢(shì)越強(qiáng)。這也很好理解，比如十個(gè)人干一個(gè)人就能完成的活兒會(huì)比它自己干更便宜？

3、ParallelStreams注意事項(xiàng)

使用并行流時(shí)，不要使用collectors.groupingBy,collectors.toMap，替代為

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

原因，并行流執(zhí)行時(shí)，通過(guò)操作Key來(lái)合并多個(gè)map的操作比較昂貴。詳細(xì)大家可以查看官網(wǎng)介紹。

https://docs.oracle.com/javase/tutorial/collections/streams/parallelism.html#concurrent_reduction

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 默認(rèn)使用 ForkJoinPool.commonPool()線程池。

注意：默認(rèn)情況下，你寫(xiě)的 ParallelStreams 都是通過(guò)該線程池調(diào)度執(zhí)行，整個(gè)應(yīng)用程序都共享這個(gè)線程池。

看一個(gè)例子，我們查詢(xún)一批新聞數(shù)據(jù)，可以利用并行化來(lái)處理遠(yuǎn)程新聞下載。

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

因?yàn)槭蔷W(wǎng)絡(luò)操作，存在很多不確定性，假如某個(gè)任務(wù)運(yùn)行時(shí)間較長(zhǎng)，導(dǎo)致線程池資源占據(jù)，阻塞其它線程，這樣就阻止了其他的并行流任務(wù)正常進(jìn)行。

如果解決這個(gè)問(wèn)題的其中一種方式，進(jìn)行線程池隔離。那么如何自定義并行流的線程池呢？

ForkJoinPool 構(gòu)造參數(shù)我們默認(rèn)設(shè)置為CPU核心數(shù)。

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

總結(jié)：

Java 1.8 提供的Stream API簡(jiǎn)化了代碼，很好用。不過(guò)在使用過(guò)程中應(yīng)該注意以上問(wèn)題。

到此這篇關(guān)于Java8 中的并行流 ParallelStreams的文章就介紹到這了,更多相關(guān)Java8 ParallelStreams內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論