Java并行数据处理和性能分析

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

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

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

    
      org.openjdk.jmh
      jmh-core
      1.21
    
    
      org.openjdk.jmh
      jmh-generator-annprocess
      1.21
    

      
        org.apache.maven.plugins
        maven-shade-plugin
        
          
            package
            
              shade
            
            
              benchmarks
              
                
                  org.openjdk.jmh.Main
                
              
            
          
        
      

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;

//测量平均时间
@BenchmarkMode(Mode.AverageTime)
//以毫秒为单位，打印benchmark结果
@OutputTimeUnit(TimeUnit.MILLISECONDS)
//执行两次，增加可靠性。堆空间是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);
  }
  
  //每次执行benchmark后，执行GC
  @TearDown(Level.Invocation)
  public void tearDown() {
    System.gc();
  }
}

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

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

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

  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; 
    }
  }

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

if (任务足够小，不再被分) {
  顺序执行任务
} else {
  把任务分成两个子任务
  递归地调用本方法，尽量分割每个子任务
  等待所有子任务的完成
  组合每个子任务的结果
}

import java.util.concurrent.RecursiveTask;

/**
 * Created by leishu on 18-12-11.
 */
public class ForkJoinSumCalculator extends RecursiveTask {
  //分割任务的阈值
  public static final long THRESHOLD = 10_000;
  //要被求和的数组
  private final long[] numbers;
  private final int start;
  private final int end;

  public ForkJoinSumCalculator(long[] numbers) {
    this(numbers, 0, numbers.length);
  }
  //生成子任务的私有构造器
  private ForkJoinSumCalculator(long[] numbers, int start, int end) {
    this.numbers = numbers;
    this.start = start;
    this.end = end;
  }

  @Override
  protected Long compute() {
    //子任务的大小
    int length = end - start;
    if (length <= THRESHOLD) {
      return computeSequentially();//小于阈值，不分割
    }
    //增加第一个子任务
    ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length / 2);
    //异步执行，新的子任务使用ForkJoinPool的另一个线程
    leftTask.fork();
    ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length / 2, end);
    //同步执行第二个子任务，允许递归
    Long rightResult = rightTask.compute();
    //读取第一个子任务的结果，如果没完成就等待
    Long leftResult = leftTask.join();
    //组合
    return leftResult + rightResult;
  }

  //顺序执行
  private long computeSequentially() {
    long sum = 0;
    for (int i = start; i 

public interface Spliterator {
  //用来按顺序消费Spliterator的元素，如果还有元素就返回true
  boolean tryAdvance(Consumer<&＃63; super T> action);
  //把一些元素分到一个新的Spliterator，以允许他们并行处理
  Spliterator trySplit();
  //剩余的可被遍历的元素数量估值
  long estimateSize();
  int characteristics();
}

  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;
  }

    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");

  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 &＃63; this : new WordCounter(counter, true);
      } else {
        //如果上一个字符是空格，而当前的不是，就加1
        return lastSpace &＃63; new WordCounter(counter + 1, false) : this;
      }
    }

    //组合，求和
    public WordCounter combine(WordCounter wordCounter) {
      return new WordCounter(counter + wordCounter.counter, wordCounter.lastSpace);
    }

    public int getCounter() {
      return counter;
    }
  }

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

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

  class WordCounterSpliterator implements Spliterator {

    private final String string;
    private int currentChar = 0;

    private WordCounterSpliterator(String string) {
      this.string = string;
    }

    @Override
    public boolean tryAdvance(Consumer<&＃63; super Character> action) {
      //消费当前字符
      action.accept(string.charAt(currentChar++));
      //如果还有字符可被消费，返回true
      return currentChar  trySplit() {
      int currentSize = string.length() - currentChar;
      //小于阈值，不再分割
      if (currentSize <10) {
        return null;
      }
      //候选的分割位置是字符串的一半长度
      for (int splitPos = currentSize / 2 + currentChar; splitPos  spliterator = new WordCounterSpliterator(string.substring(currentChar, splitPos));
          //当前位置修改为分割位置
          currentChar = splitPos;
          return spliterator;
        }
      }
      return null;
    }

    @Override
    public long estimateSize() {
      return string.length() - currentChar;
    }

    @Override
    public int characteristics() {
      return ORDERED + SIZED + SUBSIZED + NONNULL + IMMUTABLE;
    }
  }

    Spliterator spliterator = new WordCounterSpliterator(SENTENCE);
    Stream stream = StreamSupport.stream(spliterator, true);

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