1. Stream API のおさらい
すでに Stream API に馴染みがあるでしょう。これはコレクション操作の便利な手段で、データのフィルタリング、ソート、集計などの処理をコンパクトで読みやすく記述できます。
典型的な例は次のとおりです:
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int sum = numbers.stream()
.filter(n -> n % 2 == 0)
.mapToInt(n -> n)
.sum();
System.out.println(sum); // 6 (2 + 4)
この例では、コレクションをストリーム（stream()）に変換し、偶数のみを抽出してから int に変換し、sum() で合計しています。
Stream API によってコードは短く表現力豊かになります。処理を段階的に「どのように」行うかを書く代わりに、「何を」得たいのかを宣言するだけで済みます。さらに必要なら、たった 1 行で簡単に並列処理へ切り替えることができます。
2. 並列ストリーム: 構文と動作原理
ストリームを並列化するには？
簡単です。stream() の代わりに parallelStream() を使うか、既存のストリームに対して .parallel() を呼びます。
List<Integer> numbers = ...;
int sum = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.mapToInt(n -> n)
.sum();
あるいは次のように:
numbers.stream()
.parallel() // 並列ストリームに変換
.filter(...)
.map(...)
.sum();
内部では何が起きる？
- コレクションは自動的に複数のチャンクに分割されます。
- 各チャンクは別々のスレッドで処理されます（ForkJoinPool という専用のスレッドプールを使用）。
- 結果が結合され、最終的な値が得られます。
つまり、多コア CPU であれば実際に並行して処理されます。たとえば、フィルタリングや合計計算が複数コアで同時に進みます。
どんな場合に特に有効？
- 大規模なコレクション（数万要素以上）。
- 各要素ごとの計算が重い場合。
- 厳密な処理順序を保持する必要がない場合。
例: 逐次ストリームと並列ストリームの比較
大きな配列で簡単に試してみましょう。
import java.util.*;
import java.util.stream.*;
public class ParallelStreamDemo {
public static void main(String[] args) {
List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000)
.boxed()
.collect(Collectors.toList());
// 逐次ストリーム
long time1 = System.currentTimeMillis();
long count1 = numbers.stream()
.filter(n -> n % 2 == 0)
.count();
long time2 = System.currentTimeMillis();
System.out.println("逐次: " + (time2 - time1) + " ms, 偶数: " + count1);
// 並列ストリーム
long time3 = System.currentTimeMillis();
long count2 = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.count();
long time4 = System.currentTimeMillis();
System.out.println("並列: " + (time4 - time3) + " ms, 偶数: " + count2);
}
}
このコードを自分のマシンで試してみてください。多くの場合、並列ストリームの方が速く処理できるはずです（特に多コア CPU の場合）。ただし、常にそうとは限りません。詳細は後述します。
3. 仕組み: ForkJoinPool と自動分割
並列ストリームは内部で ForkJoinPool.commonPool() を使用し、スレッド数を自動的に管理します（通常は利用可能な CPU コア数）。
概要図:
+-----------------------------+
| Your collection |
+-----------------------------+
| 1 | 2 | 3 | ... | 10M |
+----+----+----+-----+--------+
| | | |
v v v v
[Thread1][Thread2]...[ThreadN]
| | | |
+----+----+-----------+
|
[Merge results]
各スレッドが自分のチャンクを処理し、最後に結果が結合されます.
4. 制約と落とし穴
並列ストリームは「すべてを高速化」する魔法のボタンではありません。 場合によっては、むしろ遅くなることもあります。
並列化が不利な場合:
- コレクションが小さい（約 1000 要素以下）。
- 各要素の処理が非常に軽い（例: 単なる n * 2）。
- 厳密な処理順序の保持が重要（例: ファイルへの逐次書き込み）。
なぜ？ スレッドの生成や同期にもコストがかかります。タスク自体が小さいと、並列化のオーバーヘッドが利点を上回ることがあります。
副作用は並列化の敵
ストリーム内の処理で外部の変数を変更している場合は注意が必要です。
悪い例:
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int[] sum = {0};
numbers.parallelStream().forEach(n -> sum[0] += n);
System.out.println(sum[0]); // ??? (15 を期待しても、何が出るかわからない)
なぜでしょうか。複数のスレッドが同じ変数を同時に変更するため、race condition（競合状態）が発生します。結果は正しくならない可能性があります。
正しい方法 — ストリームの戻り値を使う:
int sum = numbers.parallelStream().mapToInt(n -> n).sum();
すべてのコレクションが同じように並列化しやすいわけではない
例えば通常の ArrayList は分割しやすい一方で、LinkedList や無限ストリーム（例: Stream.generate(...)）はあまり向いていません。
5. 実践: 性能比較
例: 最大値の検索
import java.util.*;
import java.util.stream.*;
public class ParallelMaxDemo {
public static void main(String[] args) {
List<Integer> numbers = IntStream.rangeClosed(1, 30_000_000)
.boxed()
.collect(Collectors.toList());
// 逐次
long t1 = System.currentTimeMillis();
int max1 = numbers.stream().max(Integer::compareTo).get();
long t2 = System.currentTimeMillis();
System.out.println("逐次: " + (t2 - t1) + " ms, max = " + max1);
// 並列
long t3 = System.currentTimeMillis();
int max2 = numbers.parallelStream().max(Integer::compareTo).get();
long t4 = System.currentTimeMillis();
System.out.println("並列: " + (t4 - t3) + " ms, max = " + max2);
}
}
どうなる？ 近年のマルチコア CPU では、並列ストリームの方が一般に高速です。しかし 30_000_000 を 1000 に変えると差は出ないか、むしろ並列の方が遅くなることもあります。
6. 使用例: フィルタ、集約、ソート
フィルタとカウント
List<String> names = Arrays.asList("Anya", "Boris", "Vasya", "Grisha", "Dasha", "Egor", "Zhenya");
long count = names.parallelStream()
.filter(name -> name.length() == 4)
.count();
System.out.println("長さが 4 の名前の数: " + count);
グルーピング
List<String> words = Arrays.asList("kot", "kit", "kot", "pes", "kit", "kot");
Map<String, Long> freq = words.parallelStream()
.collect(Collectors.groupingBy(
w -> w,
Collectors.counting()
));
System.out.println(freq); // {pes=1, kit=2, kot=3}
ソート（ここでは並列化が必ずしも有利になるとは限りません）
List<Integer> bigList = IntStream.rangeClosed(1, 5_000_000)
.boxed()
.collect(Collectors.toList());
long t1 = System.currentTimeMillis();
List<Integer> sorted = bigList.parallelStream()
.sorted()
.collect(Collectors.toList());
long t2 = System.currentTimeMillis();
System.out.println("並列ソート: " + (t2 - t1) + " ms");
7. 重要な注意点と推奨
parallelStream() を使うべきとき
- コレクションが大きい（数万要素以上）。
- 各要素の処理が「重い」（複雑な計算、ファイル/ネットワーク I/O など）。
- 順序への依存がない。
- 副作用がない（外部の変数を変更しない）。
parallelStream() を使うべきでないとき
- コレクションが小さい。
- 処理が軽い。
- 順序の厳密な保持が必要。
- 共有変数へアクセスしている（thread-safe なコレクションや別の手法を検討）。
使用されるスレッド数を知るには？
デフォルトでは CPU コア数に依存します: Runtime.getRuntime().availableProcessors()。この挙動はシステムプロパティで変更できます:
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");
意味を理解している場合のみ設定してください。CPU を使い切って、かえって遅くなることがあります。
8. 並列ストリームでよくある誤り
誤り 1: forEach 内の副作用
「並列でリストを埋めれば速いはずだ！」と考えがちです。
List<Integer> result = new ArrayList<>();
IntStream.range(0, 1_000)
.parallel()
.forEach(result::add); // 危険!
System.out.println(result.size()); // 結果は不定!
なぜ良くない？ ArrayList はスレッドセーフではなく、複数スレッドから同時に追加すると結果は予測不能です。欠落や重複、例外が起きることもあります。
解決策: ストリームの収集メソッド（collect）を使って安全に集めるか、専用のコレクションを使いましょう。
List<Integer> result = IntStream.range(0, 1_000)
.parallel()
.boxed()
.collect(Collectors.toList());
誤り 2: 小さなタスクでの速度向上を期待する
並列化は無料ではありません。コレクションが小さい場合、スケジューリングや同期のオーバーヘッドで並列ストリームの方が遅くなることがあります。
誤り 3: 順序の破壊
要素の順序が重要（例: ファイルへの書き込み）なら、並列ストリームは使わないでください。順序は保証されないか、順序を保とうとすると遅くなります。
誤り 4: 「相性の悪い」コレクションの使用
一部のコレクション（例: LinkedList、独自の構造）は分割しにくく、並列化の効率が下がります。
誤り 5: 結果の収集で thread-safety を無視する
結果を手作業で集める（例: リストへ追加する）場合は、スレッドセーフなコレクション（CopyOnWriteArrayList、ConcurrentLinkedQueue）や収集メソッドを使いましょう。
GO TO FULL VERSION