Collectors & Parallel Streams

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

List<String> names = List.of("Alice", "Bob", "Charlie", "Alice");

// toList() — mutable ArrayList (pre-Java 16)
List<String> list = names.stream()
    .collect(Collectors.toList());  // [Alice, Bob, Charlie, Alice]

// toList() — immutable (Java 16+)
List<String> immutable = names.stream().toList();

// toUnmodifiableList() — immutable, null-hostile (Java 10+)
List<String> unmod = names.stream()
    .collect(Collectors.toUnmodifiableList());

// toSet() — removes duplicates
Set<String> set = names.stream()
    .collect(Collectors.toSet());  // [Alice, Bob, Charlie]

// toUnmodifiableSet() — immutable set (Java 10+)
Set<String> unmodSet = names.stream()
    .collect(Collectors.toUnmodifiableSet());

// toCollection() — choose your own collection type
TreeSet<String> sorted = names.stream()
    .collect(Collectors.toCollection(TreeSet::new));
// [Alice, Bob, Charlie] — sorted + deduplicated

LinkedList<String> linked = names.stream()
    .collect(Collectors.toCollection(LinkedList::new));

record Employee(int id, String name, String dept, int salary) {}

List<Employee> employees = List.of(
    new Employee(1, "Alice", "Engineering", 120000),
    new Employee(2, "Bob", "Engineering", 110000),
    new Employee(3, "Charlie", "Marketing", 95000)
);

// Basic toMap — key extractor + value extractor
Map<Integer, String> idToName = employees.stream()
    .collect(Collectors.toMap(
        Employee::id,       // key extractor
        Employee::name      // value extractor
    ));
// {1=Alice, 2=Bob, 3=Charlie}

// toMap with identity value
Map<Integer, Employee> idToEmployee = employees.stream()
    .collect(Collectors.toMap(Employee::id, Function.identity()));

// THIS THROWS IllegalStateException if duplicate keys exist!
Map<String, Integer> deptToSalary = employees.stream()
    .collect(Collectors.toMap(
        Employee::dept,     // "Engineering" appears twice!
        Employee::salary
    ));
// Exception: Duplicate key Engineering

// FIX: Provide merge function (third parameter)
Map<String, Integer> deptToMaxSalary = employees.stream()
    .collect(Collectors.toMap(
        Employee::dept,
        Employee::salary,
        Integer::max        // merge function: keep higher salary
    ));
// {Engineering=120000, Marketing=95000}

// With specific map type (fourth parameter)
TreeMap<String, Integer> sortedMap = employees.stream()
    .collect(Collectors.toMap(
        Employee::dept,
        Employee::salary,
        Integer::sum,           // merge: sum salaries
        TreeMap::new            // map supplier
    ));

// Group employees by department
Map<String, List<Employee>> byDept = employees.stream()
    .collect(Collectors.groupingBy(Employee::dept));
// {Engineering=[Alice, Bob], Marketing=[Charlie]}

// Count per group
Map<String, Long> countByDept = employees.stream()
    .collect(Collectors.groupingBy(Employee::dept, Collectors.counting()));
// {Engineering=2, Marketing=1}

// Sum salaries per department
Map<String, Integer> salaryByDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept, 
        Collectors.summingInt(Employee::salary)
    ));
// {Engineering=230000, Marketing=95000}

// Average salary per department
Map<String, Double> avgByDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.averagingInt(Employee::salary)
    ));

// Collect names per department
Map<String, List<String>> namesByDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.mapping(Employee::name, Collectors.toList())
    ));
// {Engineering=[Alice, Bob], Marketing=[Charlie]}

// Join names per department
Map<String, String> joinedNames = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.mapping(Employee::name, Collectors.joining(", "))
    ));
// {Engineering=Alice, Bob, Marketing=Charlie}

// Max salary employee per department
Map<String, Optional<Employee>> topEarner = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.maxBy(Comparator.comparingInt(Employee::salary))
    ));

record Employee(int id, String name, String dept, String city, int salary) {}

// Two levels: department → city → list of employees
Map<String, Map<String, List<Employee>>> nested = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.groupingBy(Employee::city)
    ));

// Three levels: department → city → count
Map<String, Map<String, Long>> nestedCount = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.groupingBy(Employee::city, Collectors.counting())
    ));

// Use TreeMap for sorted keys
TreeMap<String, List<Employee>> sorted = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        TreeMap::new,                    // map factory (second param)
        Collectors.toList()              // downstream (third param)
    ));

// Partition into high earners vs others
Map<Boolean, List<Employee>> partitioned = employees.stream()
    .collect(Collectors.partitioningBy(e -> e.salary() > 100000));
// {true=[Alice, Bob], false=[Charlie]}

// With downstream collector
Map<Boolean, Long> countPartition = employees.stream()
    .collect(Collectors.partitioningBy(
        e -> e.salary() > 100000,
        Collectors.counting()
    ));
// {true=2, false=1}

// Partition + collect names
Map<Boolean, List<String>> namePartition = employees.stream()
    .collect(Collectors.partitioningBy(
        e -> e.salary() > 100000,
        Collectors.mapping(Employee::name, Collectors.toList())
    ));
// {true=[Alice, Bob], false=[Charlie]}

// reducing() — like Stream.reduce() but as a downstream collector
Optional<Employee> highestPaid = employees.stream()
    .collect(Collectors.reducing(
        BinaryOperator.maxBy(Comparator.comparingInt(Employee::salary))
    ));

// reducing with identity
int totalSalary = employees.stream()
    .collect(Collectors.reducing(0, Employee::salary, Integer::sum));

// summarizingInt — get count, sum, min, avg, max in one pass
IntSummaryStatistics stats = employees.stream()
    .collect(Collectors.summarizingInt(Employee::salary));
// count=3, sum=325000, min=95000, average=108333.33, max=120000

System.out.println(stats.getMax());     // 120000
System.out.println(stats.getAverage()); // 108333.33
System.out.println(stats.getCount());   // 3

List<String> items = List.of("Java", "Kotlin", "Scala");

// Simple concatenation
String joined = items.stream()
    .collect(Collectors.joining());
// "JavaKotlinScala"

// With delimiter
String csv = items.stream()
    .collect(Collectors.joining(", "));
// "Java, Kotlin, Scala"

// With delimiter, prefix, suffix
String json = items.stream()
    .collect(Collectors.joining("\", \"", "[\"", "\"]"));
// ["Java", "Kotlin", "Scala"]

// Common pattern: building SQL IN clause
String inClause = ids.stream()
    .map(String::valueOf)
    .collect(Collectors.joining(", ", "WHERE id IN (", ")"));
// WHERE id IN (1, 2, 3, 4)

// Collect to list, then make unmodifiable
List<String> unmodifiable = employees.stream()
    .map(Employee::name)
    .collect(Collectors.collectingAndThen(
        Collectors.toList(),
        Collections::unmodifiableList
    ));

// Collect and get size
int size = employees.stream()
    .collect(Collectors.collectingAndThen(
        Collectors.toList(),
        List::size
    ));

// Collect to set, then check if singleton
boolean isSingleton = employees.stream()
    .map(Employee::dept)
    .collect(Collectors.collectingAndThen(
        Collectors.toSet(),
        set -> set.size() == 1
    ));

// GroupingBy + collectingAndThen for non-Optional maxBy
Map<String, Employee> topEarnerPerDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.collectingAndThen(
            Collectors.maxBy(Comparator.comparingInt(Employee::salary)),
            Optional::orElseThrow
        )
    ));

// Find min and max in one pass
record Range(int min, int max) {}

Range salaryRange = employees.stream()
    .collect(Collectors.teeing(
        Collectors.minBy(Comparator.comparingInt(Employee::salary)),
        Collectors.maxBy(Comparator.comparingInt(Employee::salary)),
        (min, max) -> new Range(
            min.orElseThrow().salary(),
            max.orElseThrow().salary()
        )
    ));
// Range[min=95000, max=120000]

// Count and sum simultaneously
record CountAndSum(long count, int sum) {}

CountAndSum result = employees.stream()
    .collect(Collectors.teeing(
        Collectors.counting(),
        Collectors.summingInt(Employee::salary),
        CountAndSum::new
    ));
// CountAndSum[count=3, sum=325000]

// Partition into pass/fail with counts
record TestResult(long passed, long failed) {}

TestResult scores = students.stream()
    .collect(Collectors.teeing(
        Collectors.filtering(s -> s.score() >= 60, Collectors.counting()),
        Collectors.filtering(s -> s.score() < 60, Collectors.counting()),
        TestResult::new
    ));

record Order(String customer, List<String> items) {}

List<Order> orders = List.of(
    new Order("Alice", List.of("Laptop", "Mouse")),
    new Order("Alice", List.of("Keyboard")),
    new Order("Bob", List.of("Monitor", "Laptop"))
);

// flatMapping — flatten nested collections per group
Map<String, Set<String>> itemsByCustomer = orders.stream()
    .collect(Collectors.groupingBy(
        Order::customer,
        Collectors.flatMapping(
            order -> order.items().stream(),
            Collectors.toSet()
        )
    ));
// {Alice=[Laptop, Mouse, Keyboard], Bob=[Monitor, Laptop]}

// filtering — filter within groups (not before grouping!)
Map<String, List<Employee>> highEarnersByDept = employees.stream()
    .collect(Collectors.groupingBy(
        Employee::dept,
        Collectors.filtering(
            e -> e.salary() > 100000,
            Collectors.toList()
        )
    ));
// Marketing key exists with empty list: {Engineering=[Alice, Bob], Marketing=[]}
// vs filter().groupingBy() which would OMIT the Marketing key entirely

public interface Collector<T, A, R> {
    Supplier<A> supplier();          // Create accumulator container
    BiConsumer<A, T> accumulator();  // Add element to container
    BinaryOperator<A> combiner();   // Merge two containers (parallel)
    Function<A, R> finisher();       // Final transformation
    Set<Characteristics> characteristics(); // Optimization hints
}

public class TopNCollector<T> implements Collector<T, PriorityQueue<T>, List<T>> {

    private final int n;
    private final Comparator<T> comparator;

    public TopNCollector(int n, Comparator<T> comparator) {
        this.n = n;
        this.comparator = comparator;
    }

    @Override
    public Supplier<PriorityQueue<T>> supplier() {
        return () -> new PriorityQueue<>(comparator);
    }

    @Override
    public BiConsumer<PriorityQueue<T>, T> accumulator() {
        return (queue, element) -> {
            queue.offer(element);
            if (queue.size() > n) {
                queue.poll();  // Remove lowest when over capacity
            }
        };
    }

    @Override
    public BinaryOperator<PriorityQueue<T>> combiner() {
        return (q1, q2) -> {
            q1.addAll(q2);
            while (q1.size() > n) q1.poll();
            return q1;
        };
    }

    @Override
    public Function<PriorityQueue<T>, List<T>> finisher() {
        return queue -> {
            List<T> result = new ArrayList<>(queue);
            result.sort(comparator.reversed());
            return Collections.unmodifiableList(result);
        };
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Set.of();  // Not IDENTITY_FINISH, not UNORDERED
    }

    // Factory method for convenience
    public static <T> TopNCollector<T> topN(int n, Comparator<T> comparator) {
        return new TopNCollector<>(n, comparator);
    }
}

// Usage
List<Employee> top3Earners = employees.stream()
    .collect(TopNCollector.topN(3, Comparator.comparingInt(Employee::salary)));

// Example: Characteristics in practice
@Override
public Set<Characteristics> characteristics() {
    return Set.of(
        Characteristics.UNORDERED,       // Result order doesn't matter
        Characteristics.IDENTITY_FINISH  // No finisher needed (A == R)
    );
}

// Creating parallel streams
List<Integer> numbers = IntStream.rangeClosed(1, 1_000_000)
    .boxed().toList();

// Method 1: parallelStream() from collection
long sum = numbers.parallelStream()
    .mapToLong(Integer::longValue)
    .sum();

// Method 2: .parallel() on existing stream
long sum2 = numbers.stream()
    .parallel()
    .mapToLong(Integer::longValue)
    .sum();

// Default parallelism level
int parallelism = ForkJoinPool.commonPool().getParallelism();
// Typically Runtime.getRuntime().availableProcessors() - 1

// PROBLEM: parallel streams share ForkJoinPool.commonPool()
// If one stream blocks, ALL parallel streams in the JVM are affected

// SOLUTION: Submit to a custom ForkJoinPool
ForkJoinPool customPool = new ForkJoinPool(8);  // 8 threads

try {
    List<Result> results = customPool.submit(() ->
        largeList.parallelStream()
            .map(this::cpuIntensiveTransform)
            .collect(Collectors.toList())
    ).get();  // .get() blocks until complete
} finally {
    customPool.shutdown();
}

// Java 21+ with virtual threads for I/O-bound work:
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<Result>> futures = items.stream()
        .map(item -> executor.submit(() -> fetchFromApi(item)))
        .toList();

    List<Result> results = futures.stream()
        .map(f -> { try { return f.get(); } catch (Exception e) { throw new RuntimeException(e); }})
        .toList();
}

// WRONG: forEach with shared mutable collection
List<Integer> results = new ArrayList<>();  // NOT thread-safe!
IntStream.rangeClosed(1, 10000)
    .parallel()
    .forEach(results::add);  // Race condition! ArrayIndexOutOfBoundsException or lost elements
System.out.println(results.size());  // Often != 10000

// CORRECT: Use collect() — it handles thread-safety internally
List<Integer> results = IntStream.rangeClosed(1, 10000)
    .parallel()
    .boxed()
    .collect(Collectors.toList());  // Thread-safe accumulation
System.out.println(results.size());  // Always 10000

// WRONG: subtraction is NOT associative
// (a - b) - c != a - (b - c)
int wrong = List.of(1, 2, 3, 4, 5).parallelStream()
    .reduce(0, (a, b) -> a - b);
// Sequential: ((((0-1)-2)-3)-4)-5 = -15
// Parallel: unpredictable results due to different split orderings!

// CORRECT: only use associative operations
int correct = List.of(1, 2, 3, 4, 5).parallelStream()
    .reduce(0, Integer::sum);  // Addition IS associative

// findFirst() on parallel — more expensive (must track order)
Optional<Integer> first = numbers.parallelStream()
    .filter(n -> n > 100)
    .findFirst();  // Must synchronize to guarantee order

// findAny() on parallel — faster (no ordering constraint)
Optional<Integer> any = numbers.parallelStream()
    .filter(n -> n > 100)
    .findAny();  // Returns whichever thread finds one first

// forEachOrdered() — preserves order but kills parallelism benefit
numbers.parallelStream()
    .map(this::transform)
    .forEachOrdered(System.out::println);  // Ordered, but sequential bottleneck

// reduce — creates new String each iteration (O(n^2) for strings!)
String concatenated = words.stream()
    .reduce("", (a, b) -> a + b);  // Slow! Creates n intermediate Strings

// collect — mutates StringBuilder in-place (O(n))
String concatenated = words.stream()
    .collect(StringBuilder::new, StringBuilder::append, StringBuilder::append)
    .toString();

// Or simply:
String concatenated = words.stream()
    .collect(Collectors.joining());