Parallel Streams

// ── Creating a parallel stream ──────────────────────────────────────────
List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000).boxed().collect(Collectors.toList());

long sumSequential = numbers.stream()
    .mapToLong(Integer::longValue)
    .sum();

long sumParallel = numbers.parallelStream()           // from a Collection
    .mapToLong(Integer::longValue)
    .sum();

long sumParallel2 = numbers.stream()
    .parallel()                                         // converts an existing stream
    .mapToLong(Integer::longValue)
    .sum();

// ── Checking and influencing the common pool size ───────────────────────
System.out.println(Runtime.getRuntime().availableProcessors());
// Common pool parallelism defaults to availableProcessors() - 1
// (can be overridden via -Djava.util.concurrent.ForkJoinPool.common.parallelism=N,
//  but this affects the ENTIRE application's shared pool, not just one stream)

// ── Spliterator splitting behavior differs by source type ──────────────
Spliterator<Integer> arrayListSplit = new ArrayList<>(numbers).spliterator();  // cheap, even splits
Spliterator<Integer> linkedListSplit = new LinkedList<>(numbers).spliterator(); // expensive, node-walking splits

// Poorly splittable source — parallelizing this offers little/no benefit:
Stream<Integer> iterateStream = Stream.iterate(0, i -> i + 1).limit(1_000_000).parallel();

// ── Overhead dominates for small/cheap workloads — parallel can be SLOWER
List<Integer> small = IntStream.rangeClosed(1, 100).boxed().collect(Collectors.toList());

long startSeq = System.nanoTime();
long sumSeq = small.stream().mapToLong(i -> i).sum();
long seqTime = System.nanoTime() - startSeq;

long startPar = System.nanoTime();
long sumPar = small.parallelStream().mapToLong(i -> i).sum();
long parTime = System.nanoTime() - startPar;
// For a workload this small, parTime is very commonly LARGER than seqTime —
// splitting/scheduling/merging overhead outweighs any parallel speedup

// ── Where parallelism genuinely helps — large data, expensive per-element work
List<BigInteger> largeNumbers = generateMillionsOfLargeNumbers();

boolean anyPrime = largeNumbers.parallelStream()
    .anyMatch(JavaParallelStreams::isProbablePrime);   // expensive CPU-bound check per element,
                                                          // large source, array-backed — good fit

// ── PITFALL: blocking I/O inside a parallel stream starves the common pool
List<String> urls = getThousandsOfUrls();

// BAD — blocks shared ForkJoinPool threads on network I/O, starving the
// whole application's other parallel-stream and CompletableFuture work:
List<String> results = urls.parallelStream()
    .map(url -> blockingHttpGet(url))   // I/O-bound — wrong tool for the job
    .collect(Collectors.toList());

// BETTER — use a dedicated executor sized for I/O-bound concurrency instead:
ExecutorService ioPool = Executors.newFixedThreadPool(50);
List<CompletableFuture<String>> futures = urls.stream()
    .map(url -> CompletableFuture.supplyAsync(() -> blockingHttpGet(url), ioPool))
    .collect(Collectors.toList());
List<String> results2 = futures.stream().map(CompletableFuture::join).collect(Collectors.toList());
ioPool.shutdown();

// ── PITFALL: mutating shared external state from forEach — data race ──
List<Integer> results = new ArrayList<>();   // NOT thread-safe
numbers.parallelStream().forEach(n -> {
    if (n % 2 == 0) {
        results.add(n);   // BUG: concurrent adds from multiple threads — race condition,
                            // can throw ConcurrentModificationException or silently drop elements
    }
});

// ── CORRECT: use the stream's own collecting operations ────────────────
List<Integer> correctResults = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .collect(Collectors.toList());   // designed to safely merge per-thread partial results

// ── PITFALL: a shared, non-atomic counter mutated inside the pipeline ──
int[] counter = {0};   // effectively shared mutable state
numbers.parallelStream().forEach(n -> counter[0]++);   // BUG: lost updates under concurrency
System.out.println(counter[0]);   // unreliable — often less than numbers.size()

// ── CORRECT: let the stream compute the count itself ────────────────────
long count = numbers.parallelStream().count();          // or use an AtomicLong if accumulating manually

// ── Ordering: forEach has no order guarantee in parallel; forEachOrdered does
numbers.parallelStream().forEach(System.out::println);          // arbitrary order across threads
numbers.parallelStream().forEachOrdered(System.out::println);   // preserves encounter order,
                                                                    // but loses most parallel benefit

// ── findFirst/sorted/limit are comparatively more expensive in parallel ─
Optional<Integer> first = numbers.parallelStream()
    .filter(n -> n > 1000)
    .findFirst();   // must coordinate across chunks to honor encounter-order "first" semantics
// findAny() is cheaper in parallel when any matching element (not specifically the first) suffices