Java parallelStreamの使い方と注意点｜速度比較と順序

parallelStream とは、コレクションの要素を複数のスレッドに分けて同時に処理する「並列ストリーム」を作るメソッドです。 list.stream() を list.parallelStream() に変えるだけで並列になり、重い計算を多くの要素に行う処理では実行時間を短くできます。一方で、要素が少ない処理ではかえって時間がかかり、処理の順序や共有する変数の扱いにも気を配る必要があります。

この記事では、要素数（100件・1万件・1000万件）と処理の重さ別に stream と parallelStream の実行時間を実測し、速くなる条件と遅くなる条件を表にまとめました。さらに、ArrayList に forEach で add して件数が欠ける様子と ArrayIndexOutOfBoundsException を再現して直し方を示し、forEach と forEachOrdered の出力差、共通プールのスレッド数の確認コード、findFirst と findAny の違いも実行結果つきで載せています。

確認環境：Ubuntu 24.04 / OpenJDK 21.0.10 / CPU 2コア（java ファイル名.java の単一ファイル実行）。掲載した出力と実行時間は、すべてこの環境で実行して得たものです。

parallelStreamの早見表

速くなる条件・遅くなる条件（この記事の実測より）

parallelStreamの基本の使い方

並列ストリームの作り方は2通りです。

import java.util.*;
import java.util.stream.*;

public class Basic {
    public static void main(String[] args) {
        List<Integer> nums = List.of(1, 2, 3, 4, 5, 6, 7, 8);

        // 1. コレクションから直接、並列ストリームを作る
        int a = nums.parallelStream().mapToInt(n -> n * n).sum();
        // 2. 途中で parallel() を呼ぶ（配列や IntStream でも使える）
        int b = nums.stream().parallel().mapToInt(n -> n * n).sum();
        int c = IntStream.rangeClosed(1, 8).parallel().map(n -> n * n).sum();
        // 3. sequential() で順次に戻す
        Stream<Integer> s = nums.parallelStream().sequential();

        System.out.println(a + " " + b + " " + c);
        System.out.println("nums.stream().isParallel()                     : " + nums.stream().isParallel());
        System.out.println("nums.parallelStream().isParallel()             : " + nums.parallelStream().isParallel());
        System.out.println("nums.parallelStream().sequential().isParallel(): " + s.isParallel());

        // 結果をListで受け取る。並列でも元の順序のまま
        System.out.println(nums.parallelStream().map(n -> n * 10).toList());
        System.out.println(nums.parallelStream().filter(n -> n % 2 == 0).collect(Collectors.toList()));
    }
}

実行結果：

204 204 204
nums.stream().isParallel()                     : false
nums.parallelStream().isParallel()             : true
nums.parallelStream().sequential().isParallel(): false
[10, 20, 30, 40, 50, 60, 70, 80]
[2, 4, 6, 8]

3通りとも合計は同じ 204 です。map・filter・sum・toList などの書き方は stream と同じで、変わるのは「どのスレッドが、どの順番で処理するか」だけです。sequential() を呼ぶと順次に戻り、isParallel() が false になります。

共通プールのスレッド数を確認する

並列ストリームは、特に指定しなければ JVM に1つある共通プール（ForkJoinPool.commonPool）のスレッドで動きます。スレッド数は次のコードで確かめられます。

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class PoolInfo {
    public static void main(String[] args) {
        System.out.println("availableProcessors : " + Runtime.getRuntime().availableProcessors());
        System.out.println("共通プールの並列数  : " + ForkJoinPool.getCommonPoolParallelism());
        System.out.println("commonPool()        : " + ForkJoinPool.commonPool().getParallelism());

        Set<String> threads = new ConcurrentSkipListSet<>();
        IntStream.range(0, 1_000_000).parallel().forEach(i -> threads.add(Thread.currentThread().getName()));
        System.out.println("使われたスレッド数  : " + threads.size());
        threads.forEach(t -> System.out.println("  " + t));

        Set<String> seq = new TreeSet<>();
        IntStream.range(0, 1_000_000).forEach(i -> seq.add(Thread.currentThread().getName()));
        System.out.println("streamのスレッド    : " + seq);
    }
}

実行結果（CPU 2コアの環境）：

availableProcessors : 2
共通プールの並列数  : 1
commonPool()        : 1
使われたスレッド数  : 2
  ForkJoinPool.commonPool-worker-1
  main
streamのスレッド    : [main]

読み取れることは3つです。

• 共通プールの並列数は 1 で、availableProcessors（2）より1つ少ない値でした
• 実際に処理したスレッドは main と ForkJoinPool.commonPool-worker-1 の2つ。呼び出したスレッド（main）も処理に加わるので、合計は CPU のコア数と同じになります
• stream（順次）は main スレッドだけで処理します

つまり、並列でどれだけ速くなるかの上限は、ほぼ CPU のコア数で決まります。この環境では最大でも約2倍です。

並列数を変える

公式ドキュメントに記載のあるシステムプロパティ java.util.concurrent.ForkJoinPool.common.parallelism で、共通プールの並列数を指定できます。同じコードを次のように起動しました。

$ java -Djava.util.concurrent.ForkJoinPool.common.parallelism=4 PoolInfo.java
availableProcessors : 2
共通プールの並列数  : 4
commonPool()        : 4
使われたスレッド数  : 5
  ForkJoinPool.commonPool-worker-1
  ForkJoinPool.commonPool-worker-2
  ForkJoinPool.commonPool-worker-3
  ForkJoinPool.commonPool-worker-4
  main
streamのスレッド    : [main]

ワーカーが4つに増え、main と合わせて5つのスレッドが使われました。共通プールはアプリケーション全体で共有されるので、変更するときは他の並列処理への影響も合わせて確かめます。

速度比較：要素数と処理の重さ別の実測

要素数を 100件・1万件・1000万件、処理を「軽い（掛け算1回）」「重い（平方根を200回）」に分けて、stream と parallelStream の時間を測りました。ウォームアップのあと複数回実行し、いちばん短かった時間を載せています。

import java.util.*;
import java.util.function.*;
import java.util.stream.*;

public class Bench {
    // 軽い処理: 掛け算1回
    static long light(int x) { return x * 2L; }
    // 重い処理: 平方根を200回計算
    static long heavy(int x) {
        double d = x;
        for (int i = 0; i < 200; i++) { d = Math.sqrt(d + i); }
        return (long) d;
    }

    static double best(LongSupplier f, int warm, int runs) {
        for (int i = 0; i < warm; i++) { f.getAsLong(); }
        long best = Long.MAX_VALUE;
        for (int i = 0; i < runs; i++) {
            long t = System.nanoTime();
            f.getAsLong();
            best = Math.min(best, System.nanoTime() - t);
        }
        return best / 1_000_000.0;
    }

    static void compare(String label, List<Integer> data, IntToLongFunction op, int warm, int runs) {
        double seq = best(() -> data.stream().mapToLong(x -> op.applyAsLong(x)).sum(), warm, runs);
        double par = best(() -> data.parallelStream().mapToLong(x -> op.applyAsLong(x)).sum(), warm, runs);
        System.out.printf("%-26s stream %10.3f ms / parallelStream %10.3f ms / 比 %.2f倍%n", label, seq, par, seq / par);
    }

    public static void main(String[] args) {
        System.out.println("Java " + Runtime.version() + " / CPU " + Runtime.getRuntime().availableProcessors()
            + " / 共通プールの並列数 " + java.util.concurrent.ForkJoinPool.getCommonPoolParallelism());
        for (int n : new int[]{100, 10_000, 10_000_000}) {
            List<Integer> data = IntStream.range(0, n).boxed().collect(Collectors.toCollection(ArrayList::new));
            int warm = n >= 10_000_000 ? 2 : 200, runs = n >= 10_000_000 ? 5 : 50;
            compare(String.format("軽い処理 %,d件", n), data, Bench::light, warm, runs);
            compare(String.format("重い処理 %,d件", n), data, Bench::heavy, warm, n >= 10_000_000 ? 3 : runs);
        }

        // 待ち時間のある処理: 1件ごとに 10 ミリ秒待つ
        List<Integer> small = IntStream.range(0, 100).boxed().toList();
        LongSupplier seqWait = () -> small.stream().mapToLong(Bench::sleep10).sum();
        LongSupplier parWait = () -> small.parallelStream().mapToLong(Bench::sleep10).sum();
        double s = best(seqWait, 0, 1), p = best(parWait, 0, 1);
        System.out.printf("%-26s stream %10.3f ms / parallelStream %10.3f ms / 比 %.2f倍%n", "10ms待つ処理 100件", s, p, s / p);

        // データの入れ物による違い（軽い処理 100万件）
        List<Integer> arrayList = IntStream.range(0, 1_000_000).boxed().collect(Collectors.toCollection(ArrayList::new));
        List<Integer> linkedList = new LinkedList<>(arrayList);
        compare("ArrayList 100万件 重い処理", arrayList, Bench::heavy, 3, 5);
        compare("LinkedList 100万件 重い処理", linkedList, Bench::heavy, 3, 5);
        double it1 = best(() -> Stream.iterate(0, i -> i + 1).limit(1_000_000).mapToLong(Bench::heavy).sum(), 3, 5);
        double it2 = best(() -> Stream.iterate(0, i -> i + 1).parallel().limit(1_000_000).mapToLong(Bench::heavy).sum(), 3, 5);
        System.out.printf("%-26s stream %10.3f ms / parallelStream %10.3f ms / 比 %.2f倍%n", "iterate+limit 100万件 重い処理", it1, it2, it1 / it2);
    }

    static long sleep10(int x) {
        try { Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return x;
    }
}

java -Xmx3g Bench.java の実行結果：

Java 21.0.10+7-Ubuntu-124.04 / CPU 2 / 共通プールの並列数 1
軽い処理 100件                  stream      0.002 ms / parallelStream      0.010 ms / 比 0.25倍
重い処理 100件                  stream      0.124 ms / parallelStream      0.066 ms / 比 1.88倍
軽い処理 10,000件               stream      0.015 ms / parallelStream      0.013 ms / 比 1.18倍
重い処理 10,000件               stream     12.649 ms / parallelStream      7.269 ms / 比 1.74倍
軽い処理 10,000,000件           stream     36.447 ms / parallelStream     18.756 ms / 比 1.94倍
重い処理 10,000,000件           stream  13864.342 ms / parallelStream   7357.329 ms / 比 1.88倍
10ms待つ処理 100件              stream   1018.700 ms / parallelStream    507.113 ms / 比 2.01倍
ArrayList 100万件 重い処理       stream   1386.812 ms / parallelStream    725.714 ms / 比 1.91倍
LinkedList 100万件 重い処理      stream   1337.592 ms / parallelStream    895.555 ms / 比 1.49倍
iterate+limit 100万件 重い処理   stream   1363.839 ms / parallelStream    870.527 ms / 比 1.57倍

「比」は stream の時間 ÷ parallelStream の時間で、1より大きいほど並列のほうが短時間です。

並列にすると、要素を分割する・スレッドに渡す・結果をまとめる、という準備の時間が加わります。「1件あたりの処理時間 × 件数」がこの準備時間より十分大きいときに、並列の効果が出ます。 軽い処理100件は全体で 0.002 ms しかかからないため、準備時間のほうが大きくなりました。

なお、この数値は CPU 2コアの共有マシンで、ほかの処理も動いている状態での計測です。実行のたびに数値は揺れ、コア数が多いマシンでは比も変わります。大小関係の目安として読み、実際に使う環境とデータ量で測って決めるのが確実です。

ロックを取り合う処理は速くならない

もう1つ、全スレッドが同じロックを取り合う処理を測りました。

import java.util.*;
import java.util.function.*;
import java.util.stream.*;

public class Bench2 {
    static final Object LOCK = new Object();

    static long heavy(int x) {
        double d = x;
        for (int i = 0; i < 200; i++) { d = Math.sqrt(d + i); }
        return (long) d;
    }
    // 全スレッドで1つのロックを取り合う
    static long heavyLocked(int x) { synchronized (LOCK) { return heavy(x); } }

    static double best(LongSupplier f) {
        for (int i = 0; i < 3; i++) { f.getAsLong(); }
        long best = Long.MAX_VALUE;
        for (int i = 0; i < 5; i++) {
            long t = System.nanoTime();
            f.getAsLong();
            best = Math.min(best, System.nanoTime() - t);
        }
        return best / 1_000_000.0;
    }

    static void print(String label, double seq, double par) {
        System.out.printf("%-30s stream %9.3f ms / parallelStream %9.3f ms / 比 %.2f倍%n", label, seq, par, seq / par);
    }

    public static void main(String[] args) {
        List<Integer> data = IntStream.range(0, 300_000).boxed().collect(Collectors.toCollection(ArrayList::new));

        print("ロックなし 30万件",
            best(() -> data.stream().mapToLong(Bench2::heavy).sum()),
            best(() -> data.parallelStream().mapToLong(Bench2::heavy).sum()));
        print("共有ロックあり 30万件",
            best(() -> data.stream().mapToLong(Bench2::heavyLocked).sum()),
            best(() -> data.parallelStream().mapToLong(Bench2::heavyLocked).sum()));
    }
}

実行結果：

ロックなし 30万件                     stream   419.513 ms / parallelStream   161.095 ms / 比 2.60倍
共有ロックあり 30万件                   stream   394.017 ms / parallelStream   683.211 ms / 比 0.58倍

ロックがなければ並列で短くなりましたが、処理全体を1つの synchronized で囲むと、同時に動けるスレッドは1つだけになり、並列のほうが時間がかかりました（0.58倍）。共有するロックや、同時に1つしか使えない資源を待つ処理は、並列にしても効果が出にくいと言えます。

速くなる条件・遅くなる条件のまとめ

注意点1：ArrayListにforEachでaddすると件数が欠ける

並列ストリームでいちばん気をつけたいのが、ラムダ式の中から、スレッドセーフでない共有オブジェクトを書き換える書き方です。ArrayList に forEach で add する例を再現しました。

import java.util.*;
import java.util.stream.*;

public class Broken {
    public static void main(String[] args) {
        for (int t = 1; t <= 8; t++) {
            List<Integer> result = new ArrayList<>();
            try {
                IntStream.range(0, 100_000).parallel().forEach(i -> result.add(i));
                System.out.println(t + "回目: 件数 " + result.size());
            } catch (Exception e) {
                System.out.println(t + "回目: " + e);
            }
        }
    }
}

実行結果：

1回目: 件数 50201
2回目: 件数 68218
3回目: 件数 80504
4回目: 件数 100000
5回目: 件数 83169
6回目: 件数 100000
7回目: 件数 76132
8回目: java.lang.ArrayIndexOutOfBoundsException: Index 2790 out of bounds for length 2776

10万件を add したはずが、件数が 50201 件や 68218 件に欠け、8回目は ArrayIndexOutOfBoundsException で止まりました。たまたま 100000 件そろう回もあるため、テストでは気づきにくいのが特徴です。公式ドキュメントのパッケージ説明でも ArrayList はスレッドセーフでないコレクションの例に挙げられており、複数のスレッドから同時に add する使い方には向きません。

直し方：collect や toList で集める

import java.util.*;
import java.util.concurrent.atomic.*;
import java.util.stream.*;

public class Fixed {
    public static void main(String[] args) {
        // 直し方1: 終端操作で集める（おすすめ）
        List<Integer> a = IntStream.range(0, 100_000).parallel().boxed().toList();
        List<Integer> b = IntStream.range(0, 100_000).parallel().boxed().collect(Collectors.toList());
        System.out.println("toList()            : 件数 " + a.size() + " 先頭 " + a.subList(0, 5));
        System.out.println("Collectors.toList() : 件数 " + b.size() + " 先頭 " + b.subList(0, 5));

        // 直し方2: スレッドセーフなリストに add する（件数はそろうが順序はばらばら）
        List<Integer> c = Collections.synchronizedList(new ArrayList<>());
        IntStream.range(0, 100_000).parallel().forEach(c::add);
        System.out.println("synchronizedList    : 件数 " + c.size() + " 先頭 " + c.subList(0, 5));

        // カウンタも同じ（100万回）: int[] への加算は欠けることがある
        int[] counter = {0};
        IntStream.range(0, 1_000_000).parallel().forEach(i -> counter[0]++);
        AtomicInteger atomic = new AtomicInteger();
        IntStream.range(0, 1_000_000).parallel().forEach(i -> atomic.incrementAndGet());
        long count = IntStream.range(0, 1_000_000).parallel().count();
        System.out.println("int[] に ++         : " + counter[0]);
        System.out.println("AtomicInteger       : " + atomic.get());
        System.out.println("count()             : " + count);

        // HashMap への put も同じ。groupingBy で集める
        Map<Integer, Long> grouped = IntStream.range(0, 100_000).parallel().boxed()
            .collect(Collectors.groupingBy(i -> i % 3, Collectors.counting()));
        System.out.println("groupingBy          : " + grouped);
    }
}

実行結果：

toList()            : 件数 100000 先頭 [0, 1, 2, 3, 4]
Collectors.toList() : 件数 100000 先頭 [0, 1, 2, 3, 4]
synchronizedList    : 件数 100000 先頭 [50000, 50001, 50002, 50003, 50004]
int[] に ++         : 970521
AtomicInteger       : 1000000
count()             : 1000000
groupingBy          : {0=33334, 1=33333, 2=33333}

「外の変数に書き込む」のではなく、「終端操作の戻り値として受け取る」形にするのが基本です。公式ドキュメントのパッケージ説明でも、forEach で ArrayList に add する書き方を、副作用のない集め方に置き換える例が示されています。

注意点2：forEachの順序とforEachOrdered

並列ストリームの forEach は、要素を元の順序で処理するとは限りません。

import java.util.*;
import java.util.stream.*;

public class Order {
    public static void main(String[] args) {
        List<Integer> nums = IntStream.rangeClosed(1, 12).boxed().toList();

        System.out.print("stream forEach           : ");
        nums.stream().forEach(n -> System.out.print(n + " "));
        System.out.println();

        for (int t = 1; t <= 3; t++) {
            System.out.print("parallel forEach (" + t + "回目) : ");
            nums.parallelStream().forEach(n -> System.out.print(n + " "));
            System.out.println();
        }

        System.out.print("parallel forEachOrdered  : ");
        nums.parallelStream().forEachOrdered(n -> System.out.print(n + " "));
        System.out.println();

        System.out.println("parallel map + toList    : " + nums.parallelStream().map(n -> n * 2).toList());
        System.out.println("parallel sorted + toList : " + nums.parallelStream().sorted(Comparator.reverseOrder()).toList());

        System.out.println("--- どのスレッドが処理したか（forEachOrdered でも map は並列）");
        List.of("A", "B", "C", "D", "E", "F").parallelStream()
            .map(s -> s + "@" + Thread.currentThread().getName())
            .forEachOrdered(System.out::println);
    }
}

実行結果：

stream forEach           : 1 2 3 4 5 6 7 8 9 10 11 12 
parallel forEach (1回目) : 7 8 9 10 11 12 4 5 6 1 2 3 
parallel forEach (2回目) : 7 8 9 10 11 12 4 5 6 1 2 3 
parallel forEach (3回目) : 7 4 5 6 1 2 3 10 11 12 8 9 
parallel forEachOrdered  : 1 2 3 4 5 6 7 8 9 10 11 12 
parallel map + toList    : [2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24]
parallel sorted + toList : [12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1]
--- どのスレッドが処理したか（forEachOrdered でも map は並列）
A@main
B@ForkJoinPool.commonPool-worker-1
C@main
D@main
E@main
F@main

公式ドキュメントでは、forEach の動作は明示的に非決定的で、並列ストリームでは元の順序を守ることを保証しない、と説明されています。順番に意味がある出力や書き込みには forEachOrdered を使います。

最後の6行は、forEachOrdered を使っても map の部分は複数のスレッド（main と worker-1）で処理されていることを示しています。forEachOrdered は「最後の処理を順序どおりに行う」もので、途中の map や filter は並列のままです。また、toList() や collect で受け取る場合は、並列でも元の順序が保たれます。順序が必要なら「forEach で出力する」より「List で受け取ってから使う」ほうが扱いやすくなります。

注意点3：findFirstとfindAnyの違い

import java.util.*;
import java.util.stream.*;

public class Find {
    public static void main(String[] args) {
        List<Integer> nums = IntStream.rangeClosed(1, 1_000_000).boxed().toList();

        System.out.println("stream   findFirst: " + nums.stream().filter(n -> n % 7 == 0).findFirst().get());
        System.out.println("stream   findAny  : " + nums.stream().filter(n -> n % 7 == 0).findAny().get());
        for (int t = 1; t <= 5; t++) {
            int first = nums.parallelStream().filter(n -> n % 7 == 0).findFirst().get();
            int any = nums.parallelStream().filter(n -> n % 7 == 0).findAny().get();
            System.out.println("parallel " + t + "回目  findFirst: " + first + "  findAny: " + any);
        }
        System.out.println("parallel anyMatch : " + nums.parallelStream().anyMatch(n -> n == 999_999));
    }
}

実行結果：

stream   findFirst: 7
stream   findAny  : 7
parallel 1回目  findFirst: 7  findAny: 500003
parallel 2回目  findFirst: 7  findAny: 250005
parallel 3回目  findFirst: 7  findAny: 250005
parallel 4回目  findFirst: 7  findAny: 250005
parallel 5回目  findFirst: 7  findAny: 250005
parallel anyMatch : true

findFirst は並列でも元の順序で最初の要素を返し、findAny は見つかったどれか1つを返します。 「条件に合うものが1つあればよい」場面では findAny、「最初の1件」に意味がある場面では findFirst を選びます。存在するかどうかだけを知りたいなら anyMatch が使えます。

注意点4：reduceに渡す関数と初期値

reduce は、並列ではデータをいくつかに分けて計算し、あとで結果をまとめます。そのため、渡す関数と初期値によっては結果が変わります。

import java.util.*;
import java.util.stream.*;

public class Reduce {
    public static void main(String[] args) {
        List<Integer> nums = IntStream.rangeClosed(1, 1000).boxed().toList();

        System.out.println("足し算 stream   : " + nums.stream().reduce(0, Integer::sum));
        System.out.println("足し算 parallel : " + nums.parallelStream().reduce(0, Integer::sum));
        System.out.println("引き算 stream   : " + nums.stream().reduce(0, (a, b) -> a - b));
        System.out.println("引き算 parallel : " + nums.parallelStream().reduce(0, (a, b) -> a - b));
        System.out.println("初期値10 stream   : " + nums.stream().reduce(10, Integer::sum));
        System.out.println("初期値10 parallel : " + nums.parallelStream().reduce(10, Integer::sum));
        System.out.println("文字連結 parallel : " + Stream.of("a", "b", "c", "d", "e", "f").parallel().collect(Collectors.joining()));
    }
}

実行結果：

足し算 stream   : 500500
足し算 parallel : 500500
引き算 stream   : -500500
引き算 parallel : 0
初期値10 stream   : 500510
初期値10 parallel : 500540
文字連結 parallel : abcdef

並列で reduce を使うときは、「どこで区切っても結果が同じになる関数」と「何回使っても結果を変えない初期値（足し算なら 0）」を渡します。 合計なら mapToInt(...).sum()、文字列の連結なら Collectors.joining() のように用意されたメソッドを使うと、並列でも同じ結果になります。

parallelStreamを使うかどうかの判断表

書いたあとのチェックリスト

• ラムダ式の中で、外の ArrayList・HashMap・配列・変数に書き込んでいないか
• 結果は toList()・collect・sum・count などの戻り値で受け取っているか
• 順序が必要な出力に forEachOrdered を使っているか
• findAny と findFirst を意図どおりに選んでいるか
• reduce の関数は、区切り方を変えても同じ結果になるか
• stream と parallelStream の両方で時間を測ったか

よくある質問

Q. parallelStream にすれば必ず速くなりますか。 A. なりません。この記事の計測では、掛け算1回の軽い処理を100件に行うと stream が 0.002 ms、parallelStream が 0.010 ms で、並列のほうが時間がかかりました。速度比較の表では、重い処理や1000万件の処理は CPU 2コアの環境で約1.7〜2倍短くなりました。

Q. parallelStream で処理した結果の順番はどうなりますか。 A. forEach は順番が実行ごとに変わります。順番どおりに処理したいときは forEachOrdered を使います。toList() や collect(Collectors.toList()) で受け取った List は、並列でも元の順序のままでした。

Q. 並列ストリームは何本のスレッドで動きますか。 A. 特に指定しなければ共通プールのスレッドと、呼び出したスレッドで動きます。この環境（CPU 2コア）では共通プールの並列数が 1 で、main と合わせて2つのスレッドが使われました。ForkJoinPool.getCommonPoolParallelism() で確かめられます。

Q. parallelStream の中で List に add してもよいですか。 A. ArrayList への add は件数が欠けたり ArrayIndexOutOfBoundsException になったりします。toList() や collect(Collectors.toList()) で集める形に書き換えてください。

まとめ

• parallelStream() か parallel() で並列になり、書き方は stream と同じ
• 速くなるのは「重い処理」か「非常に多い件数」。軽い処理100件では並列のほうが時間がかかった
• 速さの上限はほぼ CPU のコア数。共通プールの並列数はこの環境で availableProcessors − 1 だった
• 外の ArrayList に forEach で add すると件数が欠け、例外になることもある。toList() や collect で集める
• forEach は順序が変わる。順序が必要なら forEachOrdered、または List で受け取る
• findFirst は並列でも先頭、findAny はどれか1つ
• reduce には区切り方で結果が変わらない関数と初期値を渡す。最後は実際のデータ量で測って決める

参考・出典

• Oracle「Java SE 21 API仕様 Stream」https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/stream/Stream.html
• Oracle「Java SE 21 API仕様 java.util.stream パッケージ」https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/stream/package-summary.html
• Oracle「Java SE 21 API仕様 ForkJoinPool」https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/ForkJoinPool.html

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。