Javaの並列処理をマスター！parallelStreamの使い方と注意点を徹底解説

1. Javaの並列処理とparallelStreamの概要

1. Javaの並列処理とparallelStreamの概要

Javaのプログラムは、通常は上から順番に一つの道筋で処理が進んでいきます。これを「シングルスレッド」と呼びます。しかし、最近のコンピューターは複数の頭脳（CPUコア）を持っており、それらを同時に動かすことで処理速度を上げることができます。これを「並列処理」と言います。

Java 8から導入された「Stream API」には、この並列処理を極めて簡単に実現するための「parallelStream」という仕組みが備わっています。リストやセットなどのコレクションに対して、通常の stream() メソッドの代わりに parallelStream() を呼び出すだけで、内部的にデータを分割し、複数のスレッドで同時に計算を行ってくれるようになります。大量の数値を計算したり、複雑なデータ加工を行ったりする場合に、劇的なパフォーマンス向上が期待できる非常に便利な機能です。

2. parallelStreamの基本的な使い方

2. parallelStreamの基本的な使い方

まずは、最もシンプルな使い方を確認しましょう。整数のリストの各要素を二倍にして表示する処理を、並列で行う例です。通常のストリームとの違いは、メソッド名が parallelStream() になっている点だけです。

import java.util.Arrays;
import java.util.List;

public class ParallelBasicExample {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        System.out.println("--- 並列ストリームの実行結果 ---");
        numbers.parallelStream()
               .map(n -> n * 2)
               .forEach(n -> System.out.print(n + " "));
    }
}

このプログラムを実行すると、出力される数字の順番がバラバラになることに気づくはずです。これは、複数の処理が「同時並行」で行われ、終わったものから順に出力されているためです。

--- 並列ストリームの実行結果 ---
12 14 16 18 20 2 4 6 8 10 

3. 逐次処理と並列処理の動作の違い

3. 逐次処理と並列処理の動作の違い

通常の stream()（逐次処理）と parallelStream()（並列処理）では、内部的な動きが大きく異なります。逐次処理では「一つ終わったら次へ」という流れですが、並列処理では「フォーク・ジョイン（Fork/Join）フレームワーク」という仕組みが働きます。

具体的には、まず対象のコレクションを小さな塊に「分割（Fork）」します。それぞれの塊を別々のCPUコアで計算し、最後にその結果を一つに「結合（Join）」して最終的な答えを出します。この分割と結合をJavaが自動で管理してくれるため、開発者は複雑なスレッド管理を意識せずに済みます。しかし、分割する手間が発生するため、データ量が少ない場合は逆に遅くなることもあります。

4. パフォーマンスを最大化するための条件

4. パフォーマンスを最大化するための条件

並列処理を使えば何でも速くなるわけではありません。効果を発揮するためには、いくつかの条件を満たす必要があります。まず第一に「データ量が十分に多いこと」です。数件程度のデータでは、スレッドを立ち上げるコストの方が高くついてしまいます。目安として、数万件以上のデータがある場合に検討するのが一般的です。

第二に「個々の処理が重いこと」です。複雑な計算や通信待ちが発生するような処理は並列化のメリットが大きいです。反対に、単純な足し算だけのような軽い処理では、データの分割コストを上回るほどの恩恵が得られにくいです。また、使用しているマシンのCPUコア数も重要です。1コアしかない古いPCでは、並列に動かそうとしても順番待ちが発生するため、速度は向上しません。

5. 副作用とスレッドセーフに関する重大な注意点

5. 副作用とスレッドセーフに関する重大な注意点

parallelStreamを使用する際に最も注意しなければならないのが「副作用（Side Effects）」です。並列処理の中で、外部の変数を書き換えたり、スレッドセーフではない共有オブジェクト（ArrayListなど）にデータを追加したりすると、データが壊れたり、予期しないエラーが発生したりします。

以下のコードは、スレッドセーフではないリストに対して並列にデータを追加しようとする危険な例です。

import java.util.ArrayList;
import java.util.List;
import java.util.stream.IntStream;

public class UnsafeParallel {
    public static void main(String[] args) {
        List<Integer> resultList = new ArrayList<>();
        
        // 危険：ArrayListはスレッドセーフではないため、並列処理でaddすると壊れる可能性がある
        IntStream.range(0, 1000).parallel().forEach(resultList::add);
        
        System.out.println("処理後のリストのサイズ: " + resultList.size());
    }
}

このコードを実行すると、サイズが1000にならなかったり、ArrayIndexOutOfBoundsExceptionが発生したりすることがあります。並列処理では、共通の器にデータを入れるのではなく、次のセクションで紹介する collect メソッドなどを使って安全に結果をまとめる必要があります。

6. 安全に結果を取得するcollectメソッド

6. 安全に結果を取得するcollectメソッド

前述の通り、共有された変数に直接値を書き込むのは厳禁です。安全に並列処理の結果を収集するには、collect メソッドを使用します。これにより、Javaが各スレッドの結果を適切にマージしてくれます。

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class SafeParallelExample {
    public static void main(String[] args) {
        // 安全な方法：Collectors.toList() を使う
        List<Integer> safeList = IntStream.range(0, 1000)
                                          .parallel()
                                          .boxed()
                                          .collect(Collectors.toList());
        
        System.out.println("安全に取得したサイズ: " + safeList.size());
    }
}

この方法であれば、内部の同期処理をJavaが保証してくれるため、マルチスレッド特有の難しいバグに悩まされることなく、安全に高速化を図ることができます。リストへの変換だけでなく、合計値の算出（sum）や平均値の計算（average）も、標準のメソッドを使えば安全に並列化可能です。

7. 順序性が重要な場合の対処法

7. 順序性が重要な場合の対処法

並列処理では、要素が処理される順番は保証されません。しかし、最終的な結果のリストだけは元の順番通りに並んでいてほしい、という場面があります。その場合は forEach の代わりに forEachOrdered を使用するか、collect を利用します。

forEachOrdered を使うと、処理自体は並列で行いつつ、出力の段階で元の順序を守るように制御されます。ただし、順序を守るためのコストがかかるため、通常の forEach よりもパフォーマンスは低下します。処理速度と順序性のどちらが優先されるかによって、使い分ける判断が必要です。

8. 処理の重さをシミュレーションした実践的な比較

8. 処理の重さをシミュレーションした実践的な比較

並列処理の威力を実感するために、意図的に重い処理（スリープ処理）を含めた比較プログラムを作成してみましょう。1秒かかる処理を5回行う場合、逐次処理では5秒かかりますが、並列処理ならどれくらい短縮されるでしょうか。

import java.util.Arrays;
import java.util.List;

public class PerformanceComparison {
    public static void main(String[] args) {
        List<String> tasks = Arrays.asList("タスクA", "タスクB", "タスクC", "タスクD", "タスクE");

        long start = System.currentTimeMillis();

        tasks.parallelStream().forEach(task -> {
            try {
                // 1秒間の重い処理をシミュレート
                Thread.sleep(1000);
                System.out.println(task + " 完了 (実行スレッド: " + Thread.currentThread().getName() + ")");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        long end = System.currentTimeMillis();
        System.out.println("全体の処理時間: " + (end - start) + "ミリ秒");
    }
}

このプログラムを実行すると、複数の「ForkJoinPool.commonPool-worker」という名前のスレッドが同時に動いていることがわかります。全体の処理時間は、CPUのコア数にもよりますが、1秒強から数秒程度となり、逐次処理よりも明らかに短縮されます。

9. parallelStreamを使うべきかどうかの判断基準

9. parallelStreamを使うべきかどうかの判断基準

最後に、実際の開発で並列ストリームを使うべきか判断するためのポイントを整理します。まずは「NQモデル」と呼ばれる考え方を参考にしましょう。Nは要素数、Qは一つの要素あたりの処理負荷です。N × Q の値が大きければ大きいほど、並列化のメリットが生まれます。

また、データ構造も重要です。ArrayList や配列のように、要素の分割が容易なものは並列化に向いています。一方で、LinkedList のように要素を辿るのに時間がかかる構造や、Files.lines のように入出力が絡むものは、分割効率が悪いため並列化の効果が薄いことが多いです。迷ったときは、実際に両方のパターンでベンチマークテストを行い、実行時間を計測して判断するのが最も確実な道と言えます。