Collectors Deep Dive: toMap, groupingBy và Custom Collector

TL;DR: Thao tác collect(Collector) là trạm kết thúc (Terminal Operation) mạnh mẽ và linh hoạt nhất của Stream API, cho phép chuyển đổi luồng dữ liệu thành bất kỳ cấu trúc dữ liệu mong muốn nào. Trái tim của cơ chế này là interface Collector<T, A, R> gồm 4 phương thức: supplier, accumulator, combiner (cho luồng song song), và finisher. Lớp tiện ích Collectors cung cấp sẵn các bộ thu thập mạnh mẽ như toMap (cần merge function để tránh IllegalStateException), groupingBy kết hợp Downstream Collector phân nhóm dữ liệu nhiều tầng, và partitioningBy chia tách dữ liệu theo boolean. Khi các tiện ích sẵn có chưa đáp ứng được bài toán, ta có thể tự xây dựng Collector.of tùy biến.

Sau khi đã nắm vững cơ chế pipeline tại Stream Basics và các phép toán cơ bản tại Map, Filter và Reduce, bài học này sẽ giải quyết bài toán phức tạp nhất: tổng hợp dữ liệu thành các cấu trúc Map, Set và báo cáo phân tầng.

1. Interface Collector vận hành gồm những thành phần nào?

Giao diện Collector<T, A, R> quy định cách thức một luồng dữ liệu được gom vào một vùng chứa có thể biến đổi (Mutable Container):

public interface Collector<T, A, R> {
    Supplier<A> supplier();                  // Creates new empty mutable container
    BiConsumer<A, T> accumulator();          // Folds stream element into container
    BinaryOperator<A> combiner();            // Merges 2 containers in parallel execution
    Function<A, R> finisher();               // Transforms container to final result
    Set<Characteristics> characteristics();  // Optimization flags: IDENTITY_FINISH, UNORDERED, CONCURRENT
}

Ở chế độ chạy tuần tự (Sequential Stream), chu trình chỉ gồm: supplier -> gọi accumulator cho từng phần tử -> finisher. Phương thức combiner hoàn toàn không được gọi. Tuy nhiên, nếu bạn viết sai logic của combiner, lỗi ngầm sẽ lập tức phát tác ngay khi luồng được chuyển sang chế độ song song (parallelStream).

2. toMap xử lý trùng khóa (Duplicate Key) và thứ tự chèn ra sao?

Khi chuyển đổi một danh sách ấn phẩm thành một Map, lỗi phổ biến nhất là trùng khóa (ví dụ hai cuốn sách có cùng mã ISBN):

record Book(String isbn, String title, double price) { }

List<Book> books = List.of(
    new Book("ISBN-01", "Clean Code", 45.0),
    new Book("ISBN-02", "Effective Java", 50.0),
    new Book("ISBN-01", "Clean Code 2nd", 55.0) // Duplicate ISBN-01!
);

// 1. Bug: 2-arg toMap throws IllegalStateException on duplicates
// Map<String, Book> map = books.stream().collect(Collectors.toMap(Book::isbn, b -> b));

// 2. Idiomatic: 3-arg toMap provides explicit merge function
Map<String, Book> bookMap = books.stream().collect(Collectors.toMap(
    Book::isbn,
    b -> b,
    (existing, incoming) -> incoming // Keep latest version
));

// 3. Preserve insertion order: 4-arg toMap with LinkedHashMap
Map<String, Book> orderedMap = books.stream().collect(Collectors.toMap(
    Book::isbn,
    b -> b,
    (oldVal, newVal) -> newVal,
    LinkedHashMap::new
));

[!WARNING] Không bao giờ dùng toMap 2 tham số với dữ liệu lấy từ cơ sở dữ liệu hoặc file CSV bên ngoài. Luôn cung cấp Merge Function tường minh để hệ thống không bị sập bất ngờ tại runtime.

3. groupingBy và Downstream Collector phân nhóm dữ liệu nhiều tầng như thế nào?

Phương thức Collectors.groupingBy(classifier) phân loại các phần tử theo một khóa định danh và gom các phần tử cùng khóa vào một danh sách:

record LoanRecord(String memberId, String category, double fine) { }

// 1. Basic grouping: Map<String, List<LoanRecord>>
Map<String, List<LoanRecord>> byMember = loans.stream()
    .collect(Collectors.groupingBy(LoanRecord::memberId));

// 2. Downstream Collector: count loans per member
Map<String, Long> loanCountByMember = loans.stream()
    .collect(Collectors.groupingBy(
        LoanRecord::memberId,
        Collectors.counting() // Downstream Collector
    ));

// 3. Sum fines per member:
Map<String, Double> fineByMember = loans.stream()
    .collect(Collectors.groupingBy(
        LoanRecord::memberId,
        Collectors.summingDouble(LoanRecord::fine)
    ));

groupingBy lồng 2 cấp và lưu ý hiệu năng

Bạn có thể lồng groupingBy bên trong một groupingBy khác để tạo báo cáo 2 chiều (ví dụ: nhóm theo độc giả, sau đó nhóm tiếp theo thể loại sách):

Map<String, Map<String, List<LoanRecord>>> nestedReport = loans.stream()
    .collect(Collectors.groupingBy(
        LoanRecord::memberId,
        Collectors.groupingBy(LoanRecord::category)
    ));

Cảnh báo hiệu năng: Việc lồng quá 2 tầng tạo ra cấu trúc Map lồng Map rất tốn bộ nhớ Heap và tăng tỷ lệ CPU Cache Miss. Trong các hệ thống chịu tải cao, hãy thay thế bằng một Composite Key phẳng (ví dụ: record GroupKey(String memberId, String category) {}).

4. partitioningBy chia đôi luồng dữ liệu theo mệnh đề boolean ra sao?

partitioningBy(Predicate) là trường hợp chuyên biệt của groupingBy khi tiêu chí phân loại là một biểu thức logic (đúng/sai):

// Separate overdue and on-time loan records:
Map<Boolean, List<LoanRecord>> partitioned = loans.stream()
    .collect(Collectors.partitioningBy(r -> r.fine() > 0));

List<LoanRecord> overdueLoans = partitioned.get(true);  // Overdue list
List<LoanRecord> onTimeLoans = partitioned.get(false);  // On-time list

Ưu điểm vượt trội: partitioningBy bảo đảm kết quả trả về luôn luôn chứa cả 2 khóa true và false, ngay cả khi một trong hai nhóm hoàn toàn không có phần tử nào. Điều này loại bỏ hoàn toàn nguy cơ bị lỗi NullPointerException khi gọi .get(true).

5. Khi nào và làm thế nào để tự xây dựng một Custom Collector?

Khi các collector có sẵn của JDK chưa đáp ứng đúng nghiệp vụ, bạn có thể tự tạo bộ thu gom riêng thông qua Collector.of():

// Custom collector: collects elements into formatted summary string
Collector<String, ?, String> withCountCollector = Collector.of(
    ArrayList::new,                                        // 1. supplier
    List::add,                                             // 2. accumulator
    (left, right) -> { left.addAll(right); return left; }, // 3. combiner (parallel)
    list -> list.size() + " items: " + String.join(", ", list) // 4. finisher
);

String report = Stream.of("Book A", "Book B", "Book C").collect(withCountCollector);
// Output: "3 items: Book A, Book B, Book C"

Tối ưu với Characteristics

Nếu kiểu container trung gian A trùng khớp với kiểu kết quả cuối R, bạn không cần bước finisher. Hãy gắn thêm cờ Collector.Characteristics.IDENTITY_FINISH để JVM bỏ qua bước chuyển đổi này nhằm đạt tốc độ tối đa.

6. Đâu là những cạm bẫy phổ biến khi làm việc với Collectors?

❌ Cạm bẫy 1: Không có merge function khi dữ liệu có thể duplicate

// Crash ung dung khi data trung lap ma:
items.stream().collect(Collectors.toMap(LibraryItem::getBarcode, LibraryItem::getTitle));

❌ Cạm bẫy 2: Nhầm lẫn giữa Stream.toList() và Collectors.toList()

• Collectors.toList() (Java 8+) trả về một ArrayList có thể sửa đổi (mutable).
• Stream.toList() (Java 16+) trả về một danh sách bất biến (unmodifiable, không cho phép null), chạy nhanh hơn và tiết kiệm bộ nhớ hơn.

❌ Cạm bẫy 3: Viết combiner bị lỗi trong Custom Collector Nếu combiner làm thay đổi cả hai danh sách hoặc không trả về danh sách đã gộp hợp lệ, chương trình sẽ cho kết quả sai lệch khi kích hoạt .parallelStream().

7. Tóm tắt các điểm cốt lõi

• Collector<T, A, R> gồm 4 phương thức chính; combiner chỉ chạy khi kích hoạt luồng song song.
• toMap bắt buộc phải có Merge Function khi xử lý dữ liệu từ nguồn ngoài để tránh lỗi IllegalStateException.
• groupingBy kết hợp Downstream Collectors giúp tạo báo cáo phân nhóm dữ liệu nhiều tầng linh hoạt.
• partitioningBy phân chia dữ liệu theo boolean và bảo đảm cả 2 khóa true/false luôn tồn tại trong Map.

8. Câu hỏi tự kiểm tra (Self-Check)

Bài tiếp theo: Parallel Stream: Khi Nào Nên Dùng và Cạm Bẫy Luồng Phổ Biến