Collectors Deep — toMap, groupingBy, partitioningBy và custom Collector

TL;DR: collect() là terminal operation mạnh nhất của Stream API — nó chuyển stream thành bất kỳ container nào qua cơ chế Collector gồm 4 hàm: supplier (tạo container), accumulator (thêm phần tử), combiner (gộp parallel), finisher (chuyển đổi cuối). Collectors utility class cung cấp sẵn hàng chục collector dựng sẵn: toList, toMap, groupingBy, partitioningBy, joining, cùng các downstream collector để kết hợp thành pipeline phức tạp. Bài này mổ xẻ từng loại, pitfall IllegalStateException khi key trùng trong toMap, và cách tự xây custom Collector khi thư viện chuẩn chưa đủ.

1. Scenario — sales report từ 30 dòng về 3 dòng

Ở bài map/filter/reduce bạn đã gộp stream về một giá trị bằng reduce. Nhưng báo cáo thực tế hiếm khi chỉ cần một con số — nó cần Map, group nhiều tầng, chuỗi nối. Đó là việc của collect.

Hệ thống thương mại điện tử cần tổng hợp báo cáo doanh thu: nhóm đơn hàng theo khách hàng, đếm số đơn mỗi khách, tính tổng tiền. Code imperative mất 30 dòng:

// BEFORE -- imperative accumulator, 30 dong
Map<String, Long> countByCustomer = new HashMap<>();
Map<String, Integer> totalByCustomer = new HashMap<>();

for (Order o : orders) {
    countByCustomer.merge(o.customerId(), 1L, Long::sum);
    totalByCustomer.merge(o.customerId(), o.amount(), Integer::sum);
}

Code Stream + Collectors rút gọn xuống còn 3 dòng mà vẫn đọc được intent rõ hơn:

// AFTER -- Collectors pipeline, 3 dong
Map<String, Long> countByCustomer =
    orders.stream().collect(Collectors.groupingBy(Order::customerId, Collectors.counting()));

Map<String, Integer> totalByCustomer =
    orders.stream().collect(Collectors.groupingBy(Order::customerId, Collectors.summingInt(Order::amount)));

Để viết code này đúng và tự tin, cần hiểu bên dưới Collectors làm gì.

2. Collector<T, A, R> — anatomy

java.util.stream.Collector<T, A, R> (Javadoc SE 21) là interface định nghĩa cách collect() gom phần tử stream vào một container:

• T — kiểu phần tử đầu vào từ stream.
• A — kiểu container trung gian (accumulation type). Thường là mutable container như ArrayList, HashMap.
• R — kiểu kết quả cuối (result type). Có thể khác A sau khi áp finisher.

Interface có 4 method cốt lõi:

public interface Collector<T, A, R> {
    Supplier<A> supplier();           // tao mutable container moi
    BiConsumer<A, T> accumulator();   // them phan tu vao container
    BinaryOperator<A> combiner();     // gop 2 container (parallel streams)
    Function<A, R> finisher();        // bien doi container thanh result cuoi
    Set<Characteristics> characteristics(); // CONCURRENT, UNORDERED, IDENTITY_FINISH
}

Mental model — reduction container pattern:

flowchart LR
    S[Stream elements\nT1, T2, T3...] --> A["supplier()\ncreate container A"]
    A --> B["accumulator()\nadd T1 to A"]
    B --> C["accumulator()\nadd T2 to A"]
    C --> D["accumulator()\nadd T3 to A"]
    D --> E["finisher()\nA → R (result)"]
    E --> F[Result R]
    G["combiner()\nmerge 2 A containers\n(parallel only)"] -.-> D

Khi stream sequential: supplier → accumulator × N → finisher. Khi stream parallel: supplier × nhiều thread → accumulator riêng mỗi thread → combiner gộp → finisher.

3. Basic collectors — toList, toUnmodifiableList, toSet

Ba collector đơn giản nhất, hay dùng nhất:

// toList() -- Java 16+, tra ve unmodifiable List (tren Java 16+ du dung .toList() tren Stream truc tiep)
List<String> names = orders.stream()
    .map(Order::customerId)
    .collect(Collectors.toList());   // tra mutable ArrayList truoc Java 16

// Idiomatic Java 16+: dung .toList() truc tiep thay vi Collectors.toList()
List<String> names2 = orders.stream()
    .map(Order::customerId)
    .toList();   // unmodifiable, null-hostile

// toUnmodifiableList() -- explicit unmodifiable, Java 10+
List<String> immutableNames = orders.stream()
    .map(Order::customerId)
    .collect(Collectors.toUnmodifiableList());

// toSet() -- no guaranteed order, loai trung
Set<String> uniqueCustomers = orders.stream()
    .map(Order::customerId)
    .collect(Collectors.toSet());

Lưu ý: Collectors.toList() (Java 8-15) trả ArrayList mutable — bạn có thể add(), remove() sau khi collect. Từ Java 16, ưu tiên Stream.toList() (unmodifiable, không cho phép null) hoặc Collectors.toUnmodifiableList() khi cần intent immutable rõ ràng.

4. toMap — ánh xạ phần tử thành Map

4.1 Form cơ bản

// toMap(keyMapper, valueMapper)
Map<String, Integer> priceMap = products.stream()
    .collect(Collectors.toMap(
        Product::name,    // key: ten san pham
        Product::price    // value: gia
    ));

4.2 Merge function khi key trùng

Nếu hai phần tử cho ra cùng key, mặc định toMap throw IllegalStateException:

// SAI -- neu co 2 product cung ten, throw:
// IllegalStateException: Duplicate key "Apple" (attempted merging values 100 and 120)
Map<String, Integer> priceMap = products.stream()
    .collect(Collectors.toMap(Product::name, Product::price));

Form 3 tham số cho phép định nghĩa merge function — nhận 2 value trùng key, trả value thắng:

// DUNG -- lay gia cao hon khi trung key
Map<String, Integer> priceMap = products.stream()
    .collect(Collectors.toMap(
        Product::name,
        Product::price,
        (existing, incoming) -> Math.max(existing, incoming)  // merge function
    ));

Merge function phổ biến: (a, b) -> b (incoming ghi đè), (a, b) -> a (giữ existing), Integer::sum (cộng dồn).

4.3 Giữ insertion order — Map supplier tùy chỉnh

toMap form 2-3 tham số trả HashMap — không có thứ tự. Muốn LinkedHashMap giữ insertion order:

// Form 4 tham so: keyMapper, valueMapper, mergeFunction, mapFactory
Map<String, Integer> ordered = products.stream()
    .collect(Collectors.toMap(
        Product::name,
        Product::price,
        (a, b) -> a,              // giu existing khi trung key
        LinkedHashMap::new        // dung LinkedHashMap thay HashMap
    ));

5. Pitfall — toMap và duplicate key

Đây là lỗi runtime phổ biến nhất khi dùng toMap. Code biên dịch sạch nhưng crash ở runtime khi data thực tế có duplicate.

Hãy xem ví dụ dưới đây:

// Scenario: import CSV san pham, co hang trung ten
List<Product> products = List.of(
    new Product("Apple", 100),
    new Product("Banana", 80),
    new Product("Apple", 120)   // duplicate key!
);

// SAI -- throw IllegalStateException
Map<String, Integer> map = products.stream()
    .collect(Collectors.toMap(Product::name, Product::price));
// java.lang.IllegalStateException: Duplicate key Apple (attempted merging values 100 and 120)

// DUNG -- merge function xu ly trung key bằng cách giữ giá trị cũ (oldVal)
Map<String, Integer> map = products.stream()
    .collect(Collectors.toMap(
        Product::name,
        Product::price,
        (oldVal, newVal) -> oldVal   // giu gia dau tien (giu cu)
    ));
// {Apple=100, Banana=80}

6. groupingBy — nhóm phần tử theo key

Collectors.groupingBy(classifier) là collector mạnh nhất trong bộ — nhóm phần tử stream theo giá trị của classifier function, trả Map<K, List<V>>.

6.1 groupingBy đơn giản

record Order(String customerId, String category, int amount) {}

List<Order> orders = List.of(
    new Order("c1", "ELECTRONICS", 500),
    new Order("c2", "CLOTHING", 200),
    new Order("c1", "CLOTHING", 150),
    new Order("c2", "ELECTRONICS", 300)
);

// Nhom theo customer
Map<String, List<Order>> byCustomer = orders.stream()
    .collect(Collectors.groupingBy(Order::customerId));
// {c1=[Order(c1,ELECTRONICS,500), Order(c1,CLOTHING,150)],
//  c2=[Order(c2,CLOTHING,200), Order(c2,ELECTRONICS,300)]}

6.2 groupingBy lồng 2 cấp

Khi cần nhóm theo 2 tiêu chí (ví dụ: customer rồi category):

// Map<customerId, Map<category, List<Order>>>
Map<String, Map<String, List<Order>>> byCustomerThenCategory =
    orders.stream()
        .collect(Collectors.groupingBy(
            Order::customerId,
            Collectors.groupingBy(Order::category)
        ));

// Ket qua:
// {c1={ELECTRONICS=[Order(c1,ELECTRONICS,500)], CLOTHING=[Order(c1,CLOTHING,150)]},
//  c2={CLOTHING=[Order(c2,CLOTHING,200)], ELECTRONICS=[Order(c2,ELECTRONICS,300)]}}

Tham số thứ hai của groupingBy gọi là downstream collector — collector xử lý List phần tử trong mỗi group.

7. Downstream collectors — biến đổi kết quả trong group

Thay vì List<V> mặc định, downstream collector cho phép tổng hợp phức tạp hơn trong mỗi group:

7.1 counting — đếm số phần tử mỗi group

Map<String, Long> orderCountByCustomer = orders.stream()
    .collect(Collectors.groupingBy(
        Order::customerId,
        Collectors.counting()
    ));
// {c1=2, c2=2}

7.2 summingInt / summingLong — tổng số trong group

Map<String, Integer> totalAmountByCustomer = orders.stream()
    .collect(Collectors.groupingBy(
        Order::customerId,
        Collectors.summingInt(Order::amount)
    ));
// {c1=650, c2=500}

7.3 mapping — biến đổi phần tử rồi collect

// Lay danh sach category (khong trung) theo tung customer
Map<String, Set<String>> categoriesByCustomer = orders.stream()
    .collect(Collectors.groupingBy(
        Order::customerId,
        Collectors.mapping(Order::category, Collectors.toSet())
    ));
// {c1={ELECTRONICS, CLOTHING}, c2={CLOTHING, ELECTRONICS}}

7.4 filtering — lọc trong group (Java 9+)

// Chi giu order amount > 200 trong moi group
Map<String, List<Order>> highValueByCustomer = orders.stream()
    .collect(Collectors.groupingBy(
        Order::customerId,
        Collectors.filtering(o -> o.amount() > 200, Collectors.toList())
    ));
// {c1=[Order(c1,ELECTRONICS,500)], c2=[Order(c2,ELECTRONICS,300)]}
// Chu y: c1 va c2 van co entry du chi 1 phan tu -- group van duoc tao

7.5 Bảng downstream collectors thông dụng

7.6 Hiệu năng của groupingBy lồng nhiều tầng (Deep Nested Grouping Performance)

Việc lồng 2-3 tầng groupingBy (ví dụ: groupingBy(A, groupingBy(B, groupingBy(C)))) trông rất gọn gàng và đậm chất functional. Tuy nhiên, đằng sau sự thanh lịch đó là một cái bẫy hiệu năng đáng kể trên Heap:

• Cấu trúc Map-lồng-Map (Nested Map Structure): Việc lồng nhiều tầng tạo ra các cấu trúc Map-lồng-Map phức tạp trên Heap. Mỗi tầng Map lại sinh ra các node trung gian, các bucket array và các đối tượng Entry đi kèm — tốn nhiều tài nguyên Heap.
• Suy giảm tốc độ Lookup: Khi tìm kiếm phần tử, JVM phải băm và truy vết qua nhiều tầng map. Tần suất cache miss của CPU tăng cao, đồng thời Garbage Collector phải quản lý một đồ thị đối tượng chằng chịt, làm giảm thông lượng (throughput).
• Cân nhắc Key phức hợp (Composite Key): Nếu hệ thống yêu cầu thông lượng cao (high throughput), hãy chuyển sang dùng Composite Key — một key phẳng duy nhất đại diện cho tổ hợp các trường nhóm, ví dụ record GroupKey(String customerId, String category) {}. Việc này giúp lưu dữ liệu trong một map phẳng 1 tầng, tiết kiệm bộ nhớ và giữ độ phức tạp lookup gần như O(1).

8. partitioningBy — chia 2 nhóm theo boolean

Collectors.partitioningBy(predicate) là trường hợp đặc biệt của groupingBy với classifier boolean — luôn trả Map<Boolean, List<T>> với đúng 2 key true và false.

// Chia order thanh high-value (>= 300) va low-value
Map<Boolean, List<Order>> partitioned = orders.stream()
    .collect(Collectors.partitioningBy(o -> o.amount() >= 300));

List<Order> highValue = partitioned.get(true);
List<Order> lowValue = partitioned.get(false);

Ưu điểm so với groupingBy(o -> o.amount() >= 300): đảm bảo cả 2 key true và false luôn tồn tại trong map, kể cả khi một partition rỗng. Với groupingBy, nếu không có phần tử nào thỏa điều kiện, key đó không xuất hiện trong map.

Downstream collector cũng dùng được với partitioningBy:

// Dem so order moi partition
Map<Boolean, Long> countPartitioned = orders.stream()
    .collect(Collectors.partitioningBy(
        o -> o.amount() >= 300,
        Collectors.counting()
    ));
// {true=2, false=2}

Khi nào prefer partitioningBy thay groupingBy:

• Tiêu chí phân chia là boolean (pass/fail, active/inactive, premium/regular).
• Cần chắc chắn cả 2 partition luôn có entry trong map (để tránh NullPointerException khi .get(true)).
• Logic downstream khác nhau cho 2 nhóm cần xử lý riêng ngay sau.

9. joining — nối chuỗi idiomatic

Collectors.joining(delimiter, prefix, suffix) tổng hợp phần tử String (hoặc sau map) thành một chuỗi.

List<String> names = List.of("An", "Binh", "Cuong", "Dung");

// Khong delimiter
String plain = names.stream().collect(Collectors.joining());
// "AnBinhCuongDung"

// Co delimiter
String csv = names.stream().collect(Collectors.joining(", "));
// "An, Binh, Cuong, Dung"

// Co prefix + suffix
String bracketed = names.stream().collect(Collectors.joining(", ", "[", "]"));
// "[An, Binh, Cuong, Dung]"

Trong thực tế, joining hay dùng kết hợp với map:

String report = orders.stream()
    .map(o -> o.customerId() + ":" + o.amount())
    .collect(Collectors.joining(", ", "Orders=[", "]"));
// "Orders=[c1:500, c2:200, c1:150, c2:300]"

Đây là cách idiomatic Java hơn String.join(delimiter, list) vì tích hợp ngay trong pipeline, không cần intermediate List<String>.

10. Custom Collector với Collector.of

Khi không có collector chuẩn nào phù hợp, dùng Collector.of(supplier, accumulator, combiner, finisher, characteristics) để xây từ đầu.

Luồng hoạt động trong Parallel Stream (Accumulator & Combiner)

Để hiểu rõ cách viết Custom Collector hoạt động an toàn và tối ưu, đặc biệt là trong môi trường song song (parallel stream), hãy quan sát sơ đồ trực quan dưới đây:

flowchart TD
    subgraph Thread 1 [Luồng 1]
        S1["supplier() → Container A1"] --> Ac1["accumulator(A1, T1)"]
        Ac1 --> Ac2["accumulator(A1, T2)"]
    end

    subgraph Thread 2 [Luồng 2]
        S2["supplier() → Container A2"] --> Ac3["accumulator(A2, T3)"]
        Ac3 --> Ac4["accumulator(A2, T4)"]
    end

    Ac2 --> Comb["combiner(A1, A2) → Merged Container A"]
    Ac4 --> Comb
    Comb --> Fin["finisher(A) → Result R"]
    Fin --> Out["Kết quả cuối R"]

Giải nghĩa các hàm thành phần:

• supplier(): Khởi tạo container tích lũy trung gian rỗng (ví dụ ArrayList::new hoặc HashMap::new). Khi chạy parallel, mỗi thread độc lập sẽ được cấp một container riêng bằng cách gọi hàm supplier này.
• accumulator(): Nhận vào container tích lũy và một phần tử từ stream, thực hiện đưa phần tử đó vào container (ví dụ List::add).
• combiner(): Nhận vào hai container tích lũy được xử lý song song từ hai thread khác nhau và hợp nhất chúng thành một container duy nhất. Đây là bước sống còn để đảm bảo parallel stream trả về kết quả đúng đắn.
• finisher(): Nhận container tích lũy cuối cùng sau khi đã accumulator và combiner xong, chuyển đổi nó sang kiểu dữ liệu kết quả mong muốn của người dùng (ví dụ: bọc trong unmodifiable collection hoặc định dạng thành String).

Vai trò của Characteristics trong Collector

Characteristics là một tập hợp các cờ (flags) đặc tính giúp tối ưu hóa hiệu năng bằng cách báo cho Stream API biết hành vi nội bộ của Collector:

1. IDENTITY_FINISH:
   • Báo hiệu rằng bước finisher() chỉ là một hàm đồng nhất (identity function) a -> a — tức là kiểu container trung gian A và kiểu kết quả cuối R hoàn toàn trùng khớp và không cần bất kỳ bước biến đổi hay cast kiểu nào.
   • Tối ưu: Khi có cờ này, Stream API sẽ bỏ qua hoàn toàn việc gọi finisher, giúp cải thiện tốc độ và tiết kiệm overhead cuộc gọi phương thức.
2. UNORDERED:
   • Báo hiệu rằng collector không quan tâm đến thứ tự của dữ liệu đầu vào.
   • Tối ưu: Rất có lợi cho parallel stream vì JVM có thể chia nhỏ và gộp dữ liệu mà không cần giữ gìn thứ tự gốc, nâng cao đáng kể thông lượng.
3. CONCURRENT:
   • Báo hiệu rằng accumulator là thread-safe và container có thể được chỉnh sửa đồng thời từ nhiều luồng mà không cần dùng đến bước combiner().
   • Tối ưu: Stream API sẽ chỉ tạo duy nhất một container dùng chung và tất cả các thread song song cùng ghi trực tiếp vào đó, triệt tiêu chi phí merger.

Ví dụ: collect vào bounded buffer — lấy tối đa N phần tử

// Custom collector: lay toi da maxSize phan tu dau tien
public static <T> Collector<T, ?, List<T>> toBoundedList(int maxSize) {
    return Collector.of(
        ArrayList::new,                     // supplier: tao ArrayList trong
        (list, element) -> {                // accumulator: them neu chua day
            if (list.size() < maxSize) list.add(element);
        },
        (left, right) -> {                  // combiner: gop 2 partial list (parallel)
            List<T> merged = new ArrayList<>(left);
            for (T e : right) {
                if (merged.size() >= maxSize) break;
                merged.add(e);
            }
            return merged;
        }
        // finisher: khong can -- ArrayList la result cuoi (IDENTITY_FINISH implicit)
    );
}

// Su dung:
List<Order> top3 = orders.stream()
    .sorted(Comparator.comparingInt(Order::amount).reversed())
    .collect(toBoundedList(3));

Ví dụ 2: collect thành String với separator và count

// Collector tao chuoi "count: phan_tu_1, phan_tu_2, ..."
Collector<String, ?, String> withCount = Collector.of(
    ArrayList::new,
    List::add,
    (a, b) -> { a.addAll(b); return a; },
    list -> list.size() + ": " + String.join(", ", list)  // finisher transform
);

String result = Stream.of("alpha", "beta", "gamma").collect(withCount);
// "3: alpha, beta, gamma"

11. Deep Dive

• Collectors Javadoc SE 21 — docs.oracle.com/.../Collectors.html — danh sách đầy đủ tất cả collector chuẩn kèm example code trong Javadoc (đặc biệt phần groupingBy có ví dụ nested và downstream rất rõ).
• Collector interface Javadoc — docs.oracle.com/.../Collector.html — giải thích 4 function + Characteristics + cơ chế khi parallel.
• JEP 461 — Stream Gatherers (Preview Java 22+) — openjdk.org/jeps/461 — API mới mở rộng khả năng viết intermediate operation tùy chỉnh (hiện là Preview), bổ sung chỗ mà Collector chưa bao phủ (intermediate gathering, not just terminal). Đây là hướng tương lai của tùy biến Stream pipeline trong Java.
• Effective Java item 46 (Bloch, 3rd ed) — "Prefer side-effect-free functions in streams" — giải thích vì sao side-effect trong collector accumulator là nguy hiểm, và pattern đúng khi dùng forEach vs collect.

12. Self-check

Map<String, Map<Integer, Long>> result = users.stream()
  .collect(Collectors.groupingBy(
      User::department,
      Collectors.groupingBy(
          User::age,
          Collectors.counting())));

Bài tiếp theo: Parallel Stream — khi nào thực sự nhanh hơn, khi nào nguy hiểm