Hình dung việc nấu một nồi phở lớn. Có một cái nồi (supplier), bạn đổ dần từng thứ vào (accumulator), rồi múc ra bát (finisher). Giờ giả sử đông khách quá, bạn chia ra hai bếp cùng nấu — lúc đó mới cần thêm một bước: trộn hai nửa nồi lại thành một (combiner). Điều đáng chú ý: khi chỉ một bếp nấu, cái bước trộn đó không bao giờ được dùng tới. Bạn có viết nó sai bét cũng chẳng ai biết — cho tới ngày thuê thêm bếp thứ hai. Collector của Java có đúng bốn vai ấy, và cái bẫy lớn nhất nằm ở bước trộn mà phần lớn thời gian nằm im.

Collectors có sẵn hơn bốn mươi hàm, và chúng ghép được vào nhau như bài hôm qua đã thấy. Nên tự viết một Collector là chuyện hiếm.

Nhưng hiểu cấu tạo của nó thì đáng, vì ba lý do: bạn sẽ đọc được mã người khác, biết chọn giữa collect và reduce, và — quan trọng nhất — hiểu vì sao có một hàm mà 99% thời gian không bao giờ chạy, rồi bỗng nhiên chạy khi ai đó đổi một chữ.

Bốn hàm

Collector.of(
    () -> new Gom(),        // 1. supplier   — tạo thùng chứa rỗng
    Gom::them,              // 2. accumulator — bỏ một phần tử vào thùng
    Gom::gop,               // 3. combiner   — gộp hai thùng làm một
    Gom::ketQua);           // 4. finisher   — biến thùng thành kết quả cuối

Đọc theo vòng đời: tạo thùng, đổ dần từng phần tử vào, (có thể) gộp nhiều thùng, rồi chốt kết quả.

Cho một collector thống kê chuỗi, chạy tuần tự:

      supplier   : tạo thùng chứa
      accumulator: thêm 'java'
      accumulator: thêm 'go'
      accumulator: thêm 'kotlin'
      accumulator: thêm 'c'
      accumulator: thêm 'python'
  5 chuỗi, tổng dài 19, ngắn nhất 'c', dài nhất 'kotlin'

Một thùng, năm lần đổ vào, không có combiner.

Combiner chỉ xuất hiện khi song song

Đổi stream() thành parallelStream(), không sửa gì khác:

      supplier   : tạo thùng chứa
      accumulator: thêm 'kotlin'
      supplier   : tạo thùng chứa
      accumulator: thêm 'java'
      supplier   : tạo thùng chứa
      accumulator: thêm 'c'
      supplier   : tạo thùng chứa
      accumulator: thêm 'python'
      supplier   : tạo thùng chứa
      accumulator: thêm 'go'
      combiner   : gộp hai nhóm (1 + 1)
      combiner   : gộp hai nhóm (1 + 1)
      combiner   : gộp hai nhóm (1 + 2)
      combiner   : gộp hai nhóm (2 + 3)

Năm thùng riêng cho năm phần tử, rồi gộp dần theo cây: (1+1), (1+1), (1+2), (2+3).

Đây là điều đáng nhớ nhất của bài:

Combiner viết sai vẫn cho kết quả đúng trong mọi test chạy tuần tự. Nó chỉ lộ ra khi có ai đó đổi stream() thành parallelStream() — có thể là người khác, nhiều tháng sau, trong một lần tối ưu. Và lúc đó kết quả sai sẽ rất khó truy ngược về nguyên nhân.

Nên nếu tự viết collector, hãy luôn viết một test chạy song song, kể cả khi bạn không định dùng song song.

Combiner phải kết hợp được

Ràng buộc toán học: phép gộp phải kết hợp — (a+b)+c bằng a+(b+c) — vì thứ tự gộp cây không đoán trước được.

Cộng, nhân, lấy min, lấy max, hợp tập hợp: đều kết hợp. Trừ và chia thì không.

Với thống kê ở trên, mọi phép đều kết hợp: cộng số lượng, cộng tổng độ dài, lấy chuỗi ngắn nhất giữa hai bên. Nên nó an toàn.

Một cái bẫy tinh vi: nếu bạn giữ "phần tử đầu tiên thoả điều kiện" thì phép gộp không còn kết hợp theo nghĩa thông thường — kết quả phụ thuộc thứ tự. Với stream có thứ tự thì Java vẫn gộp đúng chiều, nhưng đây là chỗ dễ viết sai.

Cách viết ngắn hơn

Không cần khai một lớp riêng nếu đã có sẵn thứ đóng vai thùng chứa:

Collector<String, ?, String> noiHoa = Collector.of(
    StringBuilder::new,
    (sb, s) -> sb.append(s.toUpperCase()).append(" "),
    StringBuilder::append,
    sb -> sb.toString().trim());

  JAVA GO KOTLIN C PYTHON

StringBuilder::append vừa vặn làm combiner vì sb1.append(sb2) trả về sb1.

Chữ ? ở kiểu giữa nghĩa là "kiểu thùng chứa, người dùng không cần biết" — và đó là cách khai đúng khi trả collector ra ngoài.

IDENTITY_FINISH: bỏ được hàm thứ tư

Khi thùng chứa chính là kết quả cuối, finisher chỉ là hàm đồng nhất, và bạn khai luôn đặc tính đó:

Collector<String, ?, List<String>> vaoList = Collector.of(
    ArrayList::new,
    List::add,
    (a, b) -> { a.addAll(b); return a; },
    Collector.Characteristics.IDENTITY_FINISH);

  [java, go, kotlin, c, python]

Ba đặc tính có thể khai:

IDENTITY_FINISH — thùng chứa là kết quả, Stream bỏ qua bước finisher.

UNORDERED — kết quả không phụ thuộc thứ tự phần tử. Cho phép Stream tối ưu mạnh hơn khi song song. Đúng với toSet, sai với toList.

CONCURRENT — nhiều luồng ghi vào cùng một thùng chứa được, nên không cần combiner. Chỉ khai khi thùng chứa thật sự an toàn luồng — ConcurrentHashMap chẳng hạn. Khai sai thì dữ liệu hỏng lặng lẽ.

collect khác reduce chỗ nào

Cả hai đều gom stream thành một giá trị, nhưng:

reduce làm việc với giá trị bất biến — mỗi bước tạo ra một giá trị mới.

collect làm việc với thùng chứa thay đổi được — mỗi bước đổ thêm vào cùng một đối tượng.

Nên nối chuỗi bằng reduce((a,b) -> a+b) tạo ra n chuỗi trung gian và chậm bậc hai, đúng bài học ở bài String. Còn collect(joining()) chỉ dùng một StringBuilder.

Quy tắc: kết quả là số hoặc giá trị nhỏ thì reduce; kết quả là collection hoặc chuỗi thì collect.

Khi nào thật sự nên tự viết

Thành thật thì hiếm, vì ghép sẵn phủ gần hết:

  bằng teeing: 5 chuỗi, tổng dài 19

Cùng bài toán thống kê ở đầu bài, teeing giải quyết trong ba dòng mà không cần lớp nào.

Bốn trường hợp tôi thấy đáng tự viết:

Khi ghép sẵn phải duyệt nhiều lần mà nguồn chỉ đọc được một lần.

Khi cần một cấu trúc kết quả riêng — cây, đồ thị, chỉ mục nghịch đảo.

Khi cần tối ưu bộ nhớ — gom trực tiếp vào cấu trúc đích thay vì qua List trung gian.

Khi muốn đóng gói một phép gom phức tạp dùng lại ở nhiều nơi, để chỗ gọi chỉ còn một tên có nghĩa.

Trường hợp cuối là lý do chính đáng nhất: collect(thongKeDonHang()) đọc rõ hơn một chuỗi groupingBy lồng ba tầng, dù bên trong nó chính là chuỗi đó.

Muốn tự tay thấy cái bẫy combiner thì thử trong ba mươi giây: viết một collector đơn giản với Collector.of, cho combiner ném UnsupportedOperationException. Chạy với stream() — không sao; chạy với parallelStream() — nổ ngay. Ba mươi giây đó cho thấy vì sao một collector viết sai có thể sống yên trong dự án nhiều năm, rồi hỏng vào đúng ngày ai đó thêm một chữ parallel.

Mẫu số chung

Cấu trúc "gom dần từng phần tử rồi gộp các nửa kết quả lại" không phải phát minh của Java Stream — nó là phép gấp (fold/reduce), một trong những phép nền tảng nhất của lập trình hàm, và cái ràng buộc "combiner phải kết hợp được" chính là chìa khoá mở ra cả một thế giới tính toán phân tán.

  • JavaScript (Array.reduce), Python (functools.reduce), Haskell (foldl/foldr) đều có phép gấp tuần tự — chỉ cần một hàm hai ngôi, không cần kết hợp, vì nó chạy trái-sang-phải một mạch. Đó là accumulator của ta khi không song song.
  • Khác biệt sinh ra đúng lúc chia việc: Rust với thư viện rayon có par_iter().reduce(identity, op) — và tài liệu nói thẳng op phải kết hợp, y hệt combiner của ta. Gộp song song luôn cần một phần tử đơn vị (identity) và một phép gộp kết hợp, còn gấp tuần tự thì không.
  • Chỗ sâu nhất là MapReduce và Apache Spark: toàn bộ mô hình xử lý dữ liệu lớn dựa trên đúng ý này — hàm reduce phải kết hợp và giao hoán thì mới chia ra hàng nghìn máy rồi gộp lại được. Cái bẫy "combiner sai không lộ khi chạy một máy, hỏng khi chạy phân tán" của Spark giống hệt "không lộ với stream(), hỏng với parallelStream()" — chỉ khác quy mô là một máy hay nghìn máy.

Sợi chỉ chung đáng mang theo: gấp tuần tự và gấp song song nhìn giống nhau nhưng đòi hỏi khác nhau — gấp một mạch chỉ cần một phép hai ngôi, còn gấp chia-rồi-gộp cần phép đó kết hợp được (và thường cần một phần tử đơn vị). Đây là ranh giới mà mọi hệ thống từ Java Stream tới Spark đều phải tôn trọng, và là lý do khi viết bất kỳ phép gộp nào định chạy song song, câu hỏi sống còn luôn là: nếu tôi chia dữ liệu thành hai nửa bất kỳ, gộp hai kết quả nửa lại, có ra cùng đáp số không? Trả lời "không" thì phép gộp của bạn là một quả bom hẹn giờ, chờ ngày ai đó bật song song.

Ngày mai: stream song song — khi nào nhanh, khi nào chậm hơn tuần tự, và đo bằng JMH thay vì đoán.