並列ストリーム：性能と注意点

並列ストリームを有効にする

long count = list.parallelStream()
    .filter(n -> n % 2 == 0)
    .count();
System.out.println(count);

並列処理が効果を発揮する場合

// Good candidate: CPU-heavy computation on large dataset
long sum = LongStream.rangeClosed(1, 10_000_000)
    .parallel()
    .filter(n -> isPrime(n))
    .sum();
System.out.println(sum);

並列ストリームを使用してはいけない場合

// Bad: I/O bound — blocking ForkJoinPool threads starves other tasks
List<String> result = urls.parallelStream()
    .map(url -> httpGet(url)) // BLOCKS the common pool
    .collect(Collectors.toList());

共有可変状態のバグ

// RACE CONDITION — never do this:
List<Integer> results = new ArrayList<>(); // not thread-safe
numbers.parallelStream().forEach(n -> results.add(n)); // corrupts list!
// Fix:
List<Integer> safe = numbers.parallelStream().collect(Collectors.toList());

ステートフルな操作: sorted と distinct

// sorted() forces collect-all, then sort — parallel overhead usually not worth it:
list.parallelStream().sorted().collect(Collectors.toList());

順序に依存する操作

// Faster in parallel (order-insensitive):
Optional<Integer> any = list.parallelStream().filter(n -> n > 10).findAny();
// Slower in parallel (must preserve order):
Optional<Integer> first = list.parallelStream().filter(n -> n > 10).findFirst();

スレッドプールのサイズを選択する

ForkJoinPool custom = new ForkJoinPool(8);
long result = custom.submit(
    () -> LongStream.rangeClosed(1, 1_000_000).parallel().sum()
).get();

並列処理と逐次処理のベンチマーク

@Benchmark
public long sequential() { return LongStream.rangeClosed(1,1_000_000).sum(); }
@Benchmark
public long parallel()   { return LongStream.rangeClosed(1,1_000_000).parallel().sum(); }

並列処理での縮約操作

// Associative reduce — safe in parallel:
int sum = list.parallelStream().reduce(0, Integer::sum);
// Non-associative: subtraction — NOT safe in parallel:
int bad = list.parallelStream().reduce(0, (a, b) -> a - b); // wrong result!

分割可能性が重要

まとめ: 並列ストリームのチェックリスト

確認問題

振り返り