Stream 並列化の落とし穴（副作用） — スレッド安全性
Java の Stream API は .parallelStream() や .parallel() を使うと簡単に並列化できます。
しかし「副作用を伴う処理」を並列ストリームで行うと、スレッド安全性の問題や予期せぬ結果が発生します。初心者がつまずきやすいポイントを、例題とテンプレートで整理します。
並列ストリームの基本
- 直列ストリーム:
list.stream()→ 順番に処理。 - 並列ストリーム:
list.parallelStream()→ ForkJoinPool を使って複数スレッドで処理。
メリット
- CPU コアを活用して計算処理を高速化できる。
- 集約処理（sum, average, groupingBy など）で効果的。
デメリット（落とし穴）
- 副作用がある処理（外部変数の更新、リストへの追加など）はスレッド安全でない。
- 順序保証が崩れる（forEach は順序が保証されない）。
- 軽い処理では逆に遅くなる（並列化のオーバーヘッドが大きい）。
副作用の危険な例
例1: 外部リストへの追加
List<Integer> nums = IntStream.rangeClosed(1, 1000).boxed().toList();
List<Integer> evens = new ArrayList<>();
// 危険: 並列ストリームで外部リストに追加
nums.parallelStream()
.filter(n -> n % 2 == 0)
.forEach(evens::add);
System.out.println(evens.size()); // 期待通りにならないことがある
- 問題:
ArrayListはスレッドセーフでないため、並列でaddするとデータ欠落や例外が起きる。
例2: 外部変数の更新
List<Integer> nums = IntStream.rangeClosed(1, 1000).boxed().toList();
int[] sum = {0};
nums.parallelStream().forEach(n -> sum[0] += n); // 危険
System.out.println(sum[0]); // 正しい合計にならないことがある
- 問題: 複数スレッドが同時に
sum[0]を更新し、競合が発生。
安全な書き方（副作用を避ける）
例題1: 合計は reduce/sum を使う
int total = IntStream.rangeClosed(1, 1000)
.parallel()
.sum();
System.out.println(total); // 500500
- ポイント:
sumは内部的にスレッド安全に集約される。
例題2: リスト収集は collect を使う
List<Integer> evens = IntStream.rangeClosed(1, 20)
.parallel()
.filter(n -> n % 2 == 0)
.boxed()
.collect(Collectors.toList());
System.out.println(evens); // 偶数リスト
- ポイント:
collectは並列対応済み。外部リストに直接追加しない。
例題3: 順序を守りたいなら forEachOrdered
IntStream.rangeClosed(1, 10)
.parallel()
.forEachOrdered(System.out::println);
- ポイント: 並列でも順序を保証して出力。
テンプレート集
- 並列合計
int sum = list.parallelStream().mapToInt(x -> x).sum();
- 並列収集（安全）
List<T> result = list.parallelStream()
.filter(cond)
.collect(Collectors.toList());
- 順序保証付き出力
stream.parallel().forEachOrdered(System.out::println);
- reduce を使った安全な集約
int product = IntStream.rangeClosed(1, 10)
.parallel()
.reduce(1, (a,b) -> a*b);
落とし穴と回避策
- 外部状態の変更禁止: 並列ストリームでは外部変数やリストへの直接書き込みを避ける。
- 順序が必要なら forEachOrdered: forEach は順序保証なし。
- 軽い処理は直列の方が速い: 並列化はオーバーヘッドがある。計算が重いときだけ使う。
- スレッドセーフなコレクションを使う: どうしても外部リストに追加したいなら
ConcurrentLinkedQueueなどを使う。ただし collect の方が推奨。 - 副作用を排除する設計: 並列ストリームは「純粋関数的に」処理するのが鉄則。
まとめ
- 並列ストリームは強力だが、副作用を伴う処理は危険。
- 外部状態を書き換えず、reduce/collect などの組み込み集約を使うのが安全。
- 順序保証が必要なら
forEachOrderedを選ぶ。 - 並列化は「重い計算処理」にだけ使い、軽い処理では直列の方が効率的。
👉 練習課題: 「1〜1000 の偶数を並列ストリームで抽出し、合計を求める」コードを書いてみると、副作用を避けた安全な並列処理が体感できます。