Stream Basics: Pipeline Lazy và Terminal Operation

TL;DR: Stream không phải là cấu trúc dữ liệu lưu trữ phần tử như Collection, mà là một pipeline mô tả chuỗi biến đổi dữ liệu. Các thao tác trung gian (Intermediate Operations như filter, map) hoạt động theo cơ chế Lazy Evaluation—chúng chỉ đăng ký ý định vào chuỗi liên kết mà không duyệt dữ liệu. Chỉ khi một thao tác kết thúc (Terminal Operation như toList, count, forEach) được kích hoạt, luồng dữ liệu mới thực sự chảy qua các trạm (Sink chaining) theo nguyên tắc từng phần tử một (element-by-element). Nhờ đó, Stream không tạo mảng trung gian, hỗ trợ stream vô hạn và tối ưu hóa thời gian tính toán với cơ chế ngắt sớm (short-circuit).

Sau khi đã nắm vững cú pháp Lambda và Functional Interface cùng kỹ thuật tối ưu qua Method Reference, chúng ta bước vào ứng dụng quan trọng nhất của lập trình hàm trong Java: Stream API.

1. Stream khác biệt gì so với Collection truyền thống?

Một sai lầm phổ biến khi mới tiếp cận Java Stream là xem nó như một dạng Collection nâng cao. Về bản chất thiết kế, hai khái niệm này hoàn toàn đối nghịch nhau:

List<LibraryItem> items = getLibraryCatalog();

// 1. Collection duyet lai nhieu lan binh thuong:
items.forEach(item -> System.out.println(item.getTitle()));
items.forEach(item -> System.out.println(item.getBarcode()));

// 2. Stream chi duyet dung mot lan duy nhat:
Stream<LibraryItem> stream = items.stream();
long count = stream.count(); // Terminal operation consume stream!

// stream.forEach(System.out::println); 
// EXCEPTION: IllegalStateException: stream has already been operated upon or closed

[!NOTE] Để duyệt lại một pipeline, bạn phải tạo một Stream mới từ nguồn (items.stream()). Thao tác tạo Stream từ Collection rất nhẹ vì nó chỉ khởi tạo một đối tượng bao bọc Spliterator mà không sao chép dữ liệu.

2. Làm thế nào để khởi tạo một Stream từ các nguồn dữ liệu?

Stream có thể được tạo từ nhiều nguồn phong phú:

// 1. Tu Collection (pho bien nhat):
Stream<LibraryItem> fromList = items.stream();

// 2. Tu mang (Array):
Stream<String> fromArray = Arrays.stream(new String[]{"Java", "Spring", "SQL"});
IntStream primitiveStream = Arrays.stream(new int[]{1, 2, 3}); // Tranh boxing Integer

// 3. Tu phuong thuc Factory co dinh:
Stream<String> fixed = Stream.of("LIB-01", "LIB-02", "LIB-03");

// 4. Tu nguon vo han (Generator - bat buoc phai co dieu kien ngat):
Stream<Integer> infiniteIds = Stream.iterate(1001, id -> id + 1);
Stream<Double> randoms = Stream.generate(Math::random);

// 5. Tu I/O Stream (can try-with-resources de dong file handle):
try (Stream<String> lines = Files.lines(Path.of("catalog.csv"))) {
    lines.filter(line -> !line.isBlank()).forEach(System.out::println);
}

3. Intermediate và Terminal Operation khác nhau như thế nào?

Các phương thức trong Stream API được phân chia thành 2 nhóm rõ rệt dựa trên kiểu dữ liệu trả về:

1. Intermediate Operations: Luôn trả về một Stream<R> mới. Các thao tác này là Lazy—chúng không xử lý dữ liệu ngay mà chỉ nối thêm một mắt xích vào chuỗi pipeline:
   • Stateless: filter(), map(), flatMap(), peek(). Xử lý từng phần tử độc lập mà không cần ghi nhớ các phần tử trước.
   • Stateful: sorted(), distinct(). Cần ghi nhớ trạng thái các phần tử đã đi qua (cần bộ đệm buffer).
2. Terminal Operations: Trả về một giá trị không phải Stream (List, Map, Optional, long, void). Terminal operation là công tắc kích hoạt toàn bộ pipeline chạy và đóng Stream ngay sau khi hoàn tất:
   • Thu thập: collect(), toList(), toArray().
   • Thu gọn: reduce(), count(), min(), max().
   • Kiểm tra: anyMatch(), allMatch(), noneMatch(), findFirst().
   • Tác vụ phụ: forEach().

4. Tại sao pipeline không chạy nếu thiếu Terminal Operation?

Hãy quan sát đoạn mã lọc sách quá hạn sau:

// Doan ma nay hoan toan KHONG in gi ra man hinh:
items.stream().filter(item -> {
    System.out.println("Checking item: " + item.getBarcode());
    return item.isOverdue();
});

Không có lỗi runtime nào xảy ra, nhưng hàm println không bao giờ được gọi. Lý do: Không có Terminal Operation nào được kích hoạt.

Cơ chế Sink Chaining và Loop Fusion dưới JVM

Khi gọi filter(...) hay map(...), JDK chỉ tạo các node AbstractPipeline liên kết đơn trỏ ngược về node trước đó.

Chỉ khi bạn gắn thêm một Terminal Operation (ví dụ .toList()):

1. JVM duyệt ngược từ Terminal về Head để thiết lập một chuỗi các Sink.
2. Source Spliterator bắt đầu đẩy từng phần tử qua chuỗi Sink.
3. Toàn bộ các thao tác trung gian được nén lại thành một vòng lặp duy nhất (Loop Fusion), tận dụng tối đa L1/L2 Cache của CPU và triệt tiêu hoàn toàn nhu cầu cấp phát các danh sách trung gian trên Heap.

5. Tại sao phần tử di chuyển từng cái qua toàn pipeline thay vì chạy theo lô?

Một hiểu lầm phổ biến là nghĩ Stream hoạt động theo lô (batch): lọc hết toàn bộ danh sách, sau đó mới lấy kết quả đem đi map. Thực tế, dữ liệu di chuyển dọc theo từng phần tử (element-by-element):

List<String> titles = items.stream()
    .filter(item -> {
        System.out.println("Filter: " + item.getBarcode());
        return item.isOverdue();
    })
    .map(item -> {
        System.out.println("Map: " + item.getBarcode());
        return item.getTitle();
    })
    .limit(2)
    .toList();

Giả sử item 1 quá hạn, item 2 không quá hạn, và item 3 quá hạn:

Filter: ITEM-01
Map: ITEM-01
Filter: ITEM-02   <-- Bi filter loai, khong he goi map!
Filter: ITEM-03
Map: ITEM-03      <-- limit(2) da nhan du 2 phan tu, ngat pipeline lap tuc!

ITEM-01 đi hết từ filter qua map trước khi ITEM-02 được kiểm tra. Khi ITEM-02 bị loại ở trạm filter, nó lập tức bị hủy mà không tốn bất kỳ chu kỳ CPU nào ở trạm map phía sau.

6. Cơ chế Short-circuit giúp xử lý Stream vô hạn an toàn ra sao?

Cơ chế xử lý từng phần tử mở đường cho tính năng Short-circuit (ngắt sớm)—cho phép pipeline dừng ngay khi thỏa mãn điều kiện mà không cần duyệt phần còn lại:

• Short-circuit Intermediate: limit(n), takeWhile(predicate). Dừng kéo phần tử từ upstream khi đã gom đủ số lượng hoặc điều kiện sai.
• Short-circuit Terminal: findFirst(), findAny(), anyMatch(), allMatch(), noneMatch(). Trả kết quả ngay khi xác định được giá trị chân lý hoặc phần tử đầu tiên.

// Xu ly stream vo han an toan tuyet doi:
Optional<Integer> firstLargeId = Stream.iterate(1, id -> id + 1)
    .filter(id -> id > 10_000)
    .findFirst(); // Dung ngay lap tuc o gia tri 10001

[!WARNING] Tuyệt đối không gọi các thao tác Stateful Intermediate (sorted(), distinct()) trên một Stream vô hạn mà không có limit() đứng trước. Do phải nhìn thấy toàn bộ phần tử mới có thể sắp xếp, sorted() sẽ chạy vô tận và gây cạn kiệt bộ nhớ Heap (OutOfMemoryError).

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

❌ Cạm bẫy 1: Tái sử dụng Stream đã đóng

Stream<LibraryItem> stream = items.stream();
stream.forEach(System.out::println);
long total = stream.count(); // Throws IllegalStateException!

Khắc phục: Luôn tạo stream mới từ Collection khi cần thực hiện thao tác tính toán tiếp theo.

❌ Cạm bẫy 2: Lạm dụng peek() cho các tác vụ thay đổi trạng thái (Side-effects)

// Subtly skipped on Java 9+ if stream is SIZED
items.stream().peek(item -> item.markAudited()).count();

Khắc phục: peek() chỉ dùng để in log gỡ lỗi (debug). Các thao tác thay đổi trạng thái phải đặt trong terminal operation forEach().

❌ Cạm bẫy 3: Thay đổi Collection nguồn ngay bên trong biểu thức Lambda của Stream

// Throws ConcurrentModificationException:
items.stream().forEach(item -> {
    if (item.isDamaged()) items.remove(item);
});

Khắc phục: Sử dụng items.removeIf(LibraryItem::isDamaged) hoặc gom các phần tử hợp lệ vào một danh sách mới bằng .toList().

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

• Stream là pipeline tính toán một chiều (one-shot), không phải kho lưu trữ dữ liệu.
• Intermediate Operations (filter, map) đánh giá lười (lazy); Terminal Operations (toList, count) kích hoạt chuỗi Sink thực thi.
• Dữ liệu di chuyển tuần tự từng phần tử qua toàn bộ pipeline (Loop Fusion), tiết kiệm bộ nhớ và tối ưu cache.
• Short-circuit (limit, findFirst, anyMatch) cho phép dừng pipeline sớm, bảo đảm an toàn khi làm việc với stream vô hạn.
• Không bao giờ chạy thao tác có trạng thái (sorted) trên stream vô hạn nếu không có limit chặn trước.

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

Bài tiếp theo: Map, Filter và Reduce: Ba Thao Tác Cốt Lõi Của Stream API