Streams API

Lambda phát huy sức mạnh rõ nhất khi đi cùng Streams API. Một stream là một chuỗi phần tử đi qua một loạt phép xử lý theo phong cách dây chuyền. Thay vì viết vòng lặp tường minh và biến tạm để lọc, biến đổi rồi gom dữ liệu, ta mô tả chuỗi thao tác cần làm; Java lo phần thực thi. Mã trở nên gần với ý định hơn: đọc một dây chuyền Stream giống như đọc một câu mô tả, lọc cái gì, biến đổi ra sao, gom lại thế nào.

Hãy thử một yêu cầu quen thuộc: từ danh sách sinh viên, lấy tên các bạn điểm trên tám, viết hoa, sắp theo bảng chữ cái. Viết bằng vòng lặp, bạn cần một danh sách tạm, một vòng for, một câu if, một lệnh add, rồi một lệnh sort; người đọc phải chạy nhẩm cả sáu dòng mới đoán ra bạn định làm gì. Streams cho phép viết thẳng ý định đó thành một dây chuyền.

Khác biệt sâu hơn nằm ở chỗ vòng lặp mô tả cách làm còn Stream mô tả việc cần làm. Khi ta chỉ nói việc cần làm, các chi tiết như biến đếm hay danh sách tạm thuộc về thư viện, và thư viện được tự do chọn cách thực thi tốt nhất, kể cả gộp nhiều phép vào một lượt duyệt hay chia việc cho nhiều lõi.

Dây chuyền một Stream

📥

Nguồn

Từ tập hợp, mảng hay dải số, ví dụ stream() hoặc IntStream.rangeClosed.

→

🔎

Thao tác trung gian

filter, map, sorted, distinct, limit; mỗi phép trả về một Stream mới nên nối tiếp được.

→

📦

Thao tác kết thúc

collect, reduce, forEach, count; đóng dây chuyền và sinh kết quả không còn là Stream.

Stream không phải cấu trúc dữ liệu

Stream không lưu trữ phần tử. Nó là một khung nhìn dùng một lần trên nguồn dữ liệu. Sau khi đã chạy một thao tác kết thúc, stream coi như đã tiêu thụ; gọi lại sẽ ném IllegalStateException. Stream cũng không làm thay đổi nguồn gốc: filter không xoá phần tử khỏi danh sách ban đầu.

Thử ngay: xem stream chạy từng bước

Chưa cần đọc hết lý thuyết, bấm ▶ Chạy để thấy Stream duyệt theo từng phần tử chứ không theo từng phép. Phần tử đi qua filter rồi map, và khi findFirst đã có kết quả đầu tiên thì cả dây chuyền dừng ngay, các phần tử còn lại không bao giờ bị đụng tới. Phần lý thuyết ở dưới sẽ giải thích cơ chế lười này.

Dây chuyền lười: filter, map, findFirst dừng sớm java

1import java.util.stream.Stream;
2
3public class Demo {
4    public static void main(String[] args) {
5        int first = Stream.of(1, 2, 3, 4)
6            .filter(n -> n % 2 == 0)       // keep even numbers
7            .map(n -> n * 10)              // multiply by ten
8            .findFirst()                   // stop at the first result
9            .orElse(-1);
10        System.out.println(first);
11    }
12}

Ngăn xếp stack

main()

first= chưa tính

Bộ nhớ động heap

(trống)

Dây chuyền được dựng nhưng chưa chạy gì: filter và map mới chỉ ghi nhận ý định. Stream lười nên tất cả chờ tới thao tác kết thúc findFirst.

1/70.5×1×2×

Stream không lưu phần tử

Ngộ nhận phổ biến nhất là coi Stream như một tập hợp kiểu mới, họ hàng của List. Không phải. List là cấu trúc dữ liệu: nó sở hữu các phần tử và cho ta hỏi phần tử thứ i bất cứ lúc nào. Stream không sở hữu gì cả. Nó là một bản mô tả phép tính đặt trên một nguồn có sẵn, kèm một con trỏ đọc đi một chiều. Hỏi "phần tử thứ ba của stream này là gì" là câu hỏi vô nghĩa, vì stream chưa tính gì cho tới lúc có thao tác kết thúc.

Từ đó suy ra ba tính chất. Thứ nhất, stream không sửa nguồn: filter không xoá phần tử nào khỏi danh sách gốc, nó chỉ quyết định phần tử nào được đi tiếp. Thứ hai, stream dùng đúng một lần, vì con trỏ đọc chỉ chạy một chiều và không tua lại được; muốn duyệt hai lần thì tạo hai stream từ cùng nguồn, việc này rẻ vì stream không sao chép dữ liệu. Thứ ba, chính vì không lưu phần tử nên stream mới vô hạn được: Stream.iterate mô tả một quy luật sinh phần tử chứ không dựng sẵn danh sách vô tận trong bộ nhớ.

Tạo stream

Nguồn phổ biến nhất là một tập hợp: mọi lớp cài Collection đều có phương thức stream(). Ngoài ra còn nhiều cách tạo khác, từ mảng, từ các giá trị rời rạc, hay từ một dải số.

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

Stream<Integer> s1 = nums.stream();                  // from a collection
Stream<String>  s2 = Stream.of("a", "b", "c");       // from explicit values
IntStream       s3 = IntStream.rangeClosed(1, 10);   // a range of int
Stream<Integer> s4 = Arrays.stream(new Integer[]{7, 8, 9}); // from an array

IntStream (cùng LongStream, DoubleStream) là các stream chuyên cho kiểu nguyên thuỷ, tránh chi phí đóng hộp và cung cấp sẵn các phép tổng hợp như sum(), average().

Hai nguồn đặc biệt là Stream.iterate và Stream.generate: chúng sinh ra stream vô hạn. Nghe có vẻ vô lý, nhưng nhờ tính lười (nói ở phần dưới), stream vô hạn hoàn toàn dùng được miễn là bạn chặn nó lại bằng limit.

// infinite stream, cut short by limit
List<Integer> luyThua = Stream.iterate(1, n -> n * 2)   // 1, 2, 4, 8, ...
    .limit(6)
    .collect(Collectors.toList());   // [1, 2, 4, 8, 16, 32]

Quên limit trên một stream vô hạn thì chương trình chạy mãi không dừng, nên hãy coi cặp iterate cộng limit là đi liền với nhau.

Thao tác trung gian và thao tác kết thúc

Một thao tác trung gian (intermediate operation) nhận một stream và trả về một stream mới, nhờ vậy ta nối chúng thành dây chuyền: filter, map, sorted, distinct, limit. Một thao tác kết thúc (terminal operation) đóng dây chuyền, tiêu thụ stream và sinh ra một kết quả không còn là stream: collect, reduce, forEach, count. Phân biệt hai loại qua kiểu trả về: trung gian trả về Stream nên còn nối được, kết thúc trả về kết quả khác và chấm dứt dây chuyền.

Thao tác	Loại	Ý nghĩa
filter	trung gian	Giữ phần tử thoả vị từ.
map	trung gian	Biến đổi từng phần tử, có thể đổi kiểu.
sorted	trung gian	Sắp xếp theo thứ tự tự nhiên hoặc bộ so sánh.
distinct	trung gian	Loại bỏ phần tử trùng lặp.
limit	trung gian	Giữ lại tối đa n phần tử đầu.
forEach	kết thúc	Chạy một hành động cho mỗi phần tử, không trả về.
collect	kết thúc	Gom phần tử vào tập hợp qua một Collector.
reduce	kết thúc	Gộp các phần tử thành một giá trị duy nhất.
count	kết thúc	Đếm số phần tử của stream.

List<String> names = List.of("An", "Binh", "Cuong", "Dung", "E");

List<String> result = names.stream()
    .filter(n -> n.length() > 3)   // keep names longer than 3 chars
    .map(String::toUpperCase)      // transform each to upper case
    .sorted()                      // sort by natural order
    .collect(Collectors.toList()); // gather into a new list
// result = [BINH, CUONG, DUNG]

reduce nhận một giá trị khởi đầu và một phép kết hợp hai ngôi, gộp dần tất cả phần tử thành một giá trị duy nhất. Đây là khuôn mẫu chung cho mọi phép tổng hợp như tổng, tích, lớn nhất, nhỏ nhất.

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

// reduce: combine all elements into one value (the sum)
int total = nums.stream().reduce(0, (a, b) -> a + b);  // 0+3=3, 3+1=4, ... -> 31

// collect: gather even numbers into a new list
List<Integer> evens = nums.stream()
    .filter(n -> n % 2 == 0)
    .collect(Collectors.toList());   // [4, 2, 6]

// forEach: a side effect for each element, returns nothing
nums.stream().distinct().forEach(System.out::println);

Tính lười

Một đặc điểm cốt lõi của Stream là tính lười (laziness). Các thao tác trung gian không thực thi ngay khi được gọi; chúng chỉ ghi nhận ý định và xếp vào dây chuyền. Toàn bộ dây chuyền chỉ khởi động khi gặp một thao tác kết thúc. Nhờ tính lười, các thao tác như findFirst hay limit có thể dừng sớm ngay khi đủ kết quả, không cần đụng tới các phần tử còn lại.

List<Integer> r = Stream.of(1, 2, 3, 4, 5)
    .map(n -> {
        System.out.println("mapping " + n);  // shows when map runs
        return n * n;
    })
    .limit(2)               // stop after two elements
    .collect(Collectors.toList());
// prints only "mapping 1" and "mapping 2", not 3, 4, 5
// r = [1, 4]

Hãy đọc kỹ kết quả in của đoạn trên, vì nó nói lên toàn bộ cơ chế. Nếu Stream chạy theo lối làm xong phép này rồi mới sang phép kia, map sẽ chạy trên cả năm phần tử rồi limit mới cắt còn hai. Thực tế chỉ hai dòng được in. Nghĩa là Stream không duyệt theo từng phép, nó duyệt theo từng phần tử: phần tử 1 đi qua map rồi tới limit, phần tử 2 cũng vậy, và tới đây limit đã đủ nên báo dừng; các phần tử 3, 4, 5 không bao giờ được đụng tới.

Cách chạy này mang lại hai khoản tiết kiệm. Thứ nhất là gộp lượt duyệt: filter, map, sorted nối nhau không tạo ba lần đi qua dữ liệu cùng ba danh sách trung gian, mà gộp thành một lượt duy nhất. Thứ hai là dừng sớm: limit, findFirst, anyMatch cho phép cả dây chuyền ngừng ngay khi đủ câu trả lời, nên tìm phần tử thoả điều kiện nằm ở vị trí thứ ba trong một triệu phần tử thì chỉ ba phần tử được xử lý.

Lớp Collectors

collect thường đi cùng các bộ gom (collector) dựng sẵn trong lớp tiện ích Collectors. Nhờ chúng, ta gom kết quả thành danh sách, tập, chuỗi nối, hay thậm chí thành Map nhóm theo khoá mà không phải viết tay vòng lặp tích luỹ. groupingBy đặc biệt mạnh: nó phân loại phần tử theo một hàm khoá rồi gom mỗi nhóm; tham số thứ hai (một bộ gom hạ nguồn) cho phép xử lý tiếp từng nhóm.

record Person(String name, String city) {}
List<Person> people = List.of(
    new Person("An", "Hue"), new Person("Binh", "Da Nang"),
    new Person("Cuong", "Hue"), new Person("Dung", "Da Nang"));

// joining: concatenate names with a separator
String line = people.stream()
    .map(Person::name).collect(Collectors.joining(", "));  // "An, Binh, Cuong, Dung"

// groupingBy: build a Map keyed by a classifier
Map<String, List<Person>> byCity = people.stream()
    .collect(Collectors.groupingBy(Person::city));
// Person has no toString of its own, so the record's generated one is used:
// {Hue=[Person[name=An, city=Hue], Person[name=Cuong, city=Hue]],
//  Da Nang=[Person[name=Binh, city=Da Nang], Person[name=Dung, city=Da Nang]]}

// counting downstream: count members per group
Map<String, Long> countByCity = people.stream()
    .collect(Collectors.groupingBy(Person::city, Collectors.counting()));
// {Hue=2, Da Nang=2}

Ba bộ gom trên phủ gần hết nhu cầu hằng ngày. toList gom phần tử vào một danh sách mới, không phải khung nhìn lên nguồn cũ. joining nối các chuỗi thành một chuỗi duy nhất kèm dấu phân cách, và vì dùng StringBuilder bên trong nên nhanh hơn hẳn cộng chuỗi trong vòng lặp. groupingBy chạy một hàm phân loại trên từng phần tử, lấy giá trị trả về làm khoá, rồi ném phần tử vào đúng ngăn của khoá ấy.

Điểm khiến groupingBy thật sự linh hoạt là tham số thứ hai, gọi là bộ gom hạ nguồn (downstream collector). Mặc định mỗi ngăn được gom thành một List, nhưng ta thay được cái đích ấy: counting() cho ra số đếm mỗi ngăn, averagingDouble cho ra điểm trung bình. Nói cách khác, groupingBy lo chia ngăn, còn bộ gom hạ nguồn lo mỗi ngăn biến thành cái gì.

Xem chạy từng bước: groupingBy gom vào Map

Bản trace ở đầu bài minh hoạ dây chuyền lười duyệt theo từng phần tử rồi dừng sớm. Bản này minh hoạ một cơ chế kết thúc khác: collect(Collectors.groupingBy(...)) gom cả stream vào một Map, phân loại từng phần tử theo một hàm khoá. Khác với dừng sớm, ở đây mọi phần tử đều phải được duyệt qua thì bản đồ mới đầy đủ. Hãy quan sát bản đồ lớn dần lên mỗi khi một người được ném vào đúng ngăn theo lớp của họ. Sơ đồ bộ nhớ viết tắt mỗi người thành cái tên cho dễ nhìn, còn dòng in ra ở bước cuối mới là thứ chương trình thật sự xuất: record P không ghi đè toString nên nó dùng bản tự sinh, in ra đủ cả tên thành phần lẫn giá trị.

collect(groupingBy) gom stream vào một Map theo khoá java

1import java.util.List;
2import java.util.Map;
3import java.util.stream.Collectors;
4
5public class Demo {
6    record P(String ten, String lop) {}
7    public static void main(String[] args) {
8        List<P> ds = List.of(new P("An", "A"), new P("Binh", "B"), new P("Cuong", "A"));
9        Map<String, List<P>> theoLop = ds.stream()
10            .collect(Collectors.groupingBy(P::lop));
11        System.out.println(theoLop);
12    }
13}

Ngăn xếp stack

main()

ds→ #1

Bộ nhớ động heap

#1List<P>

[0]An (A)

[1]Binh (B)

[2]Cuong (A)

Dựng danh sách ds gồm ba người: An lớp A, Binh lớp B, Cuong lớp A.

1/60.5×1×2×

reduce và tính kết hợp

reduce là khuôn mẫu tổng quát nhất để gộp nhiều phần tử thành một giá trị. Nó cần một giá trị khởi đầu và một phép hai ngôi để nhập dần từng phần tử vào kết quả tích luỹ. Tổng, tích, lớn nhất, nối chuỗi, tất cả đều là reduce với phép hai ngôi khác nhau. Hai điều kiện dưới đây nghe hình thức nhưng quyết định reduce chạy đúng hay sai.

Giá trị khởi đầu phải là phần tử trung hoà của phép gộp, tức gộp nó với giá trị nào cũng trả lại chính giá trị đó: 0 cho phép cộng, 1 cho phép nhân, chuỗi rỗng cho phép nối. Chọn sai thì kết quả sai âm thầm; và nếu chạy song song, mỗi cụm lại dôi thêm một lần nữa.

Phép gộp phải có tính kết hợp (associativity), tức (a op b) op c phải bằng a op (b op c). Điều kiện này tồn tại vì Java có quyền chia dữ liệu thành nhiều cụm, gộp riêng từng cụm rồi mới gộp các cụm lại. Cộng, nhân, lấy lớn nhất đều kết hợp nên an toàn. Phép trừ thì không: (10-3)-2 bằng 5, còn 10-(3-2) bằng 9. Stream tuần tự vẫn cho kết quả quen thuộc vì nó gộp lần lượt từ trái sang phải, nhưng đổi sang parallelStream() thì cách chia cụm quyết định con số, và đây là loại lỗi tệ nhất: mã vẫn chạy, không ném ngoại lệ, chỉ có kết quả là sai.

Mã thuần dễ kiểm thử

Phong cách Stream khuyến khích viết mã không tác dụng phụ trong các thao tác trung gian: hàm trong map hay filter chỉ nên tính toán dựa trên phần tử đầu vào, không sửa biến bên ngoài. Mã thuần như vậy dễ đọc, dễ kiểm thử, và an toàn nếu sau này chuyển sang parallelStream() để xử lý song song.

Stream song song

Đổi stream() thành parallelStream() là đủ để Java chia dữ liệu thành nhiều mảnh, giao cho nhiều luồng chạy đồng thời, rồi gộp kết quả lại. Một dòng sửa, và dây chuyền chạy trên nhiều lõi.

long soDong = vanBan.parallelStream()      // split across threads
    .filter(dong -> dong.contains("error"))
    .count();

Chính vì dễ như vậy nên nó nguy hiểm. Song song không miễn phí: Java phải chia dữ liệu, điều phối luồng, rồi gộp kết quả. Nếu phép tính quá nhẹ hoặc dữ liệu quá ít, chi phí điều phối lớn hơn phần tiết kiệm được, và bản song song chạy chậm hơn bản tuần tự.

Không nên dùng parallelStream() trong các trường hợp sau.

• Dữ liệu ít hoặc phép tính trên mỗi phần tử rất nhẹ. Chi phí chia và gộp nuốt hết lợi ích.
• Thao tác có tác dụng phụ hoặc sửa một biến dùng chung. Nhiều luồng cùng ghi vào một ArrayList sẽ hỏng dữ liệu, vì ArrayList không an toàn với đa luồng.
• Kết quả phụ thuộc thứ tự. Giữ đúng trật tự khi chạy song song lại tốn thêm chi phí đồng bộ.
• Phép gộp trong reduce không có tính kết hợp. Các mảnh được gộp theo cụm nên (a - b) - c khác a - (b - c); phép trừ không dùng song song được.
• Công việc chờ vào ra (đọc tệp, gọi mạng) chứ không phải tính toán. Luồng chỉ nằm chờ.

Ba lỗi thường gặp với Stream

Quên thao tác kết thúc. Dây chuyền chỉ có filter và map thì không chạy gì cả, vì thao tác trung gian lười.

Dùng lại một stream. Stream tiêu thụ một lần. Gán Stream<String> s = list.stream(); rồi gọi s.count() và sau đó s.forEach(...) sẽ ném IllegalStateException. Muốn duyệt hai lần thì tạo hai stream từ nguồn.

Sửa nguồn trong lúc duyệt. Gọi list.add(...) bên trong forEach đang duyệt chính list đó sẽ ném ConcurrentModificationException. Hãy thu kết quả vào danh sách mới bằng collect.

Nguyên nhân gốc của ba lỗi trên

Ba lỗi vừa nêu trông rời rạc, nhưng đều bắt nguồn từ đúng một hiểu lầm: coi stream là một tập hợp chứa sẵn dữ liệu. Nếu tin rằng stream chứa phần tử, ta sẽ nghĩ filter và map phải làm việc ngay khi được gọi nên quên thao tác kết thúc; sẽ nghĩ stream duyệt lại được như một List nên gán vào biến rồi dùng hai lần; và sẽ quên rằng nguồn đang bị đọc dở nên tiện tay sửa luôn nguồn trong forEach.

Nhớ đúng một câu là tránh được cả ba: stream là bản mô tả một phép tính chạy một lần trên nguồn, không phải bản sao dữ liệu.

Bài tập thực hành

Bài tập 1: dây chuyền lọc, biến đổi, gom chuỗi

Cho một List<String> các từ. Dùng Streams để: lọc các từ dài hơn bốn ký tự, đổi sang chữ thường, loại trùng lặp, sắp theo bảng chữ cái, rồi gom thành một chuỗi nối bằng dấu phẩy với Collectors.joining.

Lời giải chi tiết

Mỗi yêu cầu ứng với một thao tác trung gian, nối thành dây chuyền; cuối cùng collect với Collectors.joining(", ") đóng dây chuyền và gom thành một chuỗi.

import java.util.*;
import java.util.stream.Collectors;

List<String> words = List.of("Cuong", "an", "BINH", "cuong", "dung", "An");

String result = words.stream()
    .filter(w -> w.length() > 4)        // keep words longer than 4 chars
    .map(String::toLowerCase)           // normalize to lower case
    .distinct()                         // drop duplicates
    .sorted()                           // alphabetical order
    .collect(Collectors.joining(", ")); // join with a comma
System.out.println(result);             // cuong

Ý chính: mỗi bước là một thao tác trung gian trả về Stream, dây chuyền chỉ chạy khi gặp collect.

Bài tập 2: groupingBy với bộ gom hạ nguồn

Cho một List các record SinhVien(String ten, String lop, double diem). Dùng Collectors.groupingBy để nhóm sinh viên theo lớp, và với mỗi lớp tính điểm trung bình bằng bộ gom hạ nguồn Collectors.averagingDouble. In kết quả dạng lop -> diemTB.

Lời giải chi tiết

Bước 1: groupingBy(SinhVien::lop, ...) phân loại theo hàm khoá là tên lớp. Bước 2: bộ gom hạ nguồn averagingDouble(SinhVien::diem) xử lý từng nhóm, tính trung bình điểm. Kết quả là một Map<String, Double> từ tên lớp sang điểm trung bình. Bước 3: duyệt Map bằng forEach để in.

import java.util.*;
import java.util.stream.Collectors;

record SinhVien(String ten, String lop, double diem) {}

List<SinhVien> ds = List.of(
    new SinhVien("An", "K22A", 8.0),
    new SinhVien("Binh", "K22A", 6.0),
    new SinhVien("Cuong", "K22B", 9.0),
    new SinhVien("Dung", "K22B", 7.0));

Map<String, Double> tbTheoLop = ds.stream()
    .collect(Collectors.groupingBy(
        SinhVien::lop,                          // classifier: group by class
        Collectors.averagingDouble(SinhVien::diem))); // downstream: average

tbTheoLop.forEach((lop, tb) -> System.out.println(lop + " -> " + tb));
// K22A -> 7.0
// K22B -> 8.0

Ý chính: groupingBy cộng một bộ gom hạ nguồn diễn đạt gọn cho bài toán thống kê vốn cần nhiều dòng vòng lặp lồng nhau.

Bài tập 3: reduce tìm chuỗi dài nhất

Dùng reduce để tìm chuỗi dài nhất trong một List<String> mà không dùng vòng lặp tường minh. Sau đó viết lại bằng max(Comparator.comparingInt(String::length)) và so sánh hai cách về độ rõ ràng.

Lời giải chi tiết

reduce so sánh hai chuỗi mỗi bước, giữ chuỗi dài hơn làm kết quả tích luỹ. Cách max(Comparator...) ngắn và rõ ý hơn vì nói thẳng tiêu chí so sánh là độ dài.

import java.util.*;

List<String> words = List.of("an", "cuong", "binh", "duongvan");

// (a) reduce: keep the longer of two each step
String longest = words.stream()
    .reduce("", (a, b) -> a.length() >= b.length() ? a : b);
System.out.println(longest);   // duongvan

// (b) max with a comparator on length: clearer intent
Optional<String> longest2 = words.stream()
    .max(Comparator.comparingInt(String::length));
System.out.println(longest2.orElse(""));   // duongvan

Ý chính: reduce là khuôn mẫu tổng quát, nhưng khi tiêu chí là so sánh một thuộc tính, max với Comparator đọc rõ ràng hơn.

Bài tập 4: vì sao dây chuyền này không in gì

Đoạn dưới biên dịch được nhưng chạy xong không in ra dòng nào. Hãy chỉ ra nguyên nhân và sửa lại bằng hai cách khác nhau.

List.of("an", "binh", "cuong").stream()
    .filter(s -> s.length() > 2)
    .map(String::toUpperCase);

Lời giải chi tiết

Nguyên nhân: dây chuyền chỉ có hai thao tác trung gian, mà trung gian thì lười. Không có thao tác kết thúc nên filter và map chưa từng chạy, chúng mới chỉ được ghi nhận vào dây chuyền.

// fix 1: a terminal operation with a side effect
List.of("an", "binh", "cuong").stream()
    .filter(s -> s.length() > 2)
    .map(String::toUpperCase)
    .forEach(System.out::println);      // BINH, CUONG

// fix 2: a terminal operation that produces a value
List<String> kq = List.of("an", "binh", "cuong").stream()
    .filter(s -> s.length() > 2)
    .map(String::toUpperCase)
    .collect(Collectors.toList());      // [BINH, CUONG]
System.out.println(kq);

Ý chính: thiếu thao tác kết thúc thì dây chuyền không bao giờ khởi động.

Bài tập 5: reduce và tính kết hợp

Đoạn mã dưới tính hiệu của các số bằng reduce. Bản tuần tự in ra -15. Hãy dự đoán bản song song in ra gì, giải thích, rồi nói cách sửa nếu thật sự cần một phép trừ.

import java.util.*;

public class BaiTap5 {
    public static void main(String[] args) {
        List<Integer> nums = List.of(1, 2, 3, 4, 5);

        int tuanTu = nums.stream().reduce(0, (a, b) -> a - b);
        int songSong = nums.parallelStream().reduce(0, (a, b) -> a - b);

        System.out.println(tuanTu);    // -15
        System.out.println(songSong);  // ?
    }
}

Lời giải chi tiết

Bản tuần tự gộp lần lượt từ trái sang phải: 0-1 = -1, rồi -1-2 = -3, tiếp tục tới -10-5 = -15.

Bản song song cho ra một số khác, và con số ấy phụ thuộc cách dữ liệu bị chia cụm, tức phụ thuộc kích thước dữ liệu và số lõi của máy. Nguyên nhân là phép trừ không có tính kết hợp. Khi chạy song song, Java chia danh sách thành nhiều cụm, mỗi luồng gộp riêng cụm của mình, rồi mới trừ các kết quả cụm với nhau. Chẳng hạn nếu chia thành [1, 2] và [3, 4, 5], ta được 0-1-2 = -3 và 0-3-4-5 = -12, rồi gộp hai cụm thành -3 - (-12) = 9. Thêm nữa, giá trị khởi đầu 0 được đưa vào từng cụm chứ không phải một lần, nên nó cũng không còn đóng vai phần tử trung hoà đúng nghĩa. Điều đáng sợ là mã vẫn chạy trơn tru, không ném ngoại lệ nào.

Cách sửa là đừng ép phép trừ vào reduce, hãy diễn đạt lại bài toán bằng một phép có tính kết hợp: tổng thì cộng được song song thoải mái.

import java.util.*;

public class BaiTap5Fix {
    public static void main(String[] args) {
        List<Integer> nums = List.of(1, 2, 3, 4, 5);

        // subtraction is not associative; sum is, so express it with sum
        int tong = nums.parallelStream().reduce(0, Integer::sum);  // 15
        int ketQua = 0 - tong;                                     // -15, stable

        System.out.println(ketQua);   // -15 every run, sequential or parallel
    }
}

Ý chính: reduce chỉ an toàn khi phép gộp có tính kết hợp và giá trị khởi đầu là phần tử trung hoà. Gặp một phép không kết hợp, đừng chạy song song, hãy viết lại bài toán bằng một phép kết hợp.

Câu hỏi tự kiểm

Kiểm tra nhanh: Streams API0/4 đúngchưa trả lời

1. 1Một dây chuyền chỉ gồm .filter(...).map(...) mà không có thao tác kết thúc thì chuyện gì xảy ra?
2. 2Vì sao reduce với phép trừ cho kết quả sai khi đổi sang parallelStream()?
3. 3Gọi thao tác kết thúc lần thứ hai trên cùng một stream đã tiêu thụ sẽ dẫn tới điều gì?
4. 4Trong groupingBy(Person::city, counting()), vai trò của counting() là gì?

Tóm tắt

Stream không phải một tập hợp kiểu mới: nó không sở hữu phần tử, không sửa nguồn, chỉ dùng được một lần, và là bản mô tả một phép tính đặt trên nguồn có sẵn. Nắm đúng câu này là tự giải thích được gần hết các lỗi thường gặp.

Dây chuyền gồm các thao tác trung gian trả về Stream nên nối tiếp được, và một thao tác kết thúc đóng dây chuyền để sinh kết quả. Trung gian thì lười: chúng chỉ ghi nhận ý định, cả dây chuyền chỉ khởi động khi gặp thao tác kết thúc. Tính lười cho phép gộp nhiều phép vào một lượt duyệt duy nhất và dừng sớm khi đã đủ kết quả, nhờ đó stream vô hạn cộng limit mới chạy được.

Collectors lo việc gom, trong đó groupingBy chia ngăn theo khoá và giao cho bộ gom hạ nguồn quyết định mỗi ngăn thành cái gì. reduce tổng quát hơn nhưng đòi hai điều kiện dễ quên: giá trị khởi đầu phải trung hoà và phép gộp phải có tính kết hợp. Cuối cùng, parallelStream() không phải công tắc tăng tốc miễn phí.

Ghi nhớ

• Stream không lưu dữ liệu, không sửa nguồn, và dùng đúng một lần.
• Trung gian trả về Stream nên nối được (filter, map, sorted, distinct, limit); kết thúc đóng dây chuyền (collect, forEach, reduce, count).
• Lười: không có thao tác kết thúc thì không có gì chạy; nhờ đó limit chặn được stream vô hạn và dừng sớm được.
• Collectors gom kết quả: toList, joining, groupingBy cộng bộ gom hạ nguồn.
• parallelStream() chỉ đáng dùng khi dữ liệu lớn, phép tính nặng, hàm thuần, phép gộp có tính kết hợp.