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 dừng sớm khi đủ kết quả — short-circuit terminal (findFirst, anyMatch) phát tín hiệu dừng toàn pipeline, còn short-circuit intermediate (limit) ngừng kéo element từ upstream. 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 — short-circuit terminal (findFirst, anyMatch) và short-circuit intermediate (limit) dừng sớm khi đủ kết quả, 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 (interface bên dưới Stream dùng để traverse source — "split + iterator", cho phép parallel). 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 (vo han — can limit hoac short-circuit)
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 (lazy nhung huu han — can try-with-resources de dong)
try (Stream<String> lines = Files.lines(Path.of("log.txt"))) {
    lines.forEach(System.out::println);
}

Stream vô hạn (iterate, generate) 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. Files.lines thì khác: nó lazy nhưng hữu hạn — tự kết thúc khi hết file, không cần limit; chỉ cần try-with-resources để đóng file descriptor. Lưu ý: với file rất lớn, tránh .collect(toList()) toàn bộ vào List — xử lý từng dòng qua pipeline hoặc ghi ra stream output thay vì gom hết vào RAM.

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 và terminal operation khác nhau thế nào?

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. Xem thêm cách lambda làm tham số cho các op này tại Lambda & Functional Interface.

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. Tại sao code không có terminal op không chạy gì cả?

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

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

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ế JDK — xâu chuỗi Sink & Loop Fusion

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

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 đều kế thừa từ AbstractPipeline (như ReferencePipeline cho kiểu đối tượng, IntPipeline cho số nguyên). Stage sau giữ reference trỏ về stage liền trước.

2. Cơ chế Loop Fusion (Nén vòng lặp): Khi một terminal operation được kích hoạt, JDK xâu chuỗi các Sink ngược từ terminal về head (sink wiring). Mỗi intermediate stage đăng ký logic xử lý vào Sink tương ứng; khi source Spliterator push element, element chảy qua toàn chuỗi Sink đó trong một lần duyệt duy nhất. Thay vì chạy một vòng lặp cho mỗi stage, toàn bộ intermediate op được gộp (fused) thành 1 lượt.

   Lợi ích:

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

Ngoài generic stream Stream<T>, JDK còn cung cấp IntStream, LongStream, DoubleStream — các Primitive Stream thao tác trực tiếp trên giá trị nguyên thủy, tránh boxing per-element mà Stream<Integer> phải chịu. Đào sâu boxing/unboxing và benchmark ở bài nâng cao.

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

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

// Cach tu nhien (imperative) - ton 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 intermediate lists created -> waste Heap and GC

Stream (Nhờ Loop Fusion):

List<Integer> result = nums.stream()
    .filter(n -> n % 2 == 0)
    .map(n -> n * n)
    .toList();
// Khong co bat ky list trung gian nao duoc tao ra

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

// Tim 5 so chan dau tien tu stream vo han
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.

6. Tại sao element đi từng cái qua pipeline, 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ơ đồ luồng chạy dọc — Lazy Evaluation Element-by-Element

flowchart LR
    SRC["Source [1,2,3]"]
    F["filter: n > 1"]
    M["map: n * 10"]
    T["toList()"]
    X1[" "]

    SRC -->|"element 1"| F
    F -->|"fail - stop"| X1
    SRC -->|"element 2"| F
    F -->|"pass"| M
    M -->|"20"| T
    SRC -->|"element 3"| F
    F -->|"pass"| M
    M -->|"30"| T

Đâ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ự.

Walk-through từng bước — pipeline đơn giản

Theo dõi từng "tick" của pipeline Stream.of(1,2,3).filter(n -> n > 1).map(n -> n * 10).toList():

Tick 1 — element 1:
  filter(1 > 1) -> false  => stop, map KHONG duoc goi cho 1

Tick 2 — element 2:
  filter(2 > 1) -> true   => pass
  map(2 * 10)   -> 20     => push vao result

Tick 3 — element 3:
  filter(3 > 1) -> true   => pass
  map(3 * 10)   -> 30     => push vao result

Result: [20, 30]

Điểm then chốt: map chỉ được gọi 2 lần (cho element 2 và 3), không phải 3 lần. Element bị filter loại dừng ngay tại trạm đó — các stage sau không tốn công.

7. Short-circuit hoạt động như thế nào — intermediate và terminal?

Short-circuit có hai nhóm:

Short-circuit terminal — dừng pipeline và trả kết quả ngay khi đủ:

• findFirst(), findAny() — dừng khi tìm được 1 element.
• anyMatch(predicate) — dừng khi thấy element đầu tiên mà predicate trả true.
• allMatch(predicate) — dừng khi thấy element đầu tiên mà predicate trả false.
• noneMatch(predicate) — dừng khi thấy element đầu tiên mà predicate trả true (ngược chiều với allMatch). Edge case: stream rỗng → noneMatch luôn trả true.

Short-circuit intermediate — dừng việc kéo element từ upstream khi đã đủ:

• limit(n) — emit tối đa n element rồi signal upstream dừng. Đây là intermediate operation (trả Stream<T>), không phải terminal.
• takeWhile(predicate) — emit element chừng nào predicate còn true, dừng khi gặp false đầu tiên.

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.

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.

8. Vì sao stream chỉ dùng được một lần?

Đâ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+

Terminal op count() trên stream SIZED (list có kích thước xác định, không có filter) cho phép JVM biết count = list.size() mà không cần duyệt element — peek không bao giờ được gọi. Java 9+ spec chính thức cho phép optimisation này.

✅ 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. Tài liệu tham khảo chính thức (Oracle / JEP)

11. Liên hệ các bài khác

• Lambda & Functional Interface — filter(Predicate), map(Function), forEach(Consumer) đều nhận functional interface làm tham số. Bài đó định nghĩa Predicate, Function, Consumer — đọc trước để hiểu tại sao lambda và stream khớp nhau tự nhiên.
• Method Reference & invokedynamic — System.out::println trong forEach(System.out::println) là method reference. Bài đó giải thích cách JVM compile nó thành invokedynamic và tại sao viết ngắn hơn lambda mà không chậm hơn.
• map / filter / reduce — ba operation cốt lõi — bài tiếp theo đào sâu vào chính xác 3 op được nhắc nhiều nhất trong bài này, kể cả reduce nontrivial và cách xây aggregation phức tạp.
• Parallel Stream — stream pipeline lazy (bài này) chính là nền tảng để parallel hóa. Spliterator được nhắc ở bài này là cơ chế bên dưới parallelStream() — bài đó giải thích khi nào parallel thực sự có lợi và khi nào chậm hơn.

12. 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, allMatch, noneMatch) dừng khi đủ kết quả. Short-circuit intermediate (limit, takeWhile) dừng việc kéo element từ upstream khi đủ. Cả hai 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.

13. 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