并行流：把一個(gè)內(nèi)容分成多個(gè)數(shù)據(jù)塊，并用不同的線程分別處理每個(gè)數(shù)據(jù)塊的流。
先做一個(gè)簡(jiǎn)單的測(cè)試，測(cè)試傳統(tǒng)for循環(huán)，與順序流，并行流的速度。
/**
* 并行測(cè)試 最慢
* @param n
* @return
*/
public static long parallelSum(long n){
return Stream.iterate(1L, i -> i+1)
.limit(n)
.parallel()
.reduce(0L, Long::sum);
}
/**
* 順序測(cè)試 比并行快
* @param n
* @return
*/
public static long sequentialSum(long n){
return Stream.iterate(1L, i -> i+1)
.limit(n)
.reduce(0L, Long::sum);
}
/**
* 傳統(tǒng)for 更底層 最快
* @param n
* @return
*/
public static long iteativeSum(long n){
long result=0;
for (int i = 0; i < n; i++) {
result+=i;
}
return result;
}
引入LongStream修改算法：
/**
* 比傳統(tǒng)for 還快
* @param n
* @return
*/
public static long rangedSum(long n){
return LongStream.rangeClosed(1, n)
.reduce(0L, Long::sum);
}
System.out.println("并行測(cè)試："+measureSumPerf(Test7::parallelSum, 10000000));
System.out.println("順序測(cè)試："+measureSumPerf(Test7::sequentialSum, 10000000));
System.out.println("傳統(tǒng)for："+measureSumPerf(Test7::iteativeSum, 10000000));
System.out.println("LongStream："+measureSumPerf(Test7::rangedSum, 10000000));
并行測(cè)試：404
順序測(cè)試：143
傳統(tǒng)for：7
LongStream：4
理論上，并行流比順序流要更快，事實(shí)上并不是這樣的。傳統(tǒng)for循環(huán)更接近底層，表現(xiàn)也不差。
幾點(diǎn)改善并行流的方法：
1.順序流轉(zhuǎn)換成并行流并不一定快。
2.避免裝箱，使用IntStream，LongStream，DoubleStream。
3.注意limit，findFirst依賴元素順序的流，在順序流上的性能本身就不錯(cuò)。
4.流的總成本。
5.數(shù)據(jù)量小的時(shí)候并行流并不一定有好的效果。
6.考慮分拆效率，ArrayList比LinkedList效率更高。range工產(chǎn)方法創(chuàng)建的原始流類型也可快速分解。
7.考慮處理流時(shí)篩選等丟棄元素等情況。
8.考慮合并步驟的代價(jià)再?zèng)Q定。
//流的數(shù)據(jù)源與可分解性對(duì)比
//ArrayList 優(yōu)
//LinkedList 差
//IntStream.range 優(yōu)
//Stream.iterate 差
//HashSet 好
//TreeSet 好
使用RecursiveTask分支框架
public class ForkJoinSumCalculator extends RecursiveTask<Long> {
private final long [] numbers;
private final int start;
private final int end;
//不再將任務(wù)分解為子任務(wù)的數(shù)組大小
public static final long THRESHOLD=10000;
public ForkJoinSumCalculator(long[] numbers, int start, int end) {
this.numbers = numbers;
this.start = start;
this.end = end;
}
public ForkJoinSumCalculator(long [] numbers) {
this(numbers,0,numbers.length);
}
@Override
protected Long compute() {
int length=end-start;
if (length<=THRESHOLD) {
return computeSequentially();
}
ForkJoinSumCalculator leftTask=new ForkJoinSumCalculator(numbers,start,start+length/2);
leftTask.fork();
ForkJoinSumCalculator rightTask=new ForkJoinSumCalculator(numbers,start,start+length/2);
Long rightResult=rightTask.compute();//同步執(zhí)行
Long leftResult=leftTask.join();//讀取第一個(gè)線程的結(jié)果，未完成就等待
return leftResult+rightResult;//兩個(gè)任務(wù)結(jié)果組合
}
private Long computeSequentially() {
long sum=0;
for (int i = start; i < end; i++) {
sum+=numbers[i];
}
return sum;
}
/**
* 測(cè)試方法
* @param n
* @return
*/
public static long forkJoinSum(long n){
long [] numbers=LongStream.rangeClosed(1, n).toArray();
ForkJoinTask<Long> task=new ForkJoinSumCalculator(numbers);
return new ForkJoinPool().invoke(task);
}
}
計(jì)算一串字符串中字符的個(gè)數(shù)，不含空格
public class WordCounter {
private static final String STR = "I am a Android engineer ! You can you up !";
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{
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;
}
public static int countWords(Stream<Character> stream){
WordCounter wordCounter=stream.reduce(new WordCounter(0, true),
WordCounter::accumulate,WordCounter::combine);
return wordCounter.getCounter();
}
public static void main(String[] args) {
Stream<Character> stream=IntStream.range(0, STR.length()).mapToObj(STR::charAt);
System.out.println(countWords(stream));
}
}
//改成并行流測(cè)試，出現(xiàn)異常。
System.out.println(countWords(stream.parallel()));
Spliterator實(shí)現(xiàn)上面demo
public class WordCounterSpliterator implements Spliterator<Character> {
private final String str;
private int currentChar=0;
public WordCounterSpliterator(String str) {
this.str = str;
}
/**
* 把當(dāng)前位置Character傳遞給Consumer
*/
@Override
public boolean tryAdvance(Consumer<? super Character> action) {
action.accept(str.charAt(currentChar++));//處理當(dāng)前字符串
return currentChar <str.length();//true 表示還要要處理
}
@Override
public Spliterator<Character> trySplit() {
int currentSize=str.length()-currentChar;
if (currentSize<10) {
return null; // 解析數(shù)小于10時(shí)執(zhí)行順序處理
}
for (int splitPos = currentSize/2+currentChar; splitPos < str.length(); splitPos++) {
if (Character.isWhitespace(str.charAt(splitPos))) {
Spliterator<Character> spliterator=new WordCounterSpliterator(str.substring(currentChar, splitPos));
currentChar=splitPos;//將起始位置設(shè)為裁縫位置
return spliterator;
}
}
return null;
}
/**
* 總長(zhǎng)度與當(dāng)前位置的差
*/
@Override
public long estimateSize() {
return str.length()-currentChar;
}
/**
* ORDERED 順序
* SIZED estimateSize返回值精確
* SUBSIZED trySplit創(chuàng)建的其他Spliterator 大小確切
* NONNULL 不為null
* IMMUTABLE 不可變（String本身不可變）
*/
@Override
public int characteristics() {
return ORDERED+SIZED+SUBSIZED+NONNULL+IMMUTABLE;
}
public static int countWords(Stream<Character> stream){
WordCounter wordCounter=stream.reduce(new WordCounter(0, true),
WordCounter::accumulate,WordCounter::combine);
return wordCounter.getCounter();
}
public static void main(String[] args) {
String str="Characteristic value signifying that an encounter order is defined for elements.";
Spliterator<Character> spliterator=new WordCounterSpliterator(str);
Stream<Character> stream=StreamSupport.stream(spliterator, true);
System.out.println(countWords(stream));
}
}
好了，就到這里了。
