并行流
并行流是一個(gè)把元素分成多個(gè)塊的流，每個(gè)塊用不同的線程處理。可以自動(dòng)分區(qū)，讓所有的處理器都忙起來(lái)。
假設(shè)要寫(xiě)一個(gè)方法，接受一個(gè)數(shù)量n做參數(shù)，計(jì)算1-n的和。可以這樣實(shí)現(xiàn)：
|
1
2
3
4
5
|
public long sequentialSum(long n) { return Stream.iterate(1L, i -> i + 1) .limit(n) .reduce(0L, Long::sum);} |
也許可以使用parallel方法，簡(jiǎn)單地使用并行計(jì)算，提高程序性能：
|
1
2
3
4
5
6
|
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操作可以在不同的塊上互不依賴地并行地各自工作。最后，reduction操作組合每個(gè)子流的并行reductions的返回值，返回的結(jié)果就是整個(gè)流的結(jié)果。見(jiàn)下面的示意圖
實(shí)際上，調(diào)用parallel方法，流自身不會(huì)有任何變化。在內(nèi)部，設(shè)置一個(gè)布爾類型的標(biāo)記，標(biāo)明你想在并行模式執(zhí)行操作，接下來(lái)的操作都是并行的。
類似地，你也可以使用sequential方法，把并行流轉(zhuǎn)成串行的。你也許認(rèn)為可以組合這兩個(gè)方法：
|
1
2
3
4
5
6
|
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è)量流的性能
我們聲稱并行加法應(yīng)該比串行的或者自己的迭代方法快。我們可以使用JMH測(cè)量一下。這是一個(gè)工具，使用基于注解的方法，可以為JVM程序增加
可靠的microbenchmarks。如果使用maven，可以這樣引入：
|
1
2
3
4
5
6
7
8
9
10
|
<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：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
<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> |
程序代碼如下
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
|
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í)行原始類型的裝箱和拆箱操作。我們測(cè)試一下這個(gè)方法：
|
1
2
3
4
5
6
7
8
|
@Benchmarkpublic 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ù)的輸入依賴上一個(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)。
使用更專業(yè)的方法
LongStream.rangeClosed方法使用的是原始long類型，所以不用裝箱和拆箱。而且，它生產(chǎn)的數(shù)的范圍，可以很容易地分成不依賴的塊。比如，范圍1-20可以被分成1-5、6-10、11-15和16-20。
|
1
2
3
4
5
|
@Benchmarkpublic 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支持并行：
|
1
2
3
4
5
6
|
@Benchmarkpublic 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ù)求和的例子：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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>
|
1
2
3
4
5
|
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提供的原始類型流（IntStream、LongStream和DoubleStream）可以避免類似的問(wèn)題，盡量使用他們
有些操作使用并行流性能更差。尤其是像limit和findFirst這種依賴元素順序的操作，使用并行是非常昂貴的。比如，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工廠增加的原始類型流也很容易分割。可以通過(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的子類-R是并行任務(wù)（以及每個(gè)子任務(wù)）的返回類型，或者
增加RecursiveAction的子類-當(dāng)沒(méi)有返回值的時(shí)候。要定義RecursiveTask，需要實(shí)現(xiàn)它唯一的抽象方法：
protected abstract R compute();
該方法定義分割任務(wù)和不能繼續(xù)被分割時(shí)處理一個(gè)子任務(wù)的算法的邏輯。該方法的實(shí)現(xiàn)，經(jīng)常像下面的偽代碼：
|
1
2
3
4
5
6
7
8
|
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類：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
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接口的定義：
|
1
2
3
4
5
6
7
8
9
|
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)按順序消費(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ù)。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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ì)算的字符串是但丁的“地域”的第一句
|
1
2
3
4
5
6
|
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è)流。原始類型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è)新類-WordCounter-封裝狀態(tài)：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
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)前的不是，就加1 return 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方法了
|
1
2
3
4
5
6
|
private int countWords(Stream<Character> stream) { WordCounter wordCounter = stream.reduce(new WordCounter(0, true), WordCounter::accumulate, WordCounter::combine); return wordCounter.getCounter();} |
我們做一下測(cè)試
|
1
2
3
|
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。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
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è)試
|
1
2
3
4
|
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ù)器之家。
原文鏈接：https://blog.csdn.net/weixin_43364172/article/details/84955399
