并行流

并行流是一個(gè)把元素分成多個(gè)塊的流，每個(gè)塊用不同的線程處理。可以自動(dòng)分區(qū)，讓所有的處理器都忙起來(lái)。

假設(shè)要寫(xiě)一個(gè)方法，接受一個(gè)數(shù)量n做參數(shù)，計(jì)算1-n的和。可以這樣實(shí)現(xiàn)：

public long sequentialSum(long n) { return Stream.iterate(1L, i -> i + 1).limit(n).reduce(0L, Long::sum); }

也許可以使用parallel方法，簡(jiǎn)單地使用并行計(jì)算，提高程序性能：

public long sequentialSum(long n) { return Stream.iterate(1L, i -> i + 1).limit(n).parallel().reduce(0L, Long::sum); }

這樣，流可能在內(nèi)部被分成多個(gè)塊，導(dǎo)致reduction操作可以在不同的塊上互不依賴(lài)地并行地各自工作。最后，reduction操作組合每個(gè)子流的并行reductions的返回值，返回的結(jié)果就是整個(gè)流的結(jié)果。見(jiàn)下面的示意圖

實(shí)際上，調(diào)用parallel方法，流自身不會(huì)有任何變化。在內(nèi)部，設(shè)置一個(gè)布爾類(lèi)型的標(biāo)記，標(biāo)明你想在并行模式執(zhí)行操作，接下來(lái)的操作都是并行的。

類(lèi)似地，你也可以使用sequential方法，把并行流轉(zhuǎn)成串行的。你也許認(rèn)為可以組合這兩個(gè)方法：

stream.parallel() .filter(...) .sequential() .map(...) .parallel() .reduce();

但是，最后一次調(diào)用parallel或者sequential才會(huì)全局地影響管道。上面的例子，管道將被并行地執(zhí)行。

配置并行流使用的線程池

并行流內(nèi)部使用ForkJoinPool。默認(rèn)地，線程數(shù)量等于處理器數(shù)量（Runtime.getRuntime().availableProcessors()）。但是，可以修改系統(tǒng)屬性java.util.concurrent.ForkJoinPool.common.parallelism，配置線程數(shù)量。

這是全局配置，所以，除非你認(rèn)為對(duì)性能有幫助，否則不要修改。

測(cè)量流的性能

我們聲稱(chēng)并行加法應(yīng)該比串行的或者自己的迭代方法快。我們可以使用JMH測(cè)量一下。這是一個(gè)工具，使用基于注解的方法，可以為JVM程序增加

可靠的microbenchmarks。如果使用maven，可以這樣引入：

<dependency> <groupId>org.openjdk.jmh</groupId> <artifactId>jmh-core</artifactId> <version>1.21</version> </dependency> <dependency> <groupId>org.openjdk.jmh</groupId> <artifactId>jmh-generator-annprocess</artifactId> <version>1.21</version> </dependency>

第一個(gè)庫(kù)是核心實(shí)現(xiàn)，第二個(gè)包含一個(gè)注解處理器，幫助生成JAR文件，通過(guò)它可以方便地運(yùn)行你的benchmark。maven配置里還應(yīng)該有下面的plugin：

<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <finalName>benchmarks</finalName> <transformers><transformer implementation='org.apache.maven.plugins.shade.resource.ManifestResourceTransformer'> <mainClass>org.openjdk.jmh.Main</mainClass></transformer> </transformers> </configuration> </execution></executions> </plugin>

程序代碼如下

import org.openjdk.jmh.annotations.Benchmark;import org.openjdk.jmh.annotations.BenchmarkMode;import org.openjdk.jmh.annotations.Fork;import org.openjdk.jmh.annotations.Level;import org.openjdk.jmh.annotations.Mode;import org.openjdk.jmh.annotations.OutputTimeUnit;import org.openjdk.jmh.annotations.Scope;import org.openjdk.jmh.annotations.State;import org.openjdk.jmh.annotations.TearDown;import java.util.concurrent.TimeUnit;import java.util.stream.Stream;//測(cè)量平均時(shí)間@BenchmarkMode(Mode.AverageTime)//以毫秒為單位，打印benchmark結(jié)果@OutputTimeUnit(TimeUnit.MILLISECONDS)//執(zhí)行兩次，增加可靠性。堆空間是4Gb@Fork(value = 2, jvmArgs = {'-Xms4G', '-Xmx4G'})@State(Scope.Benchmark)public class ParallelStreamBenchmark { private static final long N = 10_000_000L; @Benchmark public long sequentialSum() { return Stream.iterate(1L, i -> i + 1).limit(N).reduce(0L, Long::sum); } //每次執(zhí)行benchmark后，執(zhí)行GC @TearDown(Level.Invocation) public void tearDown() { System.gc(); }}

使用大內(nèi)存，和每次迭代以后試著GC都是為了盡量減少GC的影響。盡管如此，結(jié)果應(yīng)該再加一些鹽。很多因素會(huì)影響執(zhí)行時(shí)間，比如你的機(jī)器有多少核。

默認(rèn)地，JMH一般先執(zhí)行5次熱身迭代，這樣可以讓HotSpot優(yōu)化代碼，然后再執(zhí)行5次迭代用來(lái)計(jì)算最終的結(jié)果。你可以使用-w和-i命令行參數(shù)修改這些配置。

在我的機(jī)器上，使用JDK 1.8.0_121, Java HotSpot™ 64-Bit Server VM，執(zhí)行結(jié)果是

Benchmark Mode Cnt Score Error Units

ParallelStreamBenchmark.sequentialSum avgt 10 83.565 ± 1.841 ms/op

你應(yīng)該期望，使用經(jīng)典的for循環(huán)的迭代版本運(yùn)行得更快，因?yàn)樗诟蛯樱╨evel）工作，而且，更重要的是，它不需要執(zhí)行原始類(lèi)型的裝箱和拆箱操作。我們測(cè)試一下這個(gè)方法：

@Benchmark public long iterativeSum() { long result = 0; for (long i = 1L; i <= N; i++) { result += i; } return result; }

執(zhí)行結(jié)果是

Benchmark Mode Cnt Score Error Units

ParallelStreamBenchmark.iterativeSum avgt 10 6.877 ± 0.068 ms/op

證實(shí)了我們的期望：迭代版本比串行流快了10倍。讓我們使用并行流試一試：

Benchmark Mode Cnt Score Error Units

ParallelStreamBenchmark.parallelSum avgt 10 110.157 ± 1.882 ms/op

非常令人失望：并行版本的求和一點(diǎn)都沒(méi)有發(fā)揮多核的優(yōu)勢(shì)，比串行版還要慢。為什么會(huì)這樣？有兩個(gè)問(wèn)題混在一起：

迭代生成了裝箱對(duì)象，它們?cè)谧黾臃ㄇ埃仨毑鹣涑蓴?shù)字

迭代很難劃分獨(dú)立的塊來(lái)并行地執(zhí)行

第二點(diǎn)是特別有趣的，不是所有的流都是適合并行處理的。特別是，迭代的流就很難，這是因?yàn)椋瘮?shù)的輸入依賴(lài)上一個(gè)函數(shù)的結(jié)果。見(jiàn)下圖：

這意味著，reduction過(guò)程并沒(méi)有像第一張圖里所表示的那樣執(zhí)行。reduction開(kāi)始的時(shí)候，還沒(méi)有整個(gè)數(shù)字列表，所以沒(méi)法分塊。把流標(biāo)記為并行的，反而增加了在不同線程上執(zhí)行的求和要被串行處理的負(fù)擔(dān)。

使用更專(zhuān)業(yè)的方法

LongStream.rangeClosed方法使用的是原始long類(lèi)型，所以不用裝箱和拆箱。而且，它生產(chǎn)的數(shù)的范圍，可以很容易地分成不依賴(lài)的塊。比如，范圍1-20可以被分成1-5、6-10、11-15和16-20。

@Benchmark public long rangedSum() { return LongStream.rangeClosed(1, N).reduce(0L, Long::sum); }

輸出是

Benchmark Mode Cnt Score Error Units

ParallelStreamBenchmark.rangedSum avgt 10 7.660 ± 1.643 ms/op

可以看出來(lái)，比并行流快了很多，僅比經(jīng)典的for循環(huán)慢了一點(diǎn)。LongStream支持并行：

@Benchmark public long parallelRangedSum() { return LongStream.rangeClosed(1, N).parallel().reduce(0L, Long::sum); }

輸出是

Benchmark Mode Cnt Score Error Units

ParallelStreamBenchmark.parallelRangedSum avgt 10 4.790 ± 5.142 ms/op

可以發(fā)現(xiàn)，并行生效了。甚至比f(wàn)or循環(huán)還快了1/3。

正確使用并行流

濫用并行流產(chǎn)生錯(cuò)誤的主要原因是使用了改變共享狀態(tài)的算法。下面是一個(gè)通過(guò)改變共享的累加器來(lái)實(shí)現(xiàn)前n個(gè)自然數(shù)求和的例子：

public long sideEffectSum(long n) { Accumulator accumulator = new Accumulator(); LongStream.rangeClosed(1, n).forEach(accumulator::add); return accumulator.total; } public class Accumulator { public long total = 0; public void add(long value) { total += value; } }

這種代碼很常見(jiàn)，特別對(duì)熟悉命令式編程范式的開(kāi)發(fā)者而言。當(dāng)你迭代數(shù)字列表時(shí)，經(jīng)常這樣做：初始化一個(gè)累加器，遍歷元素，使用累加器相加。

這代碼有什么錯(cuò)？它是串行的，失去了并行性。讓我們?cè)囍褂貌⑿辛鳎?/p>

public long sideEffectParallelSum(long n) { Accumulator accumulator = new Accumulator(); LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add); return accumulator.total; }

多執(zhí)行幾次，你會(huì)發(fā)現(xiàn)，每次返回的結(jié)果都不一樣，而且都不是正確的50000005000000。這是因?yàn)槎嗑€程累加的時(shí)候，total += value并不是原子操作。那么怎樣才能寫(xiě)出并行情況下，正確的代碼呢？

如果有懷疑，就做測(cè)試

注意裝箱問(wèn)題。Java提供的原始類(lèi)型流（IntStream、LongStream和DoubleStream）可以避免類(lèi)似的問(wèn)題，盡量使用他們

有些操作使用并行流性能更差。尤其是像limit和findFirst這種依賴(lài)元素順序的操作，使用并行是非常昂貴的。比如，findAny就比f(wàn)indFirst性能好，因?yàn)樗樞驘o(wú)關(guān)。調(diào)用unordered方法，可以把一個(gè)有順序的流變成無(wú)順序的流。比如，如果你需要流的N個(gè)元素，而你對(duì)前M個(gè)感興趣，在一個(gè)無(wú)順序的流上調(diào)用limit比有順序的高效

如果數(shù)據(jù)量不大，不要選擇并行流

要考慮流的底層數(shù)據(jù)結(jié)構(gòu)的可分解程度。比如，ArrayList比LinkedList分解起來(lái)更高效，因?yàn)椴槐闅v就可以分割。使用range工廠增加的原始類(lèi)型流也很容易分割。可以通過(guò)實(shí)現(xiàn)自己的Spliterator分割流

流的特征，以及中間操作如何修改流的元素，會(huì)改變分解過(guò)程的性能。比如，一個(gè)SIZED流可以被分解成兩個(gè)相等的部分，并且每個(gè)部分可以高效得并行處理，但是，filter會(huì)過(guò)濾掉任何不滿足條件的元素，導(dǎo)致流的size成了未知的

考慮結(jié)束操作是廉價(jià)的還是昂貴的merge步驟（比如，Collector的combiner方法）。如果是昂貴的，組合并行結(jié)果的代價(jià)會(huì)比并行流帶來(lái)的好處還要高

下面的表格，總結(jié)一些流在可分解性方面的并行友好性

源 可分解性 ArrayList 優(yōu)秀 LinkedList 差 IntStream.range 優(yōu)秀 Stream.iterate 差 HashSet 好 TreeSet 好

fork/join框架

fork/join框架用來(lái)遞歸地把可并行的任務(wù)分解成小任務(wù)，然后組合每個(gè)子任務(wù)的結(jié)果，以生成總的結(jié)果。它實(shí)現(xiàn)了ExecutorService接口，這樣所有的子任務(wù)都在一個(gè)線程池（ForkJoinPool）內(nèi)工作。

RecursiveTask

要向ForkJoinPool提交任務(wù)，你不得不增加RecursiveTask的子類(lèi)-R是并行任務(wù)（以及每個(gè)子任務(wù)）的返回類(lèi)型，或者

增加RecursiveAction的子類(lèi)-當(dāng)沒(méi)有返回值的時(shí)候。要定義RecursiveTask，需要實(shí)現(xiàn)它唯一的抽象方法：

protected abstract R compute();

該方法定義分割任務(wù)和不能繼續(xù)被分割時(shí)處理一個(gè)子任務(wù)的算法的邏輯。該方法的實(shí)現(xiàn)，經(jīng)常像下面的偽代碼：

if (任務(wù)足夠小，不再被分) { 順序執(zhí)行任務(wù)} else { 把任務(wù)分成兩個(gè)子任務(wù) 遞歸地調(diào)用本方法，盡量分割每個(gè)子任務(wù) 等待所有子任務(wù)的完成 組合每個(gè)子任務(wù)的結(jié)果}

可以發(fā)現(xiàn)，這是分治算法的并行實(shí)現(xiàn)。我們繼續(xù)求和的例子，演示怎么使用fork/join框架。首先需要擴(kuò)展RecursiveTask類(lèi)：

import java.util.concurrent.RecursiveTask;/** * Created by leishu on 18-12-11. */public class ForkJoinSumCalculator extends RecursiveTask<Long> { //分割任務(wù)的閾值 public static final long THRESHOLD = 10_000; //要被求和的數(shù)組 private final long[] numbers; private final int start; private final int end; public ForkJoinSumCalculator(long[] numbers) { this(numbers, 0, numbers.length); } //生成子任務(wù)的私有構(gòu)造器 private ForkJoinSumCalculator(long[] numbers, int start, int end) { this.numbers = numbers; this.start = start; this.end = end; } @Override protected Long compute() { //子任務(wù)的大小 int length = end - start; if (length <= THRESHOLD) { return computeSequentially();//小于閾值，不分割 } //增加第一個(gè)子任務(wù) ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length / 2); //異步執(zhí)行，新的子任務(wù)使用ForkJoinPool的另一個(gè)線程 leftTask.fork(); ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length / 2, end); //同步執(zhí)行第二個(gè)子任務(wù)，允許遞歸 Long rightResult = rightTask.compute(); //讀取第一個(gè)子任務(wù)的結(jié)果，如果沒(méi)完成就等待 Long leftResult = leftTask.join(); //組合 return leftResult + rightResult; } //順序執(zhí)行 private long computeSequentially() { long sum = 0; for (int i = start; i < end; i++) { sum += numbers[i]; } return sum; }}

使用fork/join的最佳實(shí)踐

調(diào)用任務(wù)的join方法，會(huì)阻塞調(diào)用者，直到返回結(jié)果。所以，要在兩個(gè)子任務(wù)都啟動(dòng)以后在調(diào)用它

不要在RecursiveTask內(nèi)使用ForkJoinPool的invoke方法

子任務(wù)的fork方法是用來(lái)做調(diào)度的。在兩個(gè)子任務(wù)上直接調(diào)用它似乎是很自然的，但是，在其中一個(gè)上調(diào)用compute效率更高，因?yàn)檫@樣能重用相同的線程

偷工作

任務(wù)被分給ForkJoinPool里的線程。每個(gè)線程有一個(gè)保存任務(wù)的雙端鏈表，順序地執(zhí)行鏈表中的任務(wù)。如果由于某種原因（比如I/O），一個(gè)線程完成了分配給他的全部任務(wù)，它會(huì)隨機(jī)地從其他線程選擇一個(gè)隊(duì)列，從隊(duì)列的尾部偷一個(gè)任務(wù)。這個(gè)過(guò)程會(huì)持續(xù)，直到所有的隊(duì)列都空了為止。所以，要有大量的小任務(wù)，而不是幾個(gè)大任務(wù)，這樣可以更好地平衡線程的負(fù)荷。

Spliterator

Spliterator是Java 8 提供的新接口，意思是“splitable iterator”，用來(lái)并行地迭代源中的元素。也許你不用開(kāi)發(fā)自己的Spliterator，但是，理解了它，也就明白了并行流是如何工作的。Java 8已經(jīng)在Collections框架內(nèi)提供了Spliterator的默認(rèn)實(shí)現(xiàn)。Collection接口有一個(gè)default方法spliterator()，它就返回一個(gè)Spliterator對(duì)象。我們先看看Spliterator接口的定義：

public interface Spliterator<T> { //用來(lái)按順序消費(fèi)Spliterator的元素，如果還有元素就返回true boolean tryAdvance(Consumer<? super T> action); //把一些元素分到一個(gè)新的Spliterator，以允許他們并行處理 Spliterator<T> trySplit(); //剩余的可被遍歷的元素?cái)?shù)量估值 long estimateSize(); int characteristics();}

tryAdvance方法的行為類(lèi)似于迭代器，用來(lái)按順序消費(fèi)Spliterator的元素，如果還有元素就返回true。trySplit方法

用來(lái)把一些元素分到一個(gè)新的Spliterator，以允許他們并行處理。

分割過(guò)程

把一個(gè)流分割成多個(gè)部分是一個(gè)遞歸過(guò)程，如下圖所示。首先，在第一個(gè)Spliterator上調(diào)用trySplit生成一個(gè)新的。然后，在這兩個(gè)Spliterator上調(diào)用trySplit，這樣產(chǎn)生四個(gè)。一直進(jìn)行下去，直到該方法返回null，標(biāo)志著不能再被分割。最后，當(dāng)所有的trySplit都返回null時(shí)，遞歸過(guò)程結(jié)束。

分割過(guò)程也會(huì)受到Spliterator的特征（由characteristics方法聲明）的影響。

Spliterator特征

characteristics方法返回一個(gè)整數(shù)，用來(lái)更好地控制和優(yōu)化Spliterator的用法。

Characteristic 描述 ORDERED 元素是有順序的（比如List），所以Spliterator使用該順序做遍歷和分區(qū) DISTINCT 對(duì)于每對(duì)遍歷的元素x和y，x.equals(y)返回false SORTED 遍歷的元素遵循預(yù)定義的排序順序 SIZED 源的size是已知的（比如set），所以estimatedSize()返回的值是精確的 NON-NULL 元素不會(huì)為空 IMMUTABLE 源是不可變的，說(shuō)明遍歷的時(shí)候，元素不會(huì)被增加、修改和刪除 CONCURRENT 源是并發(fā)安全的，并發(fā)修改的時(shí)候，不用任何同步 SUBSIZED Spliterator和接下來(lái)產(chǎn)生的Spliterator都是SIZED

實(shí)現(xiàn)自己的Spliterator

我們開(kāi)發(fā)一個(gè)簡(jiǎn)單的方法，用來(lái)計(jì)算字符串中的單詞數(shù)。

public int countWordsIteratively(String s) { int counter = 0; boolean lastSpace = true; for (char c : s.toCharArray()) { if (Character.isWhitespace(c)) {lastSpace = true; } else {if (lastSpace) counter++;lastSpace = false; } } return counter; }

要計(jì)算的字符串是但丁的“地域”的第一句

public static final String SENTENCE =' Nel mezzo del cammin di nostra vita ' + 'mi ritrovai in una selva oscura' + ' che la dritta via era smarrita '; System.out.println('Found ' + countWordsIteratively(SENTENCE) + ' words');

注意，兩個(gè)單詞間的空格數(shù)是隨機(jī)的。執(zhí)行結(jié)果

Found 19 words

使用函數(shù)式實(shí)現(xiàn)

首先需要把字符串轉(zhuǎn)換成一個(gè)流。原始類(lèi)型int、long和double才有原始的的流，所以，我們使用Stream：

Stream<Character> stream = IntStream.range(0, SENTENCE.length())

.mapToObj(SENTENCE::charAt);

可以使用reduction計(jì)算單詞數(shù)量。當(dāng)reduce的時(shí)候，你不得不攜帶由兩個(gè)變量組成的狀態(tài)：整數(shù)型的總數(shù)和布爾型的字符是否是空格。因?yàn)镴ava沒(méi)有tuples，你得增加一個(gè)新類(lèi)-WordCounter-封裝狀態(tài)：

class WordCounter { private final int counter; private final boolean lastSpace; public WordCounter(int counter, boolean lastSpace) { this.counter = counter; this.lastSpace = lastSpace; }//遍歷，累加 public WordCounter accumulate(Character c) { if (Character.isWhitespace(c)) {return lastSpace ? this : new WordCounter(counter, true); } else {//如果上一個(gè)字符是空格，而當(dāng)前的不是，就加1return lastSpace ? new WordCounter(counter + 1, false) : this; } } //組合，求和 public WordCounter combine(WordCounter wordCounter) { return new WordCounter(counter + wordCounter.counter, wordCounter.lastSpace); } public int getCounter() { return counter; } }

下面是遍歷一個(gè)新字符時(shí)，WordCounter的狀態(tài)圖

然后，我們就可以使用流的reduce方法了

private int countWords(Stream<Character> stream) { WordCounter wordCounter = stream.reduce(new WordCounter(0, true),WordCounter::accumulate,WordCounter::combine); return wordCounter.getCounter(); }

我們做一下測(cè)試

Stream<Character> stream = IntStream.range(0, SENTENCE.length()).mapToObj(SENTENCE::charAt); System.out.println('Found ' + countWords(stream) + ' words');

執(zhí)行結(jié)果是正確的。

并行的實(shí)現(xiàn)

我們修改一下代碼

System.out.println('Found ' + countWords(stream.parallel()) + ' words');

執(zhí)行結(jié)果不是找到19個(gè)單詞了。因?yàn)樵醋址陔S意的位置被分割，一個(gè)字符被多次分割。要解決這個(gè)問(wèn)題，就需要實(shí)現(xiàn)自己的Spliterator。

class WordCounterSpliterator implements Spliterator<Character> { private final String string; private int currentChar = 0; private WordCounterSpliterator(String string) { this.string = string; } @Override public boolean tryAdvance(Consumer<? super Character> action) { //消費(fèi)當(dāng)前字符 action.accept(string.charAt(currentChar++)); //如果還有字符可被消費(fèi)，返回true return currentChar < string.length(); } @Override public Spliterator<Character> trySplit() { int currentSize = string.length() - currentChar; //小于閾值，不再分割 if (currentSize < 10) {return null; } //候選的分割位置是字符串的一半長(zhǎng)度 for (int splitPos = currentSize / 2 + currentChar; splitPos < string.length(); splitPos++) {//如果是空格，才分割if (Character.isWhitespace(string.charAt(splitPos))) { Spliterator<Character> spliterator = new WordCounterSpliterator(string.substring(currentChar, splitPos)); //當(dāng)前位置修改為分割位置 currentChar = splitPos; return spliterator;} } return null; } @Override public long estimateSize() { return string.length() - currentChar; } @Override public int characteristics() { return ORDERED + SIZED + SUBSIZED + NONNULL + IMMUTABLE; } }

然后，我們做測(cè)試

Spliterator<Character> spliterator = new WordCounterSpliterator(SENTENCE); Stream<Character> stream = StreamSupport.stream(spliterator, true); System.out.println('Found ' + countWords(stream) + ' words');

這回沒(méi)問(wèn)題了。

以上這篇Java 并行數(shù)據(jù)處理和性能分析就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持好吧啦網(wǎng)。