並列ストリーム

1 行足しただけで、速くなるはずがない

Stream には parallel() という中間操作があります。これを挟むと、要素が分割されて複数の CPU コアで同時に処理されます。

long hits = logs.stream()
    .parallel()
    .filter(line -> line.contains("ERROR"))
    .count();

追加したのは 1 行だけです。それで速くなるなら、全部の Stream に付ければよさそうに思えます。ところが実際に時間を測ると、多くの場合は変わらないか、むしろ遅くなります。

理由は、並列化そのものにも手間がかかるからです。要素を分けて、それぞれのコアに配って、終わったら結果を合わせる。この段取りに時間を使います。1 件あたりの処理が軽ければ、段取りの時間の方が大きくなります。数十件から数百件のリストを流す程度では、まず割に合いません。

parallel() は「速くする道具」ではなく「速くなるかもしれない試し方」です。付けるかどうかは、両方の書き方で実際に時間を測ってから決めます。

順番が保証されなくなる

速さ以前に、動きが変わる点があります。forEach の処理される順番です。

List.of("a", "b", "c", "d").parallelStream()
    .forEach(s -> System.out.println(s));

これを何度か実行すると、出てくる順番が毎回変わります。分割された塊が、終わった順に処理されるからです。並んだ順に出したいなら forEachOrdered を使いますが、そうすると順番を待つ分だけ並列にした意味が薄れます。

collect(Collectors.toList()) のように結果を集める終端操作なら、元の順番は保たれます。順番が崩れて困るのは、処理の途中で表示や書き込みをしている場合です。

共有している変数を書き換えると壊れる

もう 1 つ、こちらの方が深刻です。

List<String> found = new ArrayList<>();
logs.parallelStream()
    .filter(line -> line.contains("ERROR"))
    .forEach(line -> found.add(line)); // 危ない

ArrayList は、複数のスレッドから同時に追加されることを想定していません。同時に add が走ると、要素が入らずに消えたり、例外が出たりします。しかもいつも壊れるわけではなく、たまに壊れます。テストでは通って本番で落ちる、いちばん厄介な種類のバグです。

結果を集めたいときは、自分で用意したリストに詰めるのではなく collect を使います。collect は並列でも安全に集まるように作られています。

解説

直列で書いてあるコードに parallel() を足すだけで壊れる、ということは、元のコードがラムダの外側を書き換えていたということです。並列化を試すのは、そういう書き方をしていないか点検する良い機会にもなります。

やってみよう

Solution.parallelSum(int[] arr) を完成させて、配列の合計を返してください。

ファイルの先頭に import java.util.Arrays; を書き、Arrays.stream(arr) で IntStream を作ります。そこに .parallel() を挟んでから、.sum() で合計を取ります。

今回の目的は、並列にしても答えが変わらないことを確かめることです。テストの配列は数件しか無いので、速くはなりません。むしろ直列より時間がかかります。合計のように「どこで区切って足しても答えが同じ」になる計算だから、分けて足して後から合わせても結果が一致する、というところを見てください。

1. ファイル先頭で import java.util.Arrays; を書くこと
2. Arrays.stream(arr) で IntStream を作り、.parallel() で並列モードに切り替えること
3. 最後に .sum() を呼んで合計の int を return すること