并行流：把一个内容分成多个数据块，并用不同的线程分别处理每个数据块的流。
先做一个简单的测试，测试传统for循环，与顺序流，并行流的速度。
/**
* 并行测试 最慢
* @param n
* @return
*/
public static long parallelSum(long n){
return Stream.iterate(1L, i -> i+1)
.limit(n)
.parallel()
.reduce(0L, Long::sum);
}
/**
* 顺序测试 比并行快
* @param n
* @return
*/
public static long sequentialSum(long n){
return Stream.iterate(1L, i -> i+1)
.limit(n)
.reduce(0L, Long::sum);
}
/**
* 传统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修改算法：
/**
* 比传统for 还快
* @param n
* @return
*/
public static long rangedSum(long n){
return LongStream.rangeClosed(1, n)
.reduce(0L, Long::sum);
}
System.out.println("并行测试："+measureSumPerf(Test7::parallelSum, 10000000));
System.out.println("顺序测试："+measureSumPerf(Test7::sequentialSum, 10000000));
System.out.println("传统for："+measureSumPerf(Test7::iteativeSum, 10000000));
System.out.println("LongStream："+measureSumPerf(Test7::rangedSum, 10000000));
并行测试：404
顺序测试：143
传统for：7
LongStream：4
理论上，并行流比顺序流要更快，事实上并不是这样的。传统for循环更接近底层，表现也不差。
几点改善并行流的方法：
1.顺序流转换成并行流并不一定快。
2.避免装箱，使用IntStream，LongStream，DoubleStream。
3.注意limit，findFirst依赖元素顺序的流，在顺序流上的性能本身就不错。
4.流的总成本。
5.数据量小的时候并行流并不一定有好的效果。
6.考虑分拆效率，ArrayList比LinkedList效率更高。range工产方法创建的原始流类型也可快速分解。
7.考虑处理流时筛选等丢弃元素等情况。
8.考虑合并步骤的代价再决定。
//流的数据源与可分解性对比
//ArrayList 优
//LinkedList 差
//IntStream.range 优
//Stream.iterate 差
//HashSet 好
//TreeSet 好
使用RecursiveTask分支框架
public class ForkJoinSumCalculator extends RecursiveTask<Long> {
private final long [] numbers;
private final int start;
private final int end;
//不再将任务分解为子任务的数组大小
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();//同步执行
Long leftResult=leftTask.join();//读取第一个线程的结果，未完成就等待
return leftResult+rightResult;//两个任务结果组合
}
private Long computeSequentially() {
long sum=0;
for (int i = start; i < end; i++) {
sum+=numbers[i];
}
return sum;
}
/**
* 测试方法
* @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);
}
}
计算一串字符串中字符的个数，不含空格
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));
}
}
//改成并行流测试，出现异常。
System.out.println(countWords(stream.parallel()));
Spliterator实现上面demo
public class WordCounterSpliterator implements Spliterator<Character> {
private final String str;
private int currentChar=0;
public WordCounterSpliterator(String str) {
this.str = str;
}
/**
* 把当前位置Character传递给Consumer
*/
@Override
public boolean tryAdvance(Consumer<? super Character> action) {
action.accept(str.charAt(currentChar++));//处理当前字符串
return currentChar <str.length();//true 表示还要要处理
}
@Override
public Spliterator<Character> trySplit() {
int currentSize=str.length()-currentChar;
if (currentSize<10) {
return null; // 解析数小于10时执行顺序处理
}
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;//将起始位置设为裁缝位置
return spliterator;
}
}
return null;
}
/**
* 总长度与当前位置的差
*/
@Override
public long estimateSize() {
return str.length()-currentChar;
}
/**
* ORDERED 顺序
* SIZED estimateSize返回值精确
* SUBSIZED trySplit创建的其他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));
}
}
好了，就到这里了。
网友评论