Stream Nâng Cao: flatMap, takeWhile, dropWhile và Stateful Operations

TL;DR: Sau các nguyên lý cơ bản từ Stream Basics và bộ ba map, filter, reduce, 20% các tác vụ phức tạp còn lại trong Stream API được xử lý bằng các công cụ chuyên dụng: flatMap giúp phẳng hóa các cấu trúc lồng nhau (mỗi phần tử cha sinh ra một Stream con gồm 0..N phần tử) và mapMulti (Java 16+) để triệt tiêu chi phí cấp phát Stream con. Bộ đôi takeWhile/dropWhile (Java 9+) cho phép cắt ngắn chuỗi dữ liệu dựa trên tiền tố của một Ordered Stream. Cuối cùng, nhóm thao tác có trạng thái (Stateful Operations như sorted, distinct) đòi hỏi phải gom toàn bộ dữ liệu vào bộ đệm trước khi nhả kết quả, tuyệt đối không được đặt trước các nguồn dữ liệu vô hạn.

1. flatMap phẳng hóa cấu trúc dữ liệu lồng nhau như thế nào?

Trong mô hình quản lý thư viện, một tác giả (Author) có thể viết nhiều đầu sách (List<Book>). Nếu bạn muốn lấy danh sách tất cả các cuốn sách của một nhóm tác giả:

record Author(String name, List<String> bookTitles) { }

List<Author> authors = List.of(
    new Author("Joshua Bloch", List.of("Effective Java")),
    new Author("Robert Martin", List.of("Clean Code", "Clean Architecture"))
);

// 1. Neu dung map(): Tra ve danh sach long nhau List<List<String>>
List<List<String>> nested = authors.stream()
    .map(Author::bookTitles)
    .toList(); 

// 2. Dung flatMap(): Phang hoa thanh List<String>
List<String> flatTitles = authors.stream()
    .flatMap(a -> a.bookTitles().stream())
    .toList();
// Ket qua: ["Effective Java", "Clean Code", "Clean Architecture"]

2. Khi nào nên dùng mapMulti thay cho flatMap để tối ưu bộ nhớ?

flatMap nhận vào một hàm trả về Stream<R>, đồng nghĩa với việc với mỗi phần tử cha, JVM phải khởi tạo một đối tượng Stream con và một Spliterator tạm thời trên Heap. Khi xử lý hàng triệu phần tử cha, lượng object ngắn hạn này sẽ gây áp lực dọn rác (GC Young Gen).

Từ Java 16, phương thức mapMulti ra đời để giải quyết bài toán hiệu năng này bằng cách sử dụng Consumer:

// mapMulti (Java 16+): Khong khoi tao bat ky Stream con nao tren Heap
List<String> flatTitlesFast = authors.stream()
    .<String>mapMulti((author, consumer) -> {
        for (String title : author.bookTitles()) {
            consumer.accept(title); // Day truc tiep phan tu vao pipeline
        }
    })
    .toList();

[!TIP] Hãy tiếp tục sử dụng flatMap cho code thông thường vì cú pháp ngắn gọn và chuẩn mực. Chỉ cân nhắc đổi sang mapMulti khi công cụ Profiler chỉ ra flatMap là điểm nghẽn bộ nhớ trong các tác vụ xử lý hàng triệu bản ghi.

3. takeWhile và dropWhile khác biệt gì so với filter?

Được giới thiệu từ Java 9, bộ đôi này chuyên xử lý các chuỗi dữ liệu có cấu trúc tiền tố (prefix) hoặc đã được sắp xếp:

List<Integer> checkoutDays = List.of(2, 5, 8, 12, 3, 1);

// 1. filter: Duyet toan bo stream, kiem tra tung phan tu doc lap
List<Integer> filtered = checkoutDays.stream()
    .filter(d -> d < 10)
    .toList(); // [2, 5, 8, 3, 1]

// 2. takeWhile: Lay cac phan tu dau cho den khi gap phan tu dau tien >= 10 thi DUNG
List<Integer> taken = checkoutDays.stream()
    .takeWhile(d -> d < 10)
    .toList(); // [2, 5, 8] - Dung ngay o so 12, bo qua [3, 1]

// 3. dropWhile: Bo qua cac phan tu dau thoa man, lay toan bo phan con lai
List<Integer> dropped = checkoutDays.stream()
    .dropWhile(d -> d < 10)
    .toList(); // [12, 3, 1] - Bat dau lay tu so 12 den het

[!WARNING] takeWhile và dropWhile chỉ cho kết quả nhất quán (deterministic) trên Ordered Stream (như ArrayList hay stream đã sort). Nếu chạy trên một HashSet hoặc qua parallelStream(), thứ tự duyệt không được bảo đảm, dẫn đến việc ngắt stream tại các vị trí ngẫu nhiên.

4. Các thao tác có trạng thái (Stateful Operations) vận hành ra sao?

Hầu hết các thao tác như filter, map là Stateless (không lưu trạng thái, phần tử đi qua độc lập). Tuy nhiên, có một nhóm thao tác đặc biệt đòi hỏi phải duy trì bộ nhớ:

1. sorted() / sorted(Comparator): Bắt buộc phải thu thập toàn bộ phần tử của luồng vào một mảng tạm thời rồi mới tiến hành sắp xếp (Full buffer).
2. distinct(): Sử dụng HashSet bên dưới để kiểm tra equals() và hashCode(), kích thước bộ đệm tỷ lệ thuận với số lượng phần tử duy nhất.
3. limit(n) và skip(n): limit là short-circuit intermediate operation giúp ngắt stream sau khi đã phát đủ n phần tử; skip duyệt và bỏ qua n phần tử đầu tiên.

// Thu tu sap xep quyet dinh hieu nang:
// KHUYEN NGHICH: Loc truoc, sort sau de giam kich thuoc buffer
items.stream()
    .filter(LibraryItem::isOverdue)
    .sorted(Comparator.comparing(LibraryItem::getBarcode))
    .toList();

5. Tại sao không được dùng peek cho logic thay đổi trạng thái?

Phương thức peek(Consumer<T>) cho phép ta nhìn trộm dữ liệu giữa các chặng pipeline nhằm phục vụ mục đích in log gỡ lỗi (debug). Tuy nhiên, tuyệt đối không dùng peek để thực thi logic nghiệp vụ hay thay đổi trạng thái đối tượng:

// ANTI-PATTERN NGUY HIEM: peek co the bi JVM bo qua hoan toan
List<LibraryItem> audited = new ArrayList<>();
long total = items.stream()
    .peek(audited::add) // Co the KHONG BAO GIO duoc thuc thi!
    .count();

Nguyên nhân: Từ Java 9+, nếu Terminal Operation là một thao tác như count() chạy trên một Stream đã biết trước kích thước (SIZED), trình tối ưu hóa của JVM sẽ trả về ngay kích thước của nguồn mà không duyệt qua bất kỳ phần tử nào. Kết quả là danh sách audited vẫn hoàn toàn rỗng.

Để thực thi tác vụ phụ, hãy sử dụng Terminal Operation forEach() hoặc thu gom kết quả trước bằng toList().

6. Đâu là những cạm bẫy phổ biến khi dùng các thao tác nâng cao?

❌ Cạm bẫy 1: Dùng sorted() trên stream vô hạn mà không có limit() đứng trước

// Ung dung bi treo vinh vien va gay OutOfMemoryError:
Stream.iterate(1, i -> i + 1).sorted().limit(10).toList();

// Chuan: Limit truoc de bien thanh stream huu han roi moi sort
Stream.iterate(1, i -> i + 1).limit(10).sorted().toList();

❌ Cạm bẫy 2: Dùng map() thay vì flatMap() khi giải nén Optional

// Sai kieu du lieu: Stream<Optional<String>>
Stream<Optional<String>> optStream = Stream.of(Optional.of("A"), Optional.empty());
var res = optStream.map(opt -> opt.orElse("")).toList();

// Chuan muc: Dung flatMap va Optional::stream de loai bo empty tu nhien
List<String> clean = optStream.flatMap(Optional::stream).toList(); // ["A"]

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

• flatMap phẳng hóa cấu trúc lồng nhau theo tỷ lệ 1-sang-N, thay thế triệt để các vòng lặp lồng nhau.
• mapMulti (Java 16+) giúp giảm áp lực cấp phát đối tượng trên Heap trong các pipeline xử lý dữ liệu lớn.
• takeWhile và dropWhile cắt dòng dữ liệu theo tiền tố và chỉ hoạt động chính xác trên Ordered Stream.
• sorted() và distinct() là thao tác có trạng thái, tuyệt đối không đặt sau nguồn dữ liệu vô hạn nếu chưa giới hạn bằng limit().

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

Bài tiếp theo: Optional: Xử Lý Thiếu Dữ Liệu An Toàn Thay Thế Null