Java 并行數(shù)據(jù)處理和性能分析

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

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

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

//測量平均時間
@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();
  }
}

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

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();//小于閾值，不分割
    }
    //增加第一個子任務(wù)
    ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length / 2);
    //異步執(zhí)行，新的子任務(wù)使用ForkJoinPool的另一個線程
    leftTask.fork();
    ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length / 2, end);
    //同步執(zhí)行第二個子任務(wù)，允許遞歸
    Long rightResult = rightTask.compute();
    //讀取第一個子任務(wù)的結(jié)果，如果沒完成就等待
    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;
  }
}

public interface Spliterator<T> {
  //用來按順序消費Spliterator的元素，如果還有元素就返回true
  boolean tryAdvance(Consumer<? super T> action);
  //把一些元素分到一個新的Spliterator，以允許他們并行處理
  Spliterator<T> trySplit();
  //剩余的可被遍歷的元素數(shù)量估值
  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 ? this : new WordCounter(counter, true);
      } else {
        //如果上一個字符是空格，而當(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;
    }
  }

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

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

  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) {
      //消費當(dāng)前字符
      action.accept(string.charAt(currentChar++));
      //如果還有字符可被消費，返回true
      return currentChar < string.length();
    }

    @Override
    public Spliterator<Character> trySplit() {
      int currentSize = string.length() - currentChar;
      //小于閾值，不再分割
      if (currentSize < 10) {
        return null;
      }
      //候選的分割位置是字符串的一半長度
      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;
    }
  }

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

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