並列ストリーム｜コードで実践

並列ストリームとは

Stream API の parallel() で配列を複数 CPU コアに分割集計させる並列ストリームの仕組みと、結合性・スレッド安全・順序保証など並列化の落とし穴を学びます。

`parallel()` で Stream を一気にマルチコア化する

ここまでの Stream API の章で、Arrays.stream(arr).sum() のように データの流れ として配列を扱う書き方に慣れてきました。for ループを書かずに「何をしたいか」だけを並べていく宣言的なスタイルは、それだけで十分便利ですが、Stream にはもうひとつ大きな武器があります。それが、メソッドを 1 つ足すだけで処理を 並列化 できる、parallel() です。

Java

import java.util.Arrays;

public class Solution {
    public static int parallelSum(int[] arr) {
        return Arrays.stream(arr).parallel().sum();
    }
}

見た目はほとんど sum() の例と変わりません。違うのは真ん中の .parallel() だけです。それなのに、配列の要素が大量にあるときは、JVM が裏で配列を細かく分割し、複数の CPU コアに割り振って同時に合計し、最後に部分合計を足して int を返してくれます。手書きで スレッド を立てて、synchronized で守って、Future で結果を集めて……といった面倒事を、Stream API がぜんぶ巻き取ってくれる仕組みです。

並列化を「速くなる魔法」のように見てしまうと、後で必ずハマります。並列ストリームは、書き方の問題と性能の問題と正しさの問題が同時に絡む 高級な道具です。本レッスンでは、parallel() の文法だけでなく、どんな処理なら並列化していいのか、どんな処理だと壊れるのか、までを地続きで押さえます。

`parallel()` の文法と仲間たち

Stream を並列にする方法はいくつかありますが、覚えるべきなのは次の 3 つで十分です。

stream.parallel() — すでに作ってある Stream を並列モードに切り替える
collection.parallelStream() — List や Set から最初から並列 Stream を作る
Arrays.stream(arr).parallel() — 配列を IntStream に変えてから並列化する

今回の課題で使うのは 3 つ目です。具体的には次のように書きます。

Java

import java.util.Arrays;

public class Demo {
    public static void main(String[] args) {
        int[] arr = {1, 2, 3, 4, 5};
        int total = Arrays.stream(arr).parallel().sum();
        System.out.println(total); // 15
    }
}

Arrays.stream(int[]) は IntStream を返すので、Stream<Integer> ではなく IntStream の sum() average() max() min() count() がそのまま使えます。parallel() を挟む位置はパイプラインの途中ならどこでも構いません。Arrays.stream(arr).parallel().filter(x -> x > 0).sum() のような書き方も自然です。

並列モードに切り替えるかどうかは、isParallel() で確認できます。stream.isParallel() が true ならそのパイプラインは並列実行されます。sequential() を呼ぶとまた直列に戻せるので、必要な区間だけ並列にする、ということもできます。

並列化できるのは「結合的」な操作だけ

parallel() を付ければ何でも速くなる、と思いがちですが、そう単純ではありません。並列化が 正しく動くため には、reduce する操作が 結合的 (associative) であることが必要です。結合的というのは、(a + b) + c と a + (b + c) の結果が同じ、という性質のことです。

+ (足し算) — 結合的。(1 + 2) + 3 == 1 + (2 + 3) で並列 OK
* (掛け算) — 結合的。並列 OK
max / min — 結合的。並列 OK
論理 AND / 論理 OR — 結合的。並列 OK
- (引き算) — 結合的ではない。(10 - 3) - 2 != 10 - (3 - 2) なので並列で結果が変わる
文字列を + で連結 — 結合性自体はあるが、要素の順序に依存するので並列だと並びが変わる

今回の課題で扱う sum は結合的なので、配列をどこで分割して、どの順番で部分合計を足し合わせても、最終的な答えはぴったり同じになります。これが「並列化して安全」「順序に依存しない」と呼ばれる状態です。

diagram (will load when visible)

この図のように、並列ストリームは配列を再帰的に細かく分割 (fork) して、部分集計を取り、それらを合流 (join) させて 1 つの結果にまとめます。途中の合流順は毎回違っても、+ が結合的なので最終の 36 は揺らがない、という設計です。

`ForkJoinPool` 共通プールで動いている

並列ストリームの正体は、java.util.concurrent.ForkJoinPool の 共通プール (commonPool) というスレッドプールです。parallel() を呼ぶと、Stream のパイプラインはこのプールに乗って動き、Runtime.getRuntime().availableProcessors() - 1 個程度のワーカースレッドが、配列の分割タスクをチョコチョコ取り合いながら計算してくれます。

Java

int cores = Runtime.getRuntime().availableProcessors();
System.out.println("CPU コア数: " + cores);

大事なポイントは、共通プールは JVM 全体で 1 つしかない ことです。あなたのアプリ内で同時にいろいろな処理が並列ストリームを使い始めると、互いに同じプールを奪い合い、思ったほど速くならない、ということが起こります。

重いリクエストを捌くサーバーで、HTTP ハンドラの中で何でもかんでも parallelStream() を呼ぶのは、初心者向け本にはよく出てきますが、実務ではむしろアンチパターンです。共通プールが詰まると、ほかのリクエストまで巻き添えで遅くなります。並列ストリームは「データが多くて、独立した重い計算が並ぶ場面」専用の道具 だと考えてください。

並列化の落とし穴 ─ 順序・スレッド安全・副作用

並列ストリームを使うときに気をつけることは、大きく次の 3 つです。

順序を期待しない — forEach で並列に流すと、要素を出力する順番は実行のたびに変わります。順序を保ちたいなら forEachOrdered を使うか、Collectors.toList() のように順序を保つ集約を選びます
スレッドセーフでないコンテナに書き込まない — ArrayList に parallel().forEach(list::add) するのは典型的なバグです。複数スレッドから同時に add が呼ばれてデータが壊れます。並列で集めたいときは Collectors.toList() や Collectors.toConcurrentMap() を使います
副作用のある reducer を書かない — reduce や map の中で外側の変数を書き換えるのも危険です。int[] counter = {0}; arr.parallel().forEach(x -> counter[0]++); のようなコードは、複数スレッドが同時に ++ するため、ほぼ確実に正しい結果になりません

もう 1 つ、性能面の落とし穴として、データが少ないと並列のほうが遅い という現象もよく起きます。parallel() は内部で配列を分割し、ワーカースレッドを起こし、結果を合流させる、というオーバーヘッドがあるためです。要素数が 100 個前後の int 配列なら、素直に for ループで足したほうが圧倒的に速いです。

Java

// 危険: 並列の中で外側の変数を書き換えている
int[] counter = {0};
Arrays.stream(arr).parallel().forEach(x -> counter[0]++); // 値が壊れる

// 安全: 集約は reducer に任せる
long count = Arrays.stream(arr).parallel().count();       // 正しい値

「速くしたい」と思って parallel() を雑に付ける前に、まずは 単一スレッドの計測値 を取りましょう。System.nanoTime() で前後の時刻を取って差を見れば十分です。並列化は計測してから入れる、入れたあとも計測する、というのが鉄則です。

よくある間違い

並列ストリームでハマる初心者のパターンを、3 つだけ整理しておきます。これを知っているだけで、本番のバグをだいぶ減らせます。

parallel() を付ければ順序が保たれていると思い込む — forEach の出力順、reduce の集合体の組み立て順は 実行のたびに変わる と思って書く必要があります。順序が必要なら forEachOrdered か Collectors.toList() を選びましょう
非結合的 な reducer を渡してしまう — 引き算や、文字列を結合順序つきで作る処理は、並列だと結果が変わります。reduce("", String::concat) を並列で回すと、繋がる順番が毎回違うので、最終の文字列がランダムに並び替わったように見えます
小さな配列に parallel() を付けて逆に遅くなる — 配列の要素数が数百〜数千程度では、並列化のオーバーヘッドのほうが大きく、for のほうが速いことが普通です。並列はデータが多くて、要素 1 個の計算が重いとき だけが効きます

これらは「コードがコンパイルは通るのに、結果がおかしい / 速くならない」という症状で出てくるので、原因に気付きにくいタイプのバグです。parallel() を入れたコードでは、出力順 / 集約方法 / 計測値の 3 つを必ず確認する癖をつけてください。

Java の公式ドキュメントは並列ストリームの注意点を Stream の Javadoc にしっかり書いています。「結合的」「ステートレス」「非干渉」という 3 つのキーワードが何度も出てくるので、迷ったらそこに戻ると判断がブレません。

やってみよう

今回の課題は、配列の中身を並列に足し合わせて合計を返すメソッド Solution.parallelSum(int[] arr) を完成させることです。手順は次のとおりです。

ファイル先頭で import java.util.Arrays; を書く
メソッドの中で Arrays.stream(arr) を呼んで IntStream を作る
そこに .parallel() を繋いで並列モードに切り替える
最後に .sum() を呼んで int の合計を return する

書き終えたら、次のテストケースで動作を確認します。

parallelSum({1, 2, 3, 4, 5}) → 15
parallelSum({}) → 0 (空配列なら 0)
parallelSum({100}) → 100 (要素 1 個)
parallelSum({-5, 5, -10, 10}) → 0 (プラマイ相殺)

動いたら、.parallel() を抜いた Arrays.stream(arr).sum() 版でも同じ結果になることを確認してみてください。同じ仕様を直列でも並列でも書ける、というのが Stream API の楽しいところで、データが大きくなったときには parallel() をワンタッチで足すだけでマルチコアに乗せられる、という見通しが体に染み込みます。

余裕があれば、Arrays.stream(arr).parallel().filter(x -> x > 0).sum() のように filter を挟んで「正の数だけ並列で合計する」コードも試してみましょう。+ も filter も、両方とも結合的・ステートレスなので、安心して並列化できる典型例です。これで Stream API の基本パイプラインは一通り押さえたことになります。

よくある質問

Q. 中級の内容は実務でどれくらい使いますか？

A. Collection（List/Map/Set）、Stream、例外処理、ジェネリクスは毎日のように登場します。Date/Time API、Files、try-with-resources も実務で頻出するため、本コースの内容は実プロジェクトでそのまま役立ちます。

Q. Stream と for ループはどっちで書くべき？

A. 可読性で選んでください。filter + map + collect が綺麗にハマるなら Stream、副作用や複雑な分岐が多いなら for ループの方が読みやすいです。一律にどちらを使うべきという正解はなく、チームのコーディング規約に合わせるのが現実的です。

Q. 次のステップでは何を学ぶべきですか？

A. 中級の基礎が固まったら、Spring Boot で Web API を作る、JUnit でテストを書く、Maven/Gradle でビルドを管理する、といった実プロジェクトのスキルに進むと効果的です。OSS のコードを読む経験も大きく成長を促します。

次のレッスン

次は Stream API まとめクイズ で、Stream API まとめクイズを学びます。

事前確認 — 進む前に次の 3 つができることを確認しましょう。

parallel の要点を自分の言葉で説明できる
このレッスンの最小コード (または操作手順) を見ずに書ける
練習問題やクイズで間違えた箇所を読み直して理解した

理解度チェック (30 秒)

Q. parallel とは何か、1 文で説明してください。

この章のポイント

A. 本文の「このレッスンで分かること」または冒頭の説明文を見直し、自分の言葉で要約できれば OK。詰まったら本レッスンの最初の H2 セクションを読み返してみましょう。

関連レッスン

並列ストリーム

`parallel()` で Stream を一気にマルチコア化する

`parallel()` の文法と仲間たち

並列化できるのは「結合的」な操作だけ

`ForkJoinPool` 共通プールで動いている

並列化の落とし穴 ─ 順序・スレッド安全・副作用

よくある間違い

やってみよう

よくある質問

Q. 中級の内容は実務でどれくらい使いますか？

Q. Stream と for ループはどっちで書くべき？

Q. 次のステップでは何を学ぶべきですか？

次のレッスン

要件

入出力例

ヒント

メモ

並列ストリーム

parallel() で Stream を一気にマルチコア化する

parallel() の文法と仲間たち

並列化できるのは「結合的」な操作だけ

ForkJoinPool 共通プールで動いている

並列化の落とし穴 ─ 順序・スレッド安全・副作用

よくある間違い

やってみよう

よくある質問

Q. 中級の内容は実務でどれくらい使いますか？

Q. Stream と for ループはどっちで書くべき？

Q. 次のステップでは何を学ぶべきですか？

次のレッスン

要件

入出力例

ヒント

メモ

`parallel()` で Stream を一気にマルチコア化する

`parallel()` の文法と仲間たち

`ForkJoinPool` 共通プールで動いている