[Advanced Java] Stream API trong Java: map, filter, reduce và collect

Bài này mở đầu Phần 3 của khoá học, phần nói về xử lý dữ liệu theo lối hiện đại. Hai phần trước bàn về cách dựng type và cách chọn collection; phần này bàn về việc bạn làm gì với một collection sau khi đã có nó, và nó bắt đầu bằng đúng cái API đã thay đổi cách người ta đọc code Java: stream.

Stream không phải collection, cũng không phải vòng lặp. Nó là mô tả của một phép tính trên một chuỗi phần tử, được dựng lên từng operation một rồi chỉ chạy đúng một lần, tất cả ở cuối, do một terminal operation duy nhất kích hoạt. Gần như mọi lỗi người ta gặp với stream đều bắt nguồn từ việc bỏ qua câu vừa rồi, nên bài này dành phần dài nhất để chứng minh nó bằng output xen kẽ thật thay vì chỉ khẳng định suông.

Mọi dòng output, mọi exception message và mọi con số allocation bên dưới đều được tạo ra bằng cách compile và chạy code trên OpenJDK 21.0.6 (arm64). Không có phép đo thời gian nào ở đây: chỗ nào chi phí quan trọng thì bài này đếm số operation hoặc số byte cấp phát, vì hai thứ đó tái lập được, còn một con số bấm giờ trên máy dùng chung thì không.

Điều kiện cần: đúng phần cú pháp lambda mà bài này dùng

Stream nhận behavior làm argument, và cách ngắn nhất để viết behavior là lambda. Bài kế tiếp sẽ nói về lambda cho ra ngô ra khoai. Mục này chỉ dạy đủ để đọc được code bên dưới, không hơn.

Một lambda gồm danh sách parameter, một mũi tên, và phần body. Nếu body là một expression duy nhất thì cứ viết expression đó, giá trị của nó chính là giá trị trả về:

n -> n.length() > 3            // one parameter, expression body, returns boolean
(a, b) -> a + b                // two parameters, expression body
w -> { return w.trim(); }      // block body, needs an explicit return
(a, b) -> { int s = a + b; return s; }

Ngoặc đơn quanh một parameter duy nhất là tuỳ chọn; với không parameter hoặc từ hai trở lên thì bắt buộc. Body dạng block cần ngoặc nhọn và một return tường minh, trừ khi lambda không trả về gì.

Khi body của lambda chỉ làm mỗi việc gọi một method có sẵn, method reference nói cùng ý đó mà bớt ồn hơn. Có bốn dạng, và cả bốn đều xuất hiện trong bài này:

Dạng thứ ba là dạng làm người ta bất ngờ: String::length nhận phần tử của stream rồi gọi length() trên nó, nên một function một argument lại được viết mà không thấy argument nào cả. Cùng quy tắc đó khiến String::startsWith có hai parameter — object nhận lời gọi và argument.

Function<String, Integer> f1 = Integer::parseInt;
Supplier<String>          f2 = "hello"::toUpperCase;
Function<String, Integer> f3 = String::length;
Function<String, StringBuilder> f4 = StringBuilder::new;
BiFunction<String, String, Boolean> f5 = String::startsWith;

static   42
bound    HELLO
unbound  6
ctor     StringBuilder len 2
unbound2 true

Điều kiện cần chỉ có vậy. Lambda được capture những gì từ scope bao quanh, vì sao biến bị capture phải là effectively final, compiler quyết định lambda trở thành interface nào bằng cách nào, và this bên trong lambda nghĩa là gì — bài kế tiếp lo hết. Bài sau nữa lo Supplier, Consumer, Function và Predicate với tư cách những type đáng biết riêng. Còn tới lúc đó, cứ đọc String::length là "function trả về độ dài của một string" rồi đi tiếp.

Stream là gì, và không phải là gì

Tài liệu package của chính JDK định nghĩa hình dạng đó trong một câu: một pipeline "consists of a source (such as a Collection, an array, a generator function, or an I/O channel); followed by zero or more intermediate operations such as Stream.filter or Stream.map; and a terminal operation such as Stream.forEach or Stream.reduce."

Ba phần, kèm hai quy tắc cứng: intermediate operation có thể không có hoặc có nhiều, còn terminal operation thì đúng một cái, và luôn nằm cuối.

Source, intermediate operation, terminal operation

List<String> names = List.of("alice", "bob", "carol", "dan", "erin");

List<String> out = names.stream()          // source     -> Stream<String>
        .filter(n -> n.length() > 3)       // intermediate -> Stream<String>
        .map(String::toUpperCase)          // intermediate -> Stream<String>
        .collect(Collectors.toList());     // terminal   -> List<String>

result [ALICE, CAROL, ERIN]
source [alice, bob, carol, dan, erin]

Mọi intermediate operation đều trả về một Stream, đó là lý do chúng nối chuỗi được. Terminal operation trả về thứ khác — một List, một long, một Optional, hoặc không gì cả — đó là lý do chuỗi dừng ở đấy.

Stream có thể đến từ nhiều nguồn, không chỉ từ collection:

Stream.of(1, 2, 3)                              // [1, 2, 3]
Arrays.stream(new String[]{"x", "y"})           // [x, y]
"abc".chars().mapToObj(c -> (char) c)           // [a, b, c]
Stream.iterate(1, x -> x * 2).limit(6)          // [1, 2, 4, 8, 16, 32]
Stream.iterate(1, x -> x < 40, x -> x * 3)      // [1, 3, 9, 27]
new TreeMap<>(m).entrySet().stream()            // [a, b]
                .map(Map.Entry::getKey)
Stream.empty()                                  // []

Stream.iterate với hai argument là vô hạn, và điều đó hợp lệ chính vì không có gì được tính cho tới khi có ai đó hỏi tới. Dạng ba argument thêm từ Java 9 mang sẵn điều kiện dừng của riêng nó.

Stream không phải collection

Từ đó suy ra ba tính chất, và cái nào cũng bẫy người ta.

Nó không giữ dữ liệu. names.stream() không copy list. Nó ghi lại chỗ cần đọc. Không có gì được cấp phát theo từng phần tử cho tới khi một terminal operation kéo phần tử đi qua.

Nó chỉ dùng được một lần. Sau khi terminal operation chạy xong, stream coi như hết. Chạm vào nó lần nữa không phải là không làm gì trong im lặng:

Stream<String> s = names.stream();
System.out.println("count  " + s.count());
s.forEach(System.out::println);

count  5
caught java.lang.IllegalStateException: stream has already been operated upon or closed

Vẫn message đó nếu bạn chỉ lưu một intermediate stream rồi dùng lại — Stream.of("a","b").map(String::toUpperCase) giữ trong một biến rồi tiêu thụ hai lần cũng ném đúng thứ ấy. Nếu bạn thật sự cần đi qua hai lượt, hãy giữ một Supplier rồi gọi nó hai lần:

Supplier<Stream<String>> sup = () -> Stream.of("a", "b", "c");
System.out.println("sup1 " + sup.get().count() + "  sup2 " + sup.get().toList());

sup1 3  sup2 [a, b, c]

Nó không sửa source của mình. Ở ví dụ đầu tiên bên trên, names in ra y nguyên sau khi một filter và một map đã chạy trên nó. Không operation nào của stream ghi ngược vào collection mà nó đọc.

Đảm bảo đó nói về collection, không nói về các object nằm trong đó. Một function truyền cho map mà sửa chính phần tử nó nhận được thì chắc chắn sẽ làm hỏng dữ liệu của bạn:

List<StringBuilder> src = new ArrayList<>(List.of(new StringBuilder("a"), new StringBuilder("b")));
List<StringBuilder> out = src.stream().map(sb -> sb.append("!")).toList();

source after mutating map [a!, b!]  out [a!, b!]  same objects true

List vẫn giữ đúng hai object nó vẫn luôn giữ — và giờ cả hai đọc ra a! và b!. map sinh ra giá trị mới từ giá trị cũ, chứ không phải sửa tại chỗ giá trị cũ.

Lazy: không gì chạy cho tới khi có terminal operation

Đây là ý quan trọng nhất trong cả API, và cũng là ý dễ tin mà khó thật sự hiểu nhất. Tài liệu nói thẳng — intermediate operation "are always lazy; executing an intermediate operation such as filter() does not actually perform any filtering" — nhưng hệ quả chỉ thấm khi bạn nhìn thứ tự các thứ thật sự chạy.

Pipeline không có terminal operation thì không chạy gì cả

List<String> words = List.of("apple", "fig", "banana", "kiwi", "cherry");

words.stream()
     .filter(w -> { System.out.println("filter " + w); return w.length() > 4; })
     .map(w -> { System.out.println("  map  " + w); return w.toUpperCase(); });
System.out.println("(nothing above this line)");

--- no terminal operation ---
(nothing above this line)

Năm phần tử, hai operation mỗi cái có một println, và không một dòng output nào. Hai lambda đó chưa từng được gọi. filter và map chỉ dựng nên mô tả về việc sẽ làm sau này, và không ai từng yêu cầu làm cả.

Trace xen kẽ

Giờ thêm một terminal operation và không đổi gì khác:

List<String> r = words.stream()
     .filter(w -> { System.out.println("filter " + w); return w.length() > 4; })
     .map(w -> { System.out.println("  map  " + w); return w.toUpperCase(); })
     .collect(Collectors.toList());

filter apple
  map  apple
filter fig
filter banana
  map  banana
filter kiwi
filter cherry
  map  cherry
result [APPLE, BANANA, CHERRY]

Hãy đọc output đó hai lần. Mô hình hiển nhiên trong đầu — filter xử lý cả năm phần tử, rồi map xử lý ba phần tử sống sót — lẽ ra phải cho ra năm dòng filter rồi mới tới ba dòng map. Nhưng thực tế không như vậy.

Thực tế là apple đi qua filter rồi lập tức đi qua map, trước khi fig được nhìn tới. Mỗi phần tử được kéo qua trọn cả pipeline trước khi phần tử tiếp theo được lấy ra. fig trượt predicate nên không có dòng map nào theo sau; banana đậu nên dòng map của nó tới ngay sau đó. Chỉ có một lượt đi qua source, chứ không phải một lượt cho mỗi stage, và không có list trung gian nào chứa các phần tử sống sót được dựng lên cả.

Đó là lý do một pipeline mười operation không phải là mười lần duyệt, và là lý do lazy là một tính chất cấu trúc chứ không phải một phép tối ưu. Nó cũng giải thích cái exception bạn nhận được khi sửa source giữa chừng, phần bẫy bên dưới sẽ quay lại chuyện này.

limit làm pipeline dừng sớm, nên phần tử sau không hề được đọc

Nếu terminal operation ngừng hỏi thì source ngừng sinh. Thêm .limit(2) vào đúng pipeline đó:

List<String> r2 = words.stream()
     .filter(w -> { System.out.println("filter " + w); return w.length() > 4; })
     .map(w -> { System.out.println("  map  " + w); return w.toUpperCase(); })
     .limit(2)
     .collect(Collectors.toList());

filter apple
  map  apple
filter fig
filter banana
  map  banana
result [APPLE, BANANA]

kiwi và cherry không sinh ra một dòng output nào hết. Chúng không bị filter, không bị map, không được đọc. Pipeline đã có đủ thứ nó cần sau banana và ngừng kéo.

findFirst cũng dừng sớm y như vậy. Tìm từ ba chữ cái đầu tiên trong pear, fig, apple, kiwi chỉ chạm tới hai phần tử rồi dừng:

map    pear
  filt PEAR
map    fig
  filt FIG
found FIG

Chính khả năng dừng sớm làm cho một source vô hạn dùng được. Stream.iterate(0, x -> x + 1).limit(5).toList() trả về [0, 1, 2, 3, 4] và kết thúc, vì source chỉ bị hỏi đúng năm phần tử. Bỏ limit đi thì cùng terminal operation đó sẽ không bao giờ trả về.

Không phải operation nào cũng để phần tử chảy qua

filter và map là stateless: mỗi phần tử được xử lý độc lập. sorted và distinct là stateful, và riêng sorted thì không thể phát ra thứ gì trước khi đã thấy hết. Đặt một println ở hai phía của sorted là thấy ngay dòng chảy từng phần tử bị gãy:

List<String> fruit = List.of("pear", "fig", "apple", "kiwi");

fruit.stream()
     .map(w -> { System.out.println("map    " + w); return w.toUpperCase(); })
     .sorted()
     .map(w -> { System.out.println("  after sort " + w); return w; })
     .toList();

map    pear
map    fig
map    apple
map    kiwi
  after sort APPLE
  after sort FIG
  after sort KIWI
  after sort PEAR

Cả bốn dòng map ra trước, vì sorted là một hàng rào: nó buffer trọn stream rồi mới nhả phần tử đầu tiên. Đó là chi phí thật đáng biết — một sorted nằm giữa pipeline nghĩa là cả stream bị vật chất hoá tại điểm đó, và nó cũng xoá sạch lợi ích của một limit phía sau đối với mọi thứ nằm trước nó.

Thứ tự operation quyết định khối lượng việc phải làm

Vì phần tử chảy qua trọn cả chuỗi, thứ tự bạn viết các operation quyết định mỗi lambda bị gọi bao nhiêu lần. Sáu từ, lọc xuống còn ba:

List<String> six = List.of("apple", "fig", "banana", "kiwi", "cherry", "plum");

// map first
six.stream().map(...).filter(...).toList();
// filter first
six.stream().filter(...).map(...).toList();

Đếm từ trace output: đặt map trước thì function mapping chạy 6 lần và predicate chạy 6 lần. Đặt filter trước thì predicate chạy 6 lần còn function mapping chỉ chạy 3 lần. Cùng kết quả, một nửa khối lượng mapping, và khác biệt nằm ở đúng một lần đổi chỗ hai dòng.

Đó là con số đếm operation, không phải benchmark. Nó chính xác, nó không phụ thuộc máy, và nó là cách trung thực để suy luận về chi phí của một pipeline: đặt các operation rẻ và có tính loại bớt lên trước.

map và filter

Hai con ngựa kéo này có hai nhiệm vụ bù trừ nhau, và phát biểu chính xác ra là hết bối rối:

• map đổi type của từng phần tử và không bao giờ đổi số lượng. Vào n phần tử thì ra n phần tử.
• filter đổi số lượng và không bao giờ đổi type. Vào n phần tử thì ra nhiều nhất n, tất cả cùng type.

record Person(String name, int age) {}

List<Person> people = List.of(new Person("Alice", 34), new Person("Bob", 17),
                              new Person("Carol", 52), new Person("Dan", 29));

people.stream().map(Person::name).toList();
people.stream().filter(p -> p.age() >= 18).map(Person::name).toList();
people.stream().map(Person::name).map(String::length).toList();

names    [Alice, Bob, Carol, Dan]
adults   [Alice, Carol, Dan]
lengths  [5, 3, 5, 3]
map size 4 from 4

map nối chuỗi thoải mái vì mỗi lần gọi chỉ đổi element type của stream: Stream<Person> thành Stream<String> thành Stream<Integer>. Argument của map là bất kỳ function nào đi từ element type hiện tại sang một type mới — một lambda, một method reference tới instance method (String::length), hoặc một method reference tới static method do bạn viết.

filter nhận một predicate: function trả về boolean. Nó giữ lại các phần tử mà predicate đúng. Nó không biến đổi được, và không thêm được.

Vài terminal operation đáng nhớ song song với hai cái trên:

Hai trong số đó có góc cạnh đáng biết. max trả về Optional vì stream rỗng thì không có giá trị lớn nhất; cứ gọi orElse rồi đi tiếp — một bài sau trong phần này sẽ nói Optional cho tử tế. Còn allMatch trên stream rỗng trả về true, đúng về logic mà vẫn khiến người ta ngớ ra lần đầu:

max        Carol
anyMatch   true
allMatch   false
noneMatch  true
allMatch empty true

reduce: gấp cả stream lại thành một giá trị

reduce gom stream về một giá trị duy nhất bằng cách lặp lại việc gộp hai giá trị thành một. Nó có ba overload, và chọn giữa chúng chủ yếu là chọn xem bạn muốn nhận về gì khi stream rỗng.

Ba overload

List<Integer> nums = List.of(3, 1, 4, 1, 5, 9, 2, 6);
List<String>  ws   = List.of("alpha", "be", "gamma", "hi");

Optional<Integer> a = nums.stream().reduce((x, y) -> x + y);              // 1 argument
int b              = nums.stream().reduce(0, (x, y) -> x + y);            // 2 arguments
int c              = ws.stream().reduce(0, (acc, w) -> acc + w.length(),  // 3 arguments
                                        Integer::sum);

1-arg sum        31
1-arg on empty   false -> 0
2-arg sum        31
2-arg product    6480
2-arg max        9
3-arg chars      14

Dạng một argument không có giá trị khởi đầu, nên stream rỗng thì chẳng có đáp án nào cả và nó trả về một Optional. Dạng hai argument nhận một identity, nên stream rỗng trả về đúng identity đó và kết quả là một giá trị thường. Dạng ba argument tồn tại cho trường hợp type tích luỹ khác type phần tử — gấp các phần tử String thành một tổng int — và nó cần function thứ ba để trộn hai kết quả bộ phận.

Vì sao identity phải thật sự là identity

Javadoc nói rất rõ: "The identity value must be an identity for the accumulator function. This means that for all t, accumulator.apply(identity, t) is equal to t."

Đó là yêu cầu chứ không phải lời khuyên, và vi phạm nó thì hỏng trong im lặng:

nums.stream().reduce(1, (x, y) -> x + y);            // identity 1 with addition
List.<Integer>of().stream().reduce(1, (x, y) -> x + y);

wrong identity 1 32 (should be 31)
wrong id, empty  1

Tổng trả về lớn hơn đúng một đơn vị, và stream rỗng trả về 1 thay vì 0. Không có gì ném ra cả. Identity đúng phụ thuộc vào operation: 0 cho phép cộng, 1 cho phép nhân, "" cho nối string, Integer.MIN_VALUE cho max. Nếu operation của bạn không có giá trị nào như vậy thì đó là dấu hiệu nên dùng overload một argument và xử lý trường hợp rỗng cho tường minh.

Vì sao accumulator phải có tính kết hợp

Kết hợp nghĩa là (a op b) op c bằng a op (b op c). Phép cộng thì có; phép trừ thì không. JDK không kiểm tra, và chạy tuần tự thì bạn sẽ không bao giờ nhận ra — khác biệt chỉ lộ ra khi các phần tử bị chia thành từng khối rồi gộp lại:

List<Integer> small = List.of(10, 3, 2);
small.stream().reduce(0, (x, y) -> x - y);
small.parallelStream().reduce(0, (x, y) -> x - y);
small.parallelStream().reduce(0, Integer::sum);

sub sequential   -15
sub parallel     -9
add parallel     15

Cùng một biểu thức cho -15 ở dòng này và -9 ở dòng kế tiếp. Không cái nào là bug của JDK: một accumulator không kết hợp đơn giản là không có đáp án xác định khi công việc được nhóm lại theo cách khác. Phép cộng, vốn có tính kết hợp, cho 15 ở cả hai cách.

Đúng cái bẫy đó nấp trong overload ba argument, nơi function thứ ba — combiner — trộn hai kết quả bộ phận. Một combiner sai thì vô hình khi chạy tuần tự, bởi vì nó không hề được gọi:

List<String> ws = List.of("alpha", "be", "gamma", "hi");
ws.stream().reduce(0, (acc, w) -> acc + w.length(), (x, y) -> x);          // broken combiner
ws.parallelStream().reduce(0, (acc, w) -> acc + w.length(), (x, y) -> x);

bad combiner seq 14
bad combiner par 5

Lần chạy tuần tự đúng do may. Lần chạy parallel vứt đi một nửa công việc. Nếu bạn viết reduce ba argument thì combiner phải trộn thật, và cách duy nhất để tập dượt nó là chạy parallel — cũng là một lý do tốt để chọn collect cho mọi thứ phức tạp hơn một con số.

collect và bộ Collectors

collect là terminal operation dùng để dựng một container. Trong thực tế bạn không bao giờ tự viết dạng ba function của nó; bạn truyền vào một collector làm sẵn từ java.util.stream.Collectors.

toList, và chỗ nó khác Stream.toList

Java 16 thêm Stream.toList(), đọc dễ hơn và không phải cùng một thứ:

List<String> a = Stream.of("a", "b").collect(Collectors.toList());
List<String> b = Stream.of("a", "b").toList();

Collectors.toList class java.util.ArrayList
Collectors.toList add   ok -> [a, b, c]
Stream.toList class     java.util.ImmutableCollections$ListN
Stream.toList add       throws java.lang.UnsupportedOperationException

Collectors.toList() cho ra một ArrayList sửa được — nhưng javadoc không đảm bảo class cụ thể nào, nên đừng dựa vào việc nó là ArrayList. Stream.toList() cho ra một list không sửa được và add ném exception.

Chuyện null là chỗ phần lớn bài viết nói sai. Stream.toList() không sửa được nhưng có nhận null; Collectors.toUnmodifiableList() thì không:

Collectors.toList null  [a, null]
Stream.toList null      [a, null]
toUnmodifiableList cls  java.util.ImmutableCollections$List12
toUnmodifiableList null throws NullPointerException

Mặc định dùng toList(). Chỉ dùng Collectors.toList() khi bạn thật sự cần sửa kết quả về sau, và dùng toUnmodifiableList() khi một null trong dữ liệu nên bị coi là bug.

toSet, toMap và chuyện trùng key

toSet() trả về một HashSet — không thứ tự, đã khử trùng lặp, và không đảm bảo thứ tự gì hết:

toSet        [Engineering, Sales, Support]
toSet class  java.util.HashSet

toMap nhận một function tạo key và một function tạo value. Nó chạy tốt cho tới khi hai phần tử sinh ra cùng một key, lúc đó nó ném exception chứ không âm thầm bỏ bớt một cái:

record Employee(String name, String dept, int salary) {}

STAFF.stream().collect(Collectors.toMap(Employee::name, Employee::salary));  // fine
STAFF.stream().collect(Collectors.toMap(Employee::dept, Employee::salary));  // two Engineering rows

toMap        {Dan=82000, Erin=64000, Bob=95000, Alice=120000, Carol=78000}
dup key      java.lang.IllegalStateException: Duplicate key Engineering (attempted merging values 120000 and 95000)

Message nêu cả key lẫn hai value, hiếm có runtime failure nào tử tế đến vậy. Cách sửa là overload ba argument, mà function thứ ba quyết định làm gì khi đụng độ:

STAFF.stream().collect(Collectors.toMap(Employee::dept, Employee::salary, Integer::sum));
STAFF.stream().collect(Collectors.toMap(Employee::dept, Employee::salary, Integer::sum, TreeMap::new));

merge fn      {Engineering=215000, Sales=160000, Support=64000}
4-arg TreeMap {Engineering=215000, Sales=160000, Support=64000}  java.util.TreeMap

Argument thứ tư cung cấp chính cái map, đó là cách bạn lấy về TreeMap hoặc LinkedHashMap thay cho HashMap mặc định.

Còn một góc sắc nữa: toMap từ chối value null bằng NullPointerException, vì nó được cài đặt bằng Map.merge. groupingBy chấp nhận phần tử có field giá trị null nhưng từ chối key null, cũng bằng NullPointerException.

groupingBy, partitioningBy và downstream collector

groupingBy nhận một classifier function và trả về map từ key sang list các phần tử có key đó:

STAFF.stream().collect(Collectors.groupingBy(Employee::dept));

Argument thứ hai mới là chỗ có sức mạnh thật. Nó là một collector khác — downstream collector — và nó nhận các phần tử của từng group. Nó sinh ra cái gì thì cái đó trở thành value của key ấy:

Collectors.groupingBy(Employee::dept, Collectors.mapping(Employee::name, Collectors.toList()));
Collectors.groupingBy(Employee::dept, Collectors.counting());
Collectors.groupingBy(Employee::dept, Collectors.summingInt(Employee::salary));
Collectors.groupingBy(Employee::dept, Collectors.averagingDouble(Employee::salary));
Collectors.groupingBy(Employee::dept, TreeMap::new, Collectors.summingInt(Employee::salary));

groupingBy   {Engineering=[Alice, Bob], Sales=[Carol, Dan], Support=[Erin]}
counting     {Engineering=2, Sales=2, Support=1}
summingInt   {Engineering=215000, Sales=160000, Support=64000}
averaging    {Engineering=107500.0, Sales=80000.0, Support=64000.0}
3-arg group  {Engineering=215000, Sales=160000, Support=64000}  java.util.TreeMap

Bốn type value khác nhau từ một classifier duy nhất, hoàn toàn do downstream collector quyết định. Mặc định đúng là Collectors.toList() — viết groupingBy(f) và groupingBy(f, Collectors.toList()) cho cùng kết quả.

partitioningBy là groupingBy bị giới hạn vào một predicate, nên key type là Boolean và luôn có đúng hai key:

STAFF.stream().collect(Collectors.partitioningBy(e -> e.salary() >= 90000,
                       Collectors.mapping(Employee::name, Collectors.toList())));

partitioning {false=[Carol, Dan, Erin], true=[Alice, Bob]}
part class   java.util.stream.Collectors$Partition

Cái "luôn đúng hai" mới là điểm mấu chốt, và nó lộ ra trên stream rỗng:

part empty   {false=[], true=[]}
group empty  {}

groupingBy cho map rỗng vì không phần tử nào từng cung cấp một key. partitioningBy cho cả hai key với list rỗng, vì tập key của nó do type ấn định chứ không do dữ liệu. Khác biệt đó là lý do partitioningBy(p).get(true) thì an toàn còn groupingBy(f).get(k) có thể là null.

Các collector đáng biết:

Primitive stream: IntStream, LongStream và DoubleStream

Stream<Integer> và IntStream là hai type khác nhau, và khác biệt không phải chuyện thẩm mỹ. Stream<Integer> giữ reference, nên mỗi int đi vào nó đều phải được box thành một object Integer. IntStream giữ thẳng giá trị int.

IntStream.rangeClosed(1, 10)                       // an IntStream
nums.stream().mapToInt(Integer::intValue)          // Stream<Integer> -> IntStream
IntStream.range(0, 5).boxed()                      // IntStream -> Stream<Integer>
IntStream.of(1, 2, 3).mapToObj(i -> "#" + i)       // IntStream -> Stream<String>
IntStream.of(1, 2, 3).asDoubleStream()             // IntStream -> DoubleStream

mapToInt sum   55
boxed          [0, 1, 2, 3, 4]
mapToObj       [#1, #2, #3]
asDoubleStream 6.0
LongStream sum 500000500000
IntStream max  9

Các primitive stream mang theo phần số học mà Stream không có: sum(), average(), max(), min() và summaryStatistics(), tính hết trong một lượt.

IntSummaryStatistics st = nums.stream().mapToInt(Integer::intValue).summaryStatistics();

stats          IntSummaryStatistics{count=10, sum=55, min=1, average=5.500000, max=10}
average        2.5
average empty  NaN

sum() trả về int hoặc long thường, vì tổng của rỗng là 0. average() trả về OptionalDouble, vì trung bình của rỗng không phải 0.

Vì sao chúng tồn tại, tính bằng byte

Lý lẽ cho primitive stream là allocation, và allocation thì đếm chính xác được. Cộng một triệu giá trị theo hai cách, với ThreadMXBean.getThreadAllocatedBytes đọc bộ đếm allocation của thread ở hai đầu:

IntStream.range(0, n).asLongStream().sum();                        // no boxing
IntStream.range(0, n).boxed().mapToLong(Integer::longValue).sum(); // one Integer per element

IntStream, no boxing   sum 499999500000   allocated 240 bytes
boxed() then unbox     sum 499999500000   allocated 15998280 bytes
bytes per element                            15.99828

Pipeline không boxing cấp phát tổng cộng 240 byte — vài object của bộ máy pipeline, và không gì theo từng phần tử. Pipeline có boxing cấp phát gần 16 MB cho cùng đáp án, tức 16 byte cho mỗi phần tử.

Con số 16 không phải ngẫu nhiên. Trên JVM 64-bit với compressed object pointer, một Integer là header object 12 byte cộng một field int 4 byte. Và tổng số tự giải thích chính nó rất khớp: Integer.valueOf cache các giá trị từ -128 tới 127, nên 128 trong một triệu lấy sẵn từ cache mà không tốn gì, còn 999.872 object Integer mới ở 16 byte mỗi cái là 15.997.952 byte — chừa lại 328 byte overhead cố định của pipeline. Cả hai lần chạy chương trình cho ra con số giống hệt nhau tới từng byte.

Đó là thứ mapToInt mua cho bạn: không phải một tỷ lệ phần trăm, mà là một object cho mỗi phần tử vĩnh viễn không được tạo ra. Cộng lương của năm employee thì chuyện này không đáng nói. Cộng một triệu thì đó là một triệu object để garbage collector phải tìm và dọn.

Các bẫy

Lambda có state

Javadoc cảnh báo rằng "stream pipeline results may be nondeterministic or incorrect if the behavioral parameters to the stream operations are stateful". Ví dụ kinh điển là một bộ đếm dùng để đánh số các phần tử:

List<String> letters = List.of("a", "b", "c", "d", "e", "f", "g", "h");

int[] i = {0};
List<String> seq = letters.stream().map(w -> (i[0]++) + ":" + w).toList();

int[] j = {0};
List<String> par = letters.parallelStream().map(w -> (j[0]++) + ":" + w).toList();

sequential [0:a, 1:b, 2:c, 3:d, 4:e, 5:f, 6:g, 7:h]
parallel   [4:a, 7:b, 6:c, 5:d, 3:e, 2:f, 1:g, 0:h]
run 0      [6:a, 5:b, 3:c, 1:d, 3:e, 2:f, 4:g, 0:h]  counter=7
run 1      [4:a, 6:b, 7:c, 5:d, 3:e, 2:f, 1:g, 0:h]  counter=8

Nhìn run 0: chỉ số 3 xuất hiện hai lần, và bộ đếm dừng ở 7 sau tám phần tử. Đó là một lost update trên j[0]++, vốn không phải thao tác atomic. Các output parallel này là giá trị của một lần chạy và sẽ khác trên máy bạn — và đó chính là vấn đề. Kết quả tuần tự thì tình cờ đúng, và đó đúng là lý do bug kiểu này lọt ra production.

Cái mẹo mảng-một-phần-tử để bộ đếm compile được đã là dấu hiệu có gì đó sai. Nếu bạn cần chỉ số, hãy dùng IntStream.range trên các chỉ số thay vì lén nhét một biến đếm vào lambda.

Sửa source trong lúc stream đang chạy

Vì phần tử được kéo ra từ source từng cái một trong lúc terminal operation chạy, source vẫn sống suốt cả pipeline. Ghi vào nó giữa chừng thì hỏng đúng như vòng lặp for nâng cao hỏng:

List<String> src = new ArrayList<>(List.of("a", "b", "c", "d"));
src.stream().map(s -> { if (s.equals("b")) src.add("x"); return s.toUpperCase(); }).toList();

java.util.ConcurrentModificationException
   at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1714)
   at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
   at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
   at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:575)
   at java.base/java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
   at java.base/java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:616)
   at java.base/java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:622)
   at java.base/java.util.stream.ReferencePipeline.toList(ReferencePipeline.java:627)
   at Traps.main(Traps.java:9)

Frame trên cùng là ArrayListSpliterator.forEachRemaining — đúng phép kiểm tra modCount đứng sau iterator của ArrayList, chỉ là đi qua bộ máy stream. Chú ý là exception bật ra từ toList(), tức terminal operation, chứ không phải từ lời gọi map đã ghi vào list. Hãy collect vào một list mới rồi thay list cũ.

peek là để debug, và có thể bị bỏ qua hoàn toàn

peek là intermediate operation chạy một action trên từng phần tử rồi cho phần tử đi tiếp nguyên vẹn. Javadoc của nó nói nó "exists mainly to support debugging", và JDK nói thẳng là không đảm bảo nó sẽ chạy:

Stream.of("a", "b", "c").peek(s -> System.out.println("peek " + s)).count();
Stream.of("a", "b", "c").peek(s -> System.out.println("peek " + s)).filter(s -> true).count();

peek + count:
count 3
peek + filter + count:
peek a
peek b
peek c
count 3

Pipeline đầu tiên không in gì cả. count() trên một source biết trước kích thước và không có operation nào làm đổi số phần tử thì được tính thẳng từ kích thước source mà không duyệt gì hết, và javadoc nói thẳng: khi đó "no source elements will be traversed and no intermediate operations will be evaluated". Chèn một filter vào — thứ có thể làm đổi số lượng — thì phép tối ưu không còn áp dụng nữa, nên peek chạy.

Dùng peek để nhìn vào pipeline khi bạn đang debug. Đừng bao giờ dùng nó để làm việc thật.

findFirst so với findAny

Cả hai đều trả về Optional và cả hai đều dừng sớm. findFirst trả về phần tử đầu tiên theo encounter order. findAny được tự do trả về bất kỳ phần tử khớp nào, nhờ đó một pipeline parallel trả về thứ mà worker nào tìm ra trước.

findFirst seq 3
findAny   seq 3
findAny   par 15 15 15 15 15 15 15 15
findFirst par 3 3 3 3 3 3 3 3

Chạy tuần tự thì cả hai trả về 3. Chạy parallel, findAny trả về 15 ở cả tám lần chạy tại đây còn findFirst trả về 3 ở cả tám lần. Đừng coi sự ổn định đó là một đảm bảo: findAny được đặc tả là "any", nên một JVM khác, một kích thước source khác, hay một ngày khác đều được phép cho bạn thứ khác. Dùng findFirst khi thứ tự quan trọng, dùng findAny chỉ khi nó thật sự không quan trọng.

parallelStream là cái bẫy, không phải công tắc tăng tốc

parallelStream() chỉ là một lời gọi method, nên trông như một bản nâng cấp miễn phí. Không phải vậy. Ba thứ vỡ ngay khi bạn gõ nó ra, và cả ba đều đã xuất hiện trong các output bên trên.

Giả định về thứ tự vỡ. forEach không đảm bảo thứ tự nào trong một pipeline parallel:

parallel forEach        0 3 6 7 4 9 2 1 5 8
parallel forEachOrdered 0 1 2 3 4 5 6 7 8 9
parallel map+toList     [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

forEachOrdered khôi phục thứ tự và đánh đổi phần lớn tính song song để làm được điều đó. Chú ý toList() vẫn giữ đúng thứ tự — collect bảo toàn encounter order kể cả khi parallel — nên chỗ nguy hiểm nằm đúng ở các terminal operation có side effect.

Giả định về thread safety vỡ. Một collection không đồng bộ bị ghi từ một forEach parallel cho ra rác, và hai lần rác không giống nhau:

List<Integer> sink = new ArrayList<>();
IntStream.range(0, 10000).parallel().forEach(sink::add);

sink threw java.lang.ArrayIndexOutOfBoundsException
sink size 2213 (expected 10000)
sink size 5358 (expected 10000)
sink size 6882 (expected 10000)
sink size 8376 (expected 10000)

Năm lần chạy cùng một đoạn code: một lần ném exception, bốn lần âm thầm mất từ 16% tới 78% dữ liệu. Không có gì cảnh báo. Bản dùng collect của cùng pipeline đó thì an toàn, và đó là một phần lớn lý do collect tồn tại.

Giả định về tính đúng đắn vỡ, như phần reduce không kết hợp và phần lambda có state bên trên đã cho thấy.

Câu hỏi về hiệu năng là câu nên trả lời sau cùng, và trả lời cho trung thực: chạy song song tốn công để chia source, giao từng khối cho ForkJoinPool chung, rồi trộn kết quả lại, và với một pipeline nhỏ hoặc rẻ trên mỗi phần tử thì phần overhead đó có thể lấn át tất cả. Việc nó có thắng với dữ liệu của bạn trên phần cứng của bạn hay không là chuyện đo đạc, không phải chuyện đoán — và đo cho đàng hoàng nghĩa là dùng JMH, có warm-up, có fork JVM riêng và có blackhole, bởi vì một vòng lặp bấm giờ ngây thơ trên JVM đo JIT compiler nhiều ngang đo code của bạn. Bài này không có con số thời gian nào chính vì lý do đó. Đừng với tay tới parallelStream() khi chưa có một phép đo như vậy.

Khi nào không nên dùng stream

Stream hợp rất tệ với ba dạng vòng lặp, và ép dùng nó cho ra code tệ hơn chính vòng lặp mà bạn đang né.

Vòng lặp có chỉ số đơn giản. Nếu bạn cần chỉ số thì vòng lặp for đã có sẵn nó rồi. Bản stream phải tự chế ra một cái:

String[] a = {"x", "y", "z"};
IntStream.range(0, a.length).mapToObj(i -> i + "=" + a[i]).toList();  // [0=x, 1=y, 2=z]

Cách đó ổn khi bạn thật sự muốn một list các chuỗi chỉ số kèm giá trị. Còn dùng nó để thay cho for (int i = 0; i < a.length; i++) thì dài hơn, đọc chậm hơn, và chỉ chạy được nếu a là effectively final.

Vòng lặp có sửa đổi. Dồn state lại bằng phép gán chính là việc mà vòng lặp sinh ra để làm. Một stream có lambda ghi ra biến bên ngoài chính là cái bẫy lambda-có-state khoác áo khác.

Vòng lặp dừng sớm và cần giữ lại state sau đó. Stream dừng sớm được, nhưng không thể trao cho bạn các biến của vòng lặp tại đúng thời điểm dừng:

int[] data = {4, 8, 15, 16, 23, 42};

int runningTotal = 0, stoppedAt = -1;
for (int i = 0; i < data.length; i++) {
    runningTotal += data[i];
    if (runningTotal > 40) { stoppedAt = i; break; }
}

loop total 43 index 3

Hai giá trị lấy ra từ một lượt duyệt, với điều kiện phụ thuộc vào state đã tích luỹ. Mọi phiên bản stream của đoạn này đều tệ hơn.

Quy tắc sống sót được khi va vào code thật: dùng stream khi phép tính là một biến đổi từ một chuỗi sang một giá trị hoặc một chuỗi khác, và dùng vòng lặp khi nó là một thủ tục có state. Trộn hai thứ — một stream mà lambda mang state thủ tục — thì bạn nhận đủ nhược điểm của cả hai.

Câu hỏi thường gặp

Stream có lưu các phần tử của nó không?

Không. Stream không giữ dữ liệu của riêng nó; nó ghi lại chỗ cần đọc và việc cần làm với từng phần tử. Đó là lý do tạo một stream trên list một triệu phần tử gần như không cấp phát gì, và là lý do công việc chỉ xảy ra khi một terminal operation kéo phần tử đi qua.

Vì sao stream của tôi ném IllegalStateException?

Vì bạn đã dùng nó hai lần. Stream bị tiêu thụ bởi terminal operation của nó, và mọi lời gọi tiếp theo đều cho java.lang.IllegalStateException: stream has already been operated upon or closed. Chuyện này hay xảy ra nhất khi một intermediate stream được lưu vào biến rồi truyền qua truyền lại. Nếu cần hai lượt, hãy lưu một Supplier dựng ra stream mới rồi gọi nó hai lần.

Collectors.toList() và Stream.toList() khác nhau ra sao?

Stream.toList(), thêm từ Java 16, trả về list không sửa được nhưng có nhận null. Collectors.toList() trả về list mà trên JDK này tình cờ là một ArrayList sửa được, dù đặc tả không hứa gì về class. Collectors.toUnmodifiableList() không sửa được và ném NullPointerException khi gặp phần tử null. Mặc định hãy dùng Stream.toList().

Vì sao đặt println trong map mà không in ra gì?

Vì pipeline chưa có terminal operation. Intermediate operation là lazy: map trả về một stream mới và gọi function của bạn không lần nào. Thêm collect, forEach, count hay bất kỳ terminal operation nào thì cả chuỗi sẽ chạy.

reduce có chậm hơn vòng lặp for không?

Đó là câu hỏi không nên trả lời bằng một con số, và bất kỳ con số nào đo trên máy dùng chung cũng vô giá trị. Thứ đếm được là allocation: Stream<Integer> box một Integer cho mỗi phần tử, đo bên trên là 16 byte mỗi cái, còn một vòng for trên primitive hay một IntStream thì không box gì. Nếu một pipeline chạy nóng, đặt mapToInt trước phần số học là thay đổi chắc chắn có tác dụng. Còn lại đều cần một benchmark thật dưới JMH.

Có nên dùng parallelStream cho nhanh hơn không?

Gần như chắc chắn là không, và tuyệt đối không khi chưa đo. Nó phá thứ tự trong forEach, phơi bày mọi collection không thread-safe mà bạn ghi vào, và đòi accumulator của reduce phải thật sự có tính kết hợp — cả ba kiểu hỏng đều đã được minh hoạ bên trên bằng output thật. Đo cho trung thực thì cần JMH, không phải một vòng lặp bấm giờ.

filter có xoá phần tử khỏi list của tôi không?

Không. Không operation nào của stream ghi ngược lại source; collection ban đầu không đổi sau khi pipeline chạy. Cách duy nhất để dữ liệu của bạn đổi qua một stream là đưa cho map một function sửa chính object nó nhận được, và bạn không nên làm thế. Muốn xoá phần tử khỏi collection tại chỗ thì dùng removeIf.

Kết luận

Stream là một kế hoạch, không phải một container. Nó được dựng từ một source, không hoặc nhiều intermediate operation lazy, và đúng một terminal operation eager, và cho tới khi terminal operation đó chạy thì chưa có gì xảy ra cả. Trace xen kẽ trong bài này là trọn mô hình gói trong tám dòng output: mỗi phần tử đi hết cả pipeline trước khi phần tử kế tiếp được lấy ra, đó là lý do một lượt duyệt phủ hết mọi stage, lý do limit và findFirst để lại những phần tử không hề bị chạm tới, và lý do một source vô hạn vẫn dùng được.

Mọi thứ còn lại đều suy ra từ đó. map và filter mô tả việc làm trên từng phần tử; reduce gấp lại với một identity phải thật sự là identity và một accumulator phải thật sự có tính kết hợp; collect dựng nên container, với groupingBy cùng một downstream collector lo được phần lớn những gì người ta phải nhờ tới SQL. IntStream tồn tại để một triệu con số không biến thành một triệu object. Còn peek, findAny và parallelStream đều sắc hơn vẻ ngoài của chúng, vì cái nào cũng được đặc tả lỏng hơn cái tên của nó gợi ra.

Stream dựa vào lambda ở mọi bước, và bài này cố ý chỉ dạy đủ lambda để đọc được code. Bài kế tiếp sửa chuyện đó: bài 15 nói về lambda expression cho đàng hoàng — các dạng viết, lambda capture những gì từ scope bao quanh, vì sao biến bị capture phải là effectively final, compiler quyết định lambda trở thành interface nào bằng cách nào, và this bên trong lambda nghĩa là gì.