Java Stream API — Parallel Stream Data Corruption

import java.util.List;
import java.util.Optional;

public class StreamPipelineBasics {

    record Order(String id, String customerId, double total, String status) {}

    public static void main(String[] args) {

        List<Order> orders = List.of(
            new Order("ORD-001", "CUST-A", 149.99, "SHIPPED"),
            new Order("ORD-002", "CUST-B",  29.99, "PENDING"),
            new Order("ORD-003", "CUST-A", 299.00, "SHIPPED"),
            new Order("ORD-004", "CUST-C",  89.50, "CANCELLED"),
            new Order("ORD-005", "CUST-B", 450.00, "SHIPPED")
        );

        // STEP 1 — Source: .stream() registers intent, no work done yet
        // STEP 2 — Intermediate: filter keeps only SHIPPED orders over $100
        // STEP 3 — Intermediate: map extracts just the order ID string
        // STEP 4 — Terminal: findFirst() fires the pipeline, returns Optional
        Optional<String> firstHighValueShippedId = orders.stream()
            .filter(order -> order.status().equals("SHIPPED"))   // lazy
            .filter(order -> order.total() > 100.0)              // lazy
            .map(Order::id)                                      // lazy
            .findFirst();                                        // FIRES pipeline

        // Optional protects us from NullPointerException if nothing matched
        firstHighValueShippedId.ifPresent(id ->
            System.out.println("First high-value shipped order: " + id)
        );

        // Because of lazy evaluation, once ORD-001 passes both filters,
        // Java STOPS — ORD-003 and ORD-005 are never even evaluated.
        System.out.println("Pipeline executed with short-circuit optimisation");
    }
}

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

public class FilterMapCollectDemo {

    record Order(String id, String customerId, double total, String status) {}

    public static void main(String[] args) {

        List<Order> orders = List.of(
            new Order("ORD-001", "CUST-A", 149.99, "SHIPPED"),
            new Order("ORD-002", "CUST-B",  29.99, "PENDING"),
            new Order("ORD-003", "CUST-A", 299.00, "SHIPPED"),
            new Order("ORD-004", "CUST-C",  89.50, "CANCELLED"),
            new Order("ORD-005", "CUST-B", 450.00, "SHIPPED")
        );

        // --- USE CASE 1: Get IDs of all shipped orders as a List<String> ---
        List<String> shippedOrderIds = orders.stream()
            .filter(order -> order.status().equals("SHIPPED"))  // keep SHIPPED
            .map(Order::id)                                     // Order -> String
            .collect(Collectors.toList());                      // fire + gather

        System.out.println("Shipped order IDs: " + shippedOrderIds);

        // --- USE CASE 2: Total revenue per customer (SHIPPED only) ---
        // groupingBy partitions the stream into groups by a classifier key.
        // summingDouble then collapses each group into a single double.
        Map<String, Double> revenueByCustomer = orders.stream()
            .filter(order -> order.status().equals("SHIPPED"))
            .collect(
                Collectors.groupingBy(
                    Order::customerId,                           // group key
                    Collectors.summingDouble(Order::total)       // downstream collector
                )
            );

        System.out.println("\nRevenue by customer (shipped orders only):");
        // Sort by value descending for readable output
        revenueByCustomer.entrySet().stream()
            .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
            .forEach(entry ->
                System.out.printf("  %-8s -> $%.2f%n", entry.getKey(), entry.getValue())
            );

        // --- USE CASE 3: Build a comma-separated order summary string ---
        String orderSummary = orders.stream()
            .filter(order -> order.total() >= 100.0)
            .map(order -> order.id() + "(" + order.status() + ")")
            .collect(Collectors.joining(", ", "High-value orders: [", "]"));

        System.out.println("\n" + orderSummary);
    }
}

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

public class ReduceAndFlatMapDemo {

    record OrderItem(String productName, int quantity, double unitPrice) {
        double lineTotal() { return quantity * unitPrice; }
    }

    record Order(String id, String customerId, List<OrderItem> items) {
        double total() {
            // reduce with identity 0.0 — if items is empty, returns 0.0 safely
            return items.stream()
                .map(OrderItem::lineTotal)          // Stream<Double>
                .reduce(0.0, Double::sum);          // collapse to single double
        }
    }

    public static void main(String[] args) {

        List<Order> orders = List.of(
            new Order("ORD-001", "CUST-A", List.of(
                new OrderItem("Mechanical Keyboard", 1, 129.99),
                new OrderItem("USB Hub",             2,  24.99)
            )),
            new Order("ORD-002", "CUST-B", List.of(
                new OrderItem("Monitor",             1, 349.00)
            )),
            new Order("ORD-003", "CUST-A", List.of(
                new OrderItem("Mouse Pad",           3,   9.99),
                new OrderItem("Webcam",              1,  79.99)
            ))
        );

        // reduce: total revenue across ALL orders
        double totalRevenue = orders.stream()
            .map(Order::total)                      // Stream<Double> of order totals
            .reduce(0.0, Double::sum);              // sum them all

        System.out.printf("Total revenue: $%.2f%n", totalRevenue);

        // flatMap: get a FLAT list of every individual OrderItem across all orders
        // Without flatMap, .map(Order::items) gives Stream<List<OrderItem>> — nested!
        // flatMap unwraps each list and merges everything into one Stream<OrderItem>
        List<String> allProductNames = orders.stream()
            .flatMap(order -> order.items().stream()) // Stream<OrderItem> — flat!
            .map(OrderItem::productName)             // Stream<String>
            .sorted()                                // alphabetical
            .collect(Collectors.toList());

        System.out.println("\nAll products ordered (alphabetical):");
        allProductNames.forEach(name -> System.out.println("  - " + name));

        // reduce: find the single most expensive line item total
        orders.stream()
            .flatMap(order -> order.items().stream())
            .reduce((a, b) -> a.lineTotal() >= b.lineTotal() ? a : b) // no identity = Optional
            .ifPresent(item -> System.out.printf(
                "%nMost expensive line item: %s at $%.2f%n",
                item.productName(), item.lineTotal()
            ));
    }
}

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

public class ParallelStreamDemo {

    // Simulates a CPU-intensive scoring calculation
    static double calculateRiskScore(int orderId) {
        // Artificial workload — in real life this could be ML inference,
        // complex formula evaluation, or external data enrichment
        double score = 0;
        for (int i = 0; i < 10_000; i++) {
            score += Math.sin(orderId * i) * Math.cos(i);
        }
        return Math.abs(score % 100);
    }

    public static void main(String[] args) {

        // Generate 5000 order IDs to score
        List<Integer> orderIds = IntStream.rangeClosed(1, 5000)
            .boxed()
            .collect(Collectors.toList());

        // --- SEQUENTIAL stream ---
        long sequentialStart = System.currentTimeMillis();

        List<String> sequentialResults = orderIds.stream()      // sequential
            .map(id -> String.format("Order %d: %.2f risk",
                                     id, calculateRiskScore(id)))
            .collect(Collectors.toList());

        long sequentialTime = System.currentTimeMillis() - sequentialStart;
        System.out.println("Sequential: " + sequentialTime + "ms, results: " + sequentialResults.size());

        // --- PARALLEL stream — same pipeline, one word change ---
        long parallelStart = System.currentTimeMillis();

        List<String> parallelResults = orderIds.parallelStream() // parallel!
            .map(id -> String.format("Order %d: %.2f risk",
                                     id, calculateRiskScore(id)))
            // Collectors.toList() is thread-safe — it handles merging internally
            .collect(Collectors.toList());

        long parallelTime = System.currentTimeMillis() - parallelStart;
        System.out.println("Parallel:   " + parallelTime + "ms, results: " + parallelResults.size());

        System.out.printf("%nSpeedup: %.1fx on %d cores%n",
            (double) sequentialTime / parallelTime,
            Runtime.getRuntime().availableProcessors());

        // WRONG — never do this with parallel streams
        // This is a race condition: multiple threads increment the same variable
        // int[] unsafeCount = {0};
        // orderIds.parallelStream().forEach(id -> unsafeCount[0]++); // BUG!
        // Use: orderIds.parallelStream().count() instead
    }
}

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

public class StreamPerformanceTips {

    record Order(String id, double total) {}

    public static void main(String[] args) {

        // Generate 1 million orders for realistic benchmark
        List<Order> orders = IntStream.range(0, 1_000_000)
            .mapToObj(i -> new Order("ORD-" + i, Math.random() * 1000))
            .collect(Collectors.toList());

        // GOOD: filter before map — only maps the 10% that pass
        long start = System.nanoTime();
        List<String> result = orders.stream()
            .filter(order -> order.total() > 900)   // keeps ~10%
            .map(Order::id)                         // transforms only those
            .collect(Collectors.toList());
        long end = System.nanoTime();
        System.out.println("Filter first: " + (end - start) / 1_000_000 + "ms -> " + result.size() + " results");

        // BAD: map first, then filter — maps all 1 million unnecessarily
        start = System.nanoTime();
        result = orders.stream()
            .map(Order::id)                         // transforms every element
            .filter(id -> id.startsWith("ORD-"))   // still 1 million strings created
            .collect(Collectors.toList());
        end = System.nanoTime();
        System.out.println("Map first:  " + (end - start) / 1_000_000 + "ms -> " + result.size() + " results (slower!)");

        // PRIMITIVE STREAM: use IntStream for numeric operations
        // Avoid boxed streams when summing large arrays
        int[] scores = IntStream.range(0, 1_000_000).toArray();
        start = System.nanoTime();
        int sum = IntStream.of(scores).sum();   // no boxing, fast
        end = System.nanoTime();
        System.out.println("IntStream sum: " + (end - start) / 1_000_000 + "ms -> " + sum);

        start = System.nanoTime();
        sum = IntStream.of(scores).boxed().collect(Collectors.summingInt(i -> i)); // boxing overhead
        end = System.nanoTime();
        System.out.println("Boxed sum:    " + (end - start) / 1_000_000 + "ms -> " + sum + " (slower!)");
    }
}