Collectors Deep — toMap, groupingBy, partitioningBy và custom Collector
Đi sâu vào Collectors utility: toMap với merge function, groupingBy lồng nhiều cấp, downstream collectors, partitioningBy, joining, và tự xây Collector.of cho use-case đặc thù.
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
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)"] -.-> DKhi stream sequential: supplier → accumulator × N → finisher. Khi stream parallel: supplier × nhiều thread → accumulator riêng mỗi thread → combiner gộp → finisher.
IDENTITY_FINISH — finisher chỉ là identity function, JVM bỏ qua bước này. UNORDERED — kết quả không cần theo thứ tự, parallel có thể optimize. CONCURRENT — accumulator thread-safe, không cần combiner. Hầu hết Collectors chuẩn không CONCURRENT.
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());
| Collector | Kết quả | Mutability | Java version |
|---|---|---|---|
Collectors.toList() | ArrayList | Mutable | 1.8+ |
Stream.toList() | Unmodifiable | Immutable | 16+ |
Collectors.toUnmodifiableList() | Unmodifiable | Immutable | 10+ |
Collectors.toSet() | HashSet | Mutable | 1.8+ |
Collectors.toUnmodifiableSet() | Unmodifiable | Immutable | 10+ |
Collectors.toList() (Java 8-15) trả ArrayList — bạn có thể add(), remove() sau khi collect. Từ Java 16, Stream.toList() trả unmodifiable và không cho phép null. Nếu code cần immutable output, ưu tiên Stream.toList() hoặc Collectors.toUnmodifiableList() để intent 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.
// 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
Map<String, Integer> map = products.stream()
.collect(Collectors.toMap(
Product::name,
Product::price,
(a, b) -> a // giu gia dau tien
));
// {Apple=100, Banana=80}
Khi key đến từ database, CSV, API — không bao giờ đảm bảo unique 100%. Ngay cả khi database có constraint, data transformation pipeline có thể sinh ra duplicate. Quy tắc: nếu data source không phải List.of(...) hardcoded, hãy luôn cung cấp merge function cho toMap.
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
Hai cách cho kết quả khác nhau. Dùng stream().filter(predicate).collect(groupingBy(...)) loại phần tử trước khi nhóm — group có thể không tồn tại nếu không còn phần tử nào. Dùng groupingBy(key, filtering(predicate, toList())) giữ lại key ngay cả khi không có phần tử thoả điều kiện (group rỗng). Chọn theo intent: "không cần group rỗng" thì filter trước; "cần thấy group rỗng" thì filtering downstream.
7.5 Bảng downstream collectors thông dụng
| Downstream collector | Trả về | Mô tả |
|---|---|---|
counting() | Long | Đếm phần tử trong group |
summingInt(fn) | Integer | Tổng int |
summingLong(fn) | Long | Tổng long |
summingDouble(fn) | Double | Tổng double |
averagingInt(fn) | Double | Trung bình |
mapping(fn, downstream) | Tuỳ downstream | Biến đổi rồi collect |
filtering(pred, downstream) | Tuỳ downstream | Lọc trong group (Java 9+) |
joining(delim) | String | Nối chuỗi |
toList() | List | Mặc định của groupingBy |
toSet() | Set | Loại trùng trong group |
toUnmodifiableList() | List | Immutable list |
minBy(comp) / maxBy(comp) | Optional<T> | Min/max trong group |
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
NullPointerExceptionkhi.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.
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"
Mặc định Collector.of không có IDENTITY_FINISH — finisher luôn được gọi. Nếu finisher của bạn chỉ cast (không transform), thêm Collector.Characteristics.IDENTITY_FINISH làm tham số cuối để tránh bước gọi thừa.
11. Pipeline reduction — mermaid diagram tổng quan
flowchart TD
A[Stream<T>] --> B{Loại collector}
B --> C["toList/toSet\n→ Collection<T>"]
B --> D["toMap(key, value)\n→ Map<K,V>"]
B --> E["groupingBy(classifier)\n→ Map<K, List<T>>"]
B --> F["partitioningBy(pred)\n→ Map<Boolean, List<T>>"]
B --> G["joining(delim)\n→ String"]
B --> H["Collector.of(...)\n→ Custom R"]
E --> I["+ downstream collector\n counting/summing/mapping\n filtering/joining"]
I --> J["Map<K, Long> / Map<K,R>"]12. Deep Dive
CollectorsJavadoc 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ầngroupingBycó ví dụ nested và downstream rất rõ).Collectorinterface 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à
Collectorchư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
forEachvscollect.
13. Self-check
Q1`Collectors.toList()` và `Stream.toList()` có khác nhau không? Khi nào nên dùng cái nào?▸
Q2Vì sao `toMap` không có merge function ném `IllegalStateException` khi có 2 phần tử cùng key? Cơ chế nào bên dưới gây ra điều này?▸
(u, v) -> { throw new IllegalStateException("Duplicate key " + u); }. Vì vậy ngay khi accumulator gặp key đã tồn tại trong map, nó gọi merge function mặc định và throw ngay lập tức. Điều này là thiết kế có chủ ý — thay vì silently ghi đè (như `put`) hoặc silently bỏ qua (như `putIfAbsent`), `toMap` buộc developer phải quyết định policy merge khi build map. Nếu bạn biết data unique, không cần merge function. Nếu data có thể trùng, phải cung cấp merge function rõ ràng — code tự document intent của mình.Q3`groupingBy(Order::customerId)` và `groupingBy(Order::customerId, Collectors.toList())` có cho ra kết quả giống nhau không?▸
Map<K, ArrayList<V>> mutable. Nếu muốn immutable list trong group, phải explicit: `groupingBy(Order::customerId, Collectors.toUnmodifiableList())`.Q4Phân biệt `stream().filter(pred).collect(groupingBy(key))` và `stream().collect(groupingBy(key, filtering(pred, toList())))`. Kết quả có khác nhau không?▸
Q5`partitioningBy` đảm bảo điều gì mà `groupingBy` với boolean classifier không đảm bảo?▸
groupingBy(o -> pred(o)) — nếu không có phần tử thỏa điều kiện, key `true` không tồn tại trong map, `map.get(true)` trả `null` → NullPointerException khi gọi `.size()` hay iterate. Đây là lý do API tách thành 2 collector riêng: `partitioningBy` cho phân chia 2 nhóm guaranteed; `groupingBy` cho phân nhóm mở với N nhóm dynamic.Q6Bạn cần collect stream phần tử vào `LinkedHashMap` (giữ insertion order) nhóm theo category, mỗi group là tổng amount. Viết đúng collector cho use-case này.▸
orders.stream().collect(Collectors.groupingBy(Order::category, LinkedHashMap::new, Collectors.summingInt(Order::amount))) Có 3 tham số: classifier (key), map factory (supplier cho outer map), downstream collector. Form 3 tham số này ít biết hơn form 2 tham số — cần đọc Javadoc để biết `groupingBy` có overload nhận Supplier<M extends Map<K,D>>. Map factory `LinkedHashMap::new` đảm bảo outer map giữ thứ tự phần tử đầu tiên của mỗi group xuất hiện trong stream. Downstream `summingInt` tính tổng trong group.Q7Combiner trong `Collector.of` được gọi khi nào? Với sequential stream, combiner có chạy không?▸
Bài này có giúp bạn hiểu bản chất không?
Hỏi đáp về bài này
Chưa có câu hỏi
Có gì chưa rõ trong bài? Đặt câu hỏi đầu tiên — câu trả lời từ cộng đồng giúp bạn (và người sau).
Đặt câu hỏi đầu tiên