[JavaSE8 Goldへの道] その11 並列ストリーム

System.out.print("sequential:");
IntStream.rangeClosed(1, 10)
  .forEach(i -> System.out.print(i + " "));

System.out.print("parallel:");
IntStream.rangeClosed(1, 10)
  .parallel()
  .forEach(i -> System.out.print(i + " "));

IntStream.rangeClosed(1, 10)
  .parallel()
  .forEach(i -> System.out.println(Thread.currentThread().getName() + ":" + i));

試験範囲としてはこれくらいのようですが、もう少し考察してみます。

どういうときに使うべきか

ストリームが内部で持つフラグ

当然ながら、全てのストリームを並列にすればいいというのは間違いです。
CPUコアが1つの場合並列化しても意味ないですし、要素数が少ない・中間操作の処理負荷が低いと、並列化でスレッドを管理するオーバーヘッドの方が高くなってしまいます。

要素数がある程度多い、または中間操作が複雑で時間がかかるほど並列化の効果は上がると思いますが、実際のところ一度測ってみるのが確実です。

「この中間操作・終端操作は並列ストリームの方が良い」みたいなのがあればいいのですが、APIドキュメントに一部書いてあるものの網羅しきってはいません。

例えばfindFirstとfindAnyという終端操作があります。どちらも基本は中間操作filterと組み合わせて、条件を満たす最初の要素/いずれかの要素が見つかった時点で返します。
「いずれか」ってなんだよという話ですが、findFirstの方は要素が現れる順番を意識していて、仮に並列ストリームであっても出てきた順番の中で最初の要素を返します。
findAnyの方は順番を意識しないため、並列ストリームだと結果が不定になる代わりに、パフォーマンスが上がります。

long st = System.currentTimeMillis();
OptionalDouble o = LongStream.range(1, 10_000_000_000L)
  .parallel()
  .mapToDouble(Math::sqrt)
  .filter(l -> l > 30000)
  .findAny();
//  .findFirst();
System.out.println(System.currentTimeMillis() - st);
System.out.println(o);

実行結果(findAny 何度か実行)

long st = System.currentTimeMillis();
AtomicLong al = new AtomicLong(1);
OptionalDouble o = LongStream.generate(() -> al.getAndAdd(1))
  .parallel()
  .mapToDouble(Math::sqrt)
  .filter(l -> l > 30000)
  .findFirst();
System.out.println(o);
System.out.println(System.currentTimeMillis() - st);

OptionalDouble o = LongStream.iterate(1, l -> l + 1)
//あとは同じ

• DISTINCT
• SORTED
• ORDERED
• SIZED
• SHORT_CIRCUIT

ソースを見ても具体的にそれぞれが立っている時どうなるまでは書いてないのですが、これらを見て最適な処理方法を選択しているようです。

LongStream#rangeとLongStream#iterateを比べたとき、おそらく前者はサイズが決まっているからSIZEDが立っていそうです。

peekメソッドでスレッド名と値を出力してみると、LongStream#rangeの方はスレッドに割り当てられる値が突然大きくなったりしており、複数スレッドに割り当てる要素をサイズに応じて変えてるように見えます。

LongStream#iterateの方はサイズがわからない無限ストリームなので、出力を見ると1024個ずつ割り振ってるようでした。
そこでLongStream#iterateを使ったコードでfilterの条件をもっと小さくしてみると、findAnyで結果がばらつくようになりました。

終端操作の並列性

前回その10で紹介したCollectors#toMap、Collectors#groupingByには並列化バージョンがあり、それぞれCollectors#toConcurrentMapとCollectors#groupingByConcurrentです。
Mapの代わりにjava.util.concurrent.ConcurrentMapを生成します。

しかし、APIドキュメントを見ると「並列ストリームの場合Concurrentが付く方を使わなければならない」とは書いていません。

これは通常並列ストリームでは各スレッドごとにコンテナが用意され最後に集約されるのですが、Mapを集約するには他のコレクションよりコストが高くなる可能性があるため、Concurrent付きの方ではConcurrentMapを1つだけ生成して全スレッドが値を入れていくということのようです。
現にtoListやtoSetにはConcurrentバージョンがありません。

プログラマは、ストリームのソースの特性（ListなのかSetなのか、サイズが決まっているか、順序付けされているか）を考慮した上で、適切な終端操作を選んで並列ストリームを使う必要がありそうです。

というわけで

後半は試験から外れましたが、並列ストリーム奥が深いです。
いろいろ試してみて経験値を積む必要がありそうです。

• Exam 1Z0-810: Upgrade Java SE 7 to Java SE 8 OCP Programmer Study Guide