Stream basics — pipeline lazy và terminal operation

TL;DR: Stream không phải collection — nó là pipeline mô tả chuỗi xử lý, không lưu element và chỉ duyệt được 1 lần. Intermediate operation (filter, map, sorted) lazy: chỉ ghi lại ý định; terminal operation (collect, count, forEach) mới kích hoạt pipeline chạy. Nhờ lazy, element đi từng cái qua toàn pipeline (không batch), stream vô hạn khả thi, và short-circuit (findFirst, anyMatch, limit) dừng sớm khi đủ kết quả. Pitfall lớn nhất: quên terminal op (pipeline không chạy gì), reuse stream đã consume (IllegalStateException), và stateful op (sorted, distinct) trên stream vô hạn (treo).

Đoạn code sau in gì?

Stream.of(1, 2, 3)
    .filter(x -> { System.out.println("check " + x); return x > 1; });

Câu trả lời: không in gì. Không phải lỗi, không phải bug — đây là đặc tính cốt lõi của Stream API.

Với dev đến từ JavaScript/Python/Ruby, phản xạ đầu tiên là: "filter chắc chắn phải chạy, nó nhận 3 element, in ra 3 dòng check". Trong Java Stream, filter không chạy gì cho đến khi có terminal operation — chủ đề chính của bài này.

Đây không phải quirk kỳ lạ. Lazy evaluation là thiết kế có chủ đích, cho phép 3 điều:

1. Pipeline dài vẫn chạy trong 1 lần duyệt — filter → map → filter → map → collect không tạo 4 collection trung gian.
2. Stream vô hạn khả thi — Stream.iterate(0, i -> i + 1) không thực sự chạy cho đến khi terminal op yêu cầu.
3. Short-circuit — findFirst, anyMatch, limit dừng ngay khi đủ, tiết kiệm công.

Bài này giải thích Stream là gì (và không phải gì), phân biệt intermediate vs terminal operation, element flow element-by-element qua pipeline, và khi nào short-circuit kích hoạt.

1. Analogy — Dây chuyền lắp ráp

Tưởng tượng nhà máy sản xuất ô tô. Băng chuyền có nhiều trạm: gắn bánh, sơn, kiểm tra, đóng gói. Công nhân đứng tại mỗi trạm.

Trạng thái ban đầu: băng chuyền đứng yên. Kể cả có linh kiện đang xếp đầu băng chuyền, không có gì xảy ra. Công nhân đứng chờ. Nguyên liệu ngồi đó.

Nhấn nút "bắt đầu" (công tắc): linh kiện đầu tiên chạy qua tất cả trạm — gắn bánh, sơn, kiểm tra, đóng gói — thành 1 xe hoàn chỉnh. Rồi linh kiện thứ 2 chạy. Rồi linh kiện thứ 3.

Không phải: "gắn bánh cho cả 1000 linh kiện, rồi sơn cả 1000, rồi kiểm tra cả 1000". Đó là cách build theo batch — tốn không gian (phải chứa 1000 xe sơn xong chờ kiểm tra), tốn thời gian (muốn xe đầu tiên phải đợi cả batch).

Stream hoạt động giống băng chuyền. Intermediate operation là các trạm — filter, map, sorted. Terminal operation (collect, count, forEach) là công tắc — không nhấn, băng chuyền đứng yên.

2. Stream là gì — và không phải gì

Đây là điểm hay nhầm của người đến từ ngôn ngữ khác. Trong Python, Ruby, JavaScript, bạn gọi list.filter(...).map(...) — mỗi bước tạo list mới, giữ trong memory, xử lý eager.

Java Stream không phải cấu trúc lưu trữ. Nó là abstraction cho chuỗi xử lý. Khác biệt rõ ràng với List:

Code minh hoạ:

List<Integer> list = List.of(1, 2, 3, 4, 5);

// List duyet nhieu lan duoc
list.forEach(System.out::println);
list.forEach(System.out::println);   // Chay dung lan thu 2

// Stream chi duyet 1 lan
Stream<Integer> s = list.stream();
s.forEach(System.out::println);   // OK
s.forEach(System.out::println);   // IllegalStateException: stream has already been operated upon or closed

Vì sao chỉ 1 lần? Vì stream không lưu element — nó giữ con trỏ tới source (list, file, generator). Khi terminal op chạy, nó consume source qua spliterator. Consume xong, stream mark closed.

Lý do thiết kế: stream có thể từ nguồn không rewind được — file I/O, network socket, generator. Nếu cho phép duyệt lại, mỗi source cần implement "reset" — vừa phức tạp vừa không khả thi cho một số source. Java chọn nhất quán: mọi stream one-shot.

Muốn duyệt lại? Tạo stream mới từ source:

list.stream().forEach(System.out::println);
list.stream().forEach(System.out::println);   // OK - stream moi

Tạo stream rẻ (chỉ là con trỏ), không copy data.

3. Tạo stream — nhiều nguồn

Stream có thể đến từ đâu?

// Tu collection - pho bien nhat
Stream<String> fromList = List.of("a", "b", "c").stream();

// Tu mang
Stream<Integer> fromArray = Arrays.stream(new Integer[]{1, 2, 3});
IntStream fromIntArray = Arrays.stream(new int[]{1, 2, 3});   // Primitive stream

// Tu factory method
Stream<String> fromOf = Stream.of("a", "b", "c");
Stream<String> empty = Stream.empty();

// Tu generator (co the vo han)
Stream<Integer> infiniteInts = Stream.iterate(0, i -> i + 1);         // 0, 1, 2, 3, ...
Stream<Double> randoms = Stream.generate(Math::random);                // random infinite
IntStream range = IntStream.range(0, 10);                              // 0..9 (exclusive)
IntStream rangeClosed = IntStream.rangeClosed(1, 10);                  // 1..10 (inclusive)

// Tu file (phai close)
try (Stream<String> lines = Files.lines(Path.of("log.txt"))) {
    lines.forEach(System.out::println);
}

Quan trọng: stream vô hạn (iterate, generate, Files.lines với file streaming) bắt buộc phải kết hợp với limit hoặc short-circuit terminal op — nếu không sẽ chạy mãi.

Stream.iterate(0, i -> i + 1)
    .limit(5)                 // BAT BUOC - khong co limit se vo han
    .forEach(System.out::println);

Stream.iterate(seed, UnaryOperator) còn có dạng 3-arg (Java 9+) với predicate dừng — giống for:

// Java 9+: iterate co predicate
Stream.iterate(0, i -> i < 100, i -> i + 1)
    .forEach(System.out::println);
// Tuong duong: for (int i = 0; i < 100; i++)

4. Intermediate vs terminal operation

Stream API có 2 nhóm operation, phân biệt bởi return type:

Intermediate operation trả về Stream<T> (hoặc IntStream, LongStream, ...) — vẫn là stream, chain tiếp được. Lazy — không chạy logic, chỉ ghi lại ý định vào pipeline.

Terminal operation trả về kết quả khác stream (List, long, boolean, void, ...) — đóng stream, kích hoạt pipeline.

List<Integer> nums = List.of(1, 2, 3, 4, 5);

long count = nums.stream()
    .filter(n -> n % 2 == 0)   // intermediate -> Stream<Integer>
    .map(n -> n * n)           // intermediate -> Stream<Integer>
    .count();                   // terminal -> long
// count = 2 (so 2 va 4 qua filter)

Bảng đầy đủ các op thường dùng:

Cách phân biệt nhanh: trả Stream<X> → intermediate. Trả khác → terminal.

5. Lazy evaluation — chi tiết

Quay lại code mở đầu bài:

Stream.of(1, 2, 3)
    .filter(x -> { System.out.println("check " + x); return x > 1; });
// Không in gì

Không có terminal op → pipeline không chạy. Thêm toList():

var result = Stream.of(1, 2, 3)
    .filter(x -> { System.out.println("check " + x); return x > 1; })
    .toList();

Output:

check 1
check 2
check 3

result = [2, 3]. Filter chạy 3 lần — 1 cho mỗi element.

Cơ chế biểu diễn Pipeline của JDK & Loop Fusion

Để thực thi cơ chế lazy evaluation, JDK thiết kế Stream API dựa trên mô hình cấu trúc dữ liệu chặt chẽ bên dưới:

1. Mô hình Stage liên kết (Pipeline Stages): Mỗi intermediate operation thực chất là một node được xâu chuỗi trong một danh sách liên kết đơn các pipeline stages. Các stage này đều kế thừa từ class abstract AbstractPipeline (như ReferencePipeline cho kiểu đối tượng, IntPipeline cho số nguyên). Stage sau giữ một reference trỏ về stage liền trước của nó.

2. Cơ chế Loop Fusion (Nén vòng lặp): Khi một terminal operation được kích hoạt, JDK sẽ tiến hành biên dịch ngược toàn bộ pipeline. Thay vì chạy một vòng lặp cho mỗi stage (lọc hết dữ liệu rồi mới biến đổi hết), JDK sử dụng cơ chế Loop Fusion để gộp toàn bộ các intermediate operations thành một lượt duyệt duy nhất trên nguồn dữ liệu thông qua cấu trúc điều phối dòng dữ liệu gọi là Sink (đầu thu nhận dữ liệu trung gian) và bộ phân tách nguồn Spliterator.

   Lợi ích:

   • Tận dụng CPU cache: dữ liệu của một phần tử được giữ trong thanh ghi/cache L1-L2 từ lúc đọc ra khỏi nguồn, qua các bộ lọc và biến đổi, đến khi đẩy vào terminal — không nhảy vùng nhớ liên tục gây cache miss.
   • Không tạo collection trung gian: không cần cấu trúc lưu trữ tạm giữa các bước — giảm bộ nhớ và áp lực GC.

Sự khác biệt hiệu năng của Primitive Stream vs Generic Stream

Bên cạnh generic stream thông thường Stream<T>, JDK cung cấp các Primitive Stream chuyên biệt: IntStream, LongStream, và DoubleStream.

• Generic Stream (Stream<Integer>): khi thao tác trên kiểu nguyên thủy (int, long, double), generic stream phải dùng wrapper (Integer, Long, Double). Mỗi phần tử đi qua pipeline bị boxing thành object (tốn ~16 byte header trên Heap mỗi giá trị) và unboxing ngược lại khi tính toán. Với hàng triệu phần tử, chi phí này rất đáng kể.
• Primitive Stream: IntStream, LongStream, DoubleStream thao tác trực tiếp trên giá trị nguyên thủy, không tạo wrapper object per-element. Lưu ý: dữ liệu vẫn nằm trên Heap (trong mảng/collection nguồn) — lợi ích đến từ việc pipeline tránh được toàn bộ chi phí boxing/unboxing, không phải từ việc "chuyển dữ liệu sang Stack".

Ưu thế 1: không tạo collection trung gian

So sánh với viết thủ công bằng loop eager:

// Cách tự nhiên (imperative) - Tốn memory
List<Integer> filtered = new ArrayList<>();
for (int n : nums) {
    if (n % 2 == 0) filtered.add(n);
}
List<Integer> squared = new ArrayList<>();
for (int n : filtered) {
    squared.add(n * n);
}
// 2 list trung gian tạo ra làm tốn Heap và GC

Stream (Nhờ Loop Fusion):

List<Integer> result = nums.stream()
    .filter(n -> n % 2 == 0)
    .map(n -> n * n)
    .toList();
// Không có bất kỳ list trung gian nào được tạo ra

Ưu thế 2: stream vô hạn khả thi

// Tìm 5 số chẵn đầu tiên từ stream vô hạn
List<Integer> first5Even = Stream.iterate(0, i -> i + 1)
    .filter(i -> i % 2 == 0)
    .limit(5)
    .toList();
// [0, 2, 4, 6, 8]

Nếu chạy theo cơ chế eager, filter phải hoàn thành trên toàn bộ stream vô hạn trước rồi mới truyền sang limit -> chương trình lập tức bị treo và tràn bộ nhớ. Nhờ đặc tính lazy, stream chỉ yêu cầu phần tử khi có tín hiệu pull từ limit. Khi limit nhận đủ 5 phần tử chẵn, nó phát tín hiệu short-circuit dừng toàn bộ pipeline ngược lên nguồn dữ liệu.

6. Luồng chạy dọc (Element-by-element), không chạy theo lô

Điểm phản trực giác tiếp theo: pipeline không chạy theo lô (batch processing) "filter cho tất cả, rồi map cho tất cả". Mỗi element đi dọc từ đầu đến cuối toàn bộ các stage của pipeline trước khi element kế tiếp được bắt đầu từ nguồn.

Stream.of(1, 2, 3)
    .filter(x -> { System.out.println("filter " + x); return x > 1; })
    .map(x -> { System.out.println("map " + x); return x * 10; })
    .toList();

Output:

filter 1
filter 2
map 2
filter 3
map 3

Sơ đồ trực quan luồng chạy dọc (Lazy Evaluation Element-by-Element)

Dưới đây là sơ đồ ASCII mô tả chi tiết luồng di chuyển dọc của từng phần tử qua pipeline, thể hiện rõ bản chất lazy thực thi thay vì xử lý theo lô:

  [ Nguồn dữ liệu (Source) ]
             │
             ├──> [ Phần tử 1 ] ──> filter: check 1 (Fail) ──> (Dừng & Loại bỏ ngay)
             │                                                      │ (Không đi qua map)
             │
             ├──> [ Phần tử 2 ] ──> filter: check 2 (Pass) ──> map: 2 * 10 (20) ──> Terminal (Gom kết quả)
             │
             └──> [ Phần tử 3 ] ──> filter: check 3 (Pass) ──> map: 3 * 10 (30) ──> Terminal (Gom kết quả)
                                                                                          │
                                                                                          ▼
                                                                                  [ [20, 30] ]

Đây là tính chất quan trọng giúp tối ưu hiệu năng:

• Tiết kiệm công sức tính toán: phần tử 1 bị filter loại sẽ dừng ngay, không phải chịu chi phí biến đổi ở các bước map hay flatMap phía sau.
• Khả thi hóa short-circuit: terminal operation phát tín hiệu dừng toàn pipeline ngay khi nhận đủ dữ liệu yêu cầu.
• An toàn với dữ liệu vô hạn: xử lý tuần tự từng phần tử giúp giữ bộ nhớ Heap ổn định trước các nguồn dữ liệu có độ dài vô hạn.

Ngoại lệ: stateful intermediate

Một số intermediate op phải buffer element trước khi emit:

• sorted() — phải thấy toàn bộ element để sort. Nếu stream vô hạn + sorted → hang forever.
• distinct() — phải giữ set element đã thấy để loại trùng. Memory tăng với số distinct.

Stream.of(3, 1, 4, 1, 5, 9, 2, 6)
    .sorted()                // Tai day phai buffer het
    .limit(3)                // Roi moi limit
    .forEach(System.out::println);
// 1, 1, 2

limit(3) trước hay sau sorted()? Khác nhau hoàn toàn. Thứ tự này: sort hết, lấy 3 đầu. Đảo limit(3).sorted(): lấy 3 element đầu từ source (3, 1, 4), sort → (1, 3, 4). Tư duy theo pipeline thứ tự.

7. Short-circuit terminal operation

Terminal op short-circuit dừng ngay khi đủ kết quả, không duyệt hết stream:

• findFirst(), findAny() — dừng khi tìm được 1 element.
• anyMatch(predicate) — dừng khi thấy element true đầu tiên.
• allMatch(predicate) — dừng khi thấy element false đầu tiên.
• noneMatch(predicate) — tương tự.
• limit(n) — dừng upstream khi đã emit n element.

Ví dụ dùng short-circuit trên stream vô hạn:

Optional<Integer> firstBig = Stream.iterate(0, i -> i + 1)
    .filter(i -> i > 100)
    .findFirst();
// Optional[101] - dung ngay khi tim duoc 101

Không có short-circuit → stream vô hạn chạy forever. Short-circuit là điều làm stream API "safe" với infinite source.

Tương tự:

boolean hasNegative = nums.stream().anyMatch(n -> n < 0);
// Stop ngay khi thay so am dau tien

Nếu nums có 1 triệu element và element thứ 3 đã âm, chỉ duyệt 3 element. Imperative loop tương đương:

boolean hasNegative = false;
for (int n : nums) {
    if (n < 0) { hasNegative = true; break; }
}

Stream version ngắn hơn, intent rõ hơn.

8. Stream bất biến — không dùng lại được

Đây là bẫy khác của người mới:

Stream<Integer> s = list.stream();

long count = s.count();              // OK, terminal op
s.forEach(System.out::println);      // IllegalStateException!
// stream has already been operated upon or closed

Mỗi stream dùng 1 lần. Có terminal op → consume → closed.

Muốn tái dùng logic, tạo function trả stream:

Supplier<Stream<Integer>> filtered = () -> list.stream().filter(n -> n > 0);

List<Integer> a = filtered.get().toList();
List<Integer> b = filtered.get().toList();   // OK - stream moi

Hoặc collect thành List rồi dùng lại list đó.

9. Pitfall tổng hợp

❌ Nhầm 1: Reuse stream.

Stream<Integer> s = list.stream();
s.count();
s.forEach(System.out::println);   // IllegalStateException

✅ Tạo stream mới mỗi lần hoặc collect thành List:

list.stream().count();
list.stream().forEach(System.out::println);

❌ Nhầm 2: Quên terminal op.

list.stream().filter(x -> x > 10);   // Khong chay gi

✅ Thêm terminal: .toList(), .forEach(...), .count().

❌ Nhầm 3: Dùng peek cho logic chính.

list.stream().peek(x -> save(x)).count();   // peek co the bi skip o Java 9+

✅ peek là intermediate, chỉ dùng debug/log. Dùng forEach cho side-effect chính:

list.stream().forEach(this::save);

❌ Nhầm 4: Modify collection gốc trong lambda stream.

list.stream().forEach(x -> list.add(x * 2));   // ConcurrentModificationException

✅ Collect thành list mới:

List<Integer> doubled = list.stream().map(x -> x * 2).toList();

❌ Nhầm 5: sorted() trên stream vô hạn.

Stream.iterate(0, i -> i + 1).sorted().limit(5);   // Hang forever

✅ sorted() cần thấy hết element. Dùng limit TRƯỚC sorted nếu phù hợp, hoặc đổi source.

10. 📚 Deep Dive Oracle

11. Tóm tắt

• Stream là pipeline mô tả xử lý, không phải cấu trúc lưu trữ. Không giữ element, không size() (với stream vô hạn), chỉ duyệt 1 lần.
• Intermediate op trả Stream<T>, lazy — chỉ ghi lại ý định vào pipeline.
• Terminal op trả kết quả khác stream, kích hoạt pipeline chạy từ source lên.
• Lazy evaluation cho phép: pipeline dài không tạo collection trung gian, stream vô hạn khả thi, short-circuit tiết kiệm công.
• Element đi từng cái qua toàn pipeline (filter → map → ...) trước khi element kế bắt đầu. Không batch.
• Stateful intermediate (sorted, distinct) phải buffer — không dùng với stream vô hạn.
• Short-circuit terminal (findFirst, anyMatch, limit) dừng khi đủ, làm stream vô hạn an toàn.
• Stream one-shot — consume xong là closed. Cần reuse → tạo stream mới từ source hoặc collect thành list.
• Tạo stream từ: collection, array, factory, generator (iterate, generate), file. Primitive stream (IntStream, ...) tránh boxing.

12. Tự kiểm tra

f1     // element 1 vao filter
f2     // element 2 vao filter
m2     // 2 qua filter, chay map
f3     // element 3 vao filter
m3     // 3 qua filter, chay map

stream
  .peek(x -> System.out.println("before filter: " + x))
  .filter(x -> x > 0)
  .peek(x -> System.out.println("after filter: " + x))
  .toList();

Bài tiếp theo: map / filter / reduce — ba operation cốt lõi