99 Ngày Java — Ngày 60: Stream thực chiến

Chín ngày qua ta gom đủ mảnh: lambda, functional interface, method reference, rồi pipeline lười với các phép trung gian, phép kết thúc, Collectors, Optional và stream song song. Hôm nay ghép chúng vào một bài toán thật: bảng đơn hàng nhiều dòng hàng, tiền là BigDecimal, cần doanh thu theo khách theo tháng, top 3 đơn mỗi khách, sản phẩm bán chạy, và một báo cáo in ra từ file CSV. Quan trọng không kém là biết lúc nào stream đã thành khó đọc — và ba cách cứu: đặt tên collector, dùng record cho kết quả trung gian, hoặc quay về vòng for.

groupingBy đa cấp trên dữ liệu thật

Dữ liệu thật hiếm khi phẳng: một đơn hàng có nhiều dòng hàng, mỗi dòng có sản phẩm, số lượng, đơn giá. Năm công thức dưới đây giải quyết gần hết câu hỏi thống kê thường gặp — gom hai cấp khách rồi tháng, flatMap xuống dòng hàng trước khi gom, collectingAndThen để cắt top N sau khi gom, teeing lấy min và max trong một lần duyệt, và toMap có hàm merge cùng LinkedHashMap để vừa xử lý trùng khóa vừa giữ thứ tự.

// DỮ LIỆU THỰC TẾ — đơn hàng, mỗi đơn nhiều dòng hàng. Tiền dùng BigDecimal, KHÔNG BAO GIỜ double.
record DongHang(String sanPham, int soLuong, BigDecimal donGia) {
    BigDecimal thanhTien() { return donGia.multiply(BigDecimal.valueOf(soLuong)); }
}
enum TrangThai { MOI, DA_THANH_TOAN, DA_HUY }
record DonHang(String ma, String khachHang, TrangThai trangThai, LocalDate ngay, List<DongHang> dongHang) {
    BigDecimal tong() {
        return dongHang.stream().map(DongHang::thanhTien).reduce(BigDecimal.ZERO, BigDecimal::add);
    }
}

// ① DOANH THU THEO KHÁCH → THEO THÁNG — groupingBy HAI CẤP, TreeMap để tháng có thứ tự
Map<String, Map<YearMonth, BigDecimal>> doanhThu = donHang.stream()
    .filter(d -> d.trangThai() == TrangThai.DA_THANH_TOAN)       // ⭐ lọc TRƯỚC, gom SAU
    .collect(groupingBy(DonHang::khachHang,
             groupingBy(d -> YearMonth.from(d.ngay()), TreeMap::new,
                        reducing(BigDecimal.ZERO, DonHang::tong, BigDecimal::add))));
// ➜ {An={2026-07=1250000, 2026-08=300000}, Bình={2026-08=780000}}
// ❗ groupingBy mặc định trả HashMap — tháng lộn xộn. Truyền TreeMap::new khi thứ tự có ý nghĩa.

// ② SỐ LƯỢNG BÁN THEO SẢN PHẨM — flatMap XUỐNG dòng hàng rồi mới gom
Map<String, Integer> soLuongBan = donHang.stream()
    .filter(d -> d.trangThai() != TrangThai.DA_HUY)
    .flatMap(d -> d.dongHang().stream())                           // Stream<DonHang> → Stream<DongHang>
    .collect(groupingBy(DongHang::sanPham, summingInt(DongHang::soLuong)));

// ③ TOP 3 ĐƠN LỚN NHẤT CỦA MỖI KHÁCH — collectingAndThen: gom xong rồi cắt
Map<String, List<DonHang>> top3 = donHang.stream()
    .collect(groupingBy(DonHang::khachHang,
             collectingAndThen(toList(), ds -> ds.stream()
                 .sorted(comparing(DonHang::tong).reversed())
                 .limit(3).toList())));

// ④ MIN và MAX trong MỘT lần duyệt — teeing (Java 12): hai collector, một hàm ghép
record KhoangGia(BigDecimal min, BigDecimal max) {}
KhoangGia khoang = donHang.stream().map(DonHang::tong)
    .collect(teeing(minBy(naturalOrder()), maxBy(naturalOrder()),
                    (mn, mx) -> new KhoangGia(mn.orElse(BigDecimal.ZERO), mx.orElse(BigDecimal.ZERO))));

// ⑤ toMap GIỮ THỨ TỰ và XỬ LÝ TRÙNG KHÓA — đơn MỚI NHẤT của mỗi khách
Map<String, DonHang> donMoiNhat = donHang.stream()
    .sorted(comparing(DonHang::ngay))
    .collect(toMap(DonHang::khachHang, d -> d,
                   (cu, moi) -> moi,            // ❗ thiếu hàm merge → IllegalStateException khi trùng khóa
                   LinkedHashMap::new));        // giữ thứ tự gặp lần đầu

// ⑥ Vài mảnh nhỏ hay quên: mapMulti (Java 16) thay flatMap khi cần lọc + biến đổi cùng lúc,
//    takeWhile/dropWhile trên dữ liệu ĐÃ SẮP XẾP, Stream.iterate có điều kiện dừng (Java 9)
List<DonHang> dauThang = donHang.stream().sorted(comparing(DonHang::ngay))
    .takeWhile(d -> d.ngay().getDayOfMonth() <= 7).toList();
List<LocalDate> tuan = Stream.iterate(LocalDate.of(2026, 9, 1), n -> n.isBefore(LocalDate.of(2026, 10, 1)), n -> n.plusWeeks(1)).toList();

• Lọc trước, gom sau — filter đứng trước groupingBy để không gom cả đơn đã hủy rồi mới loại; và groupingBy mặc định là HashMap, cần thứ tự thì truyền TreeMap::new.

• Tiền là BigDecimal — reducing(ZERO, DonHang::tong, BigDecimal::add) thay cho summingDouble; 0.1 + 0.2 của double không bao giờ ra 0.3.

• toMap không có hàm merge là ném IllegalStateException ngay khóa trùng đầu tiên; muốn giữ thứ tự thì thêm LinkedHashMap::new làm tham số thứ tư.

• teeing (Java 12) chạy hai collector trên cùng một lần duyệt — đúng chỗ cho "vừa min vừa max", "vừa đếm vừa tổng" mà không duyệt hai lần.

Đánh đổi readability — khi nào dừng lại

Stream lồng bốn cấp compile được, chạy đúng, và không ai dám sửa. Có ba cách cứu, theo thứ tự ưu tiên: đặt tên cho collector bằng một method tĩnh trả Collector — collector là giá trị, tách ra và test riêng được; dùng record cho kết quả trung gian thay vì Map.Entry lồng nhau; và quay về vòng for khi logic có nhánh, cần break, hay phải ném checked exception. Cùng lúc, nhớ hai điều: peek không phải chỗ để log, và GROUP BY của database luôn thắng groupingBy trên một triệu dòng.

// KHI NÀO STREAM ĐÃ THÀNH KHÓ ĐỌC — và ba cách cứu.

// ❌ Bốn cấp lồng nhau: người đọc phải "mở ngoặc" trong đầu, và không ai dám sửa
Map<String, Map<YearMonth, Optional<DonHang>>> donLonNhat = donHang.stream()
    .collect(groupingBy(DonHang::khachHang,
             groupingBy(d -> YearMonth.from(d.ngay()),
                        maxBy(comparing(DonHang::tong)))));

// ✅ ① ĐẶT TÊN cho collector — một method tĩnh, pipeline đọc như tiếng người
static Collector<DonHang, ?, BigDecimal> tongTien() {
    return reducing(BigDecimal.ZERO, DonHang::tong, BigDecimal::add);
}
static Collector<DonHang, ?, Map<YearMonth, BigDecimal>> theoThang() {
    return groupingBy(d -> YearMonth.from(d.ngay()), TreeMap::new, tongTien());
}
var doanhThu = donHang.stream().collect(groupingBy(DonHang::khachHang, theoThang()));
// ⭐ Collector là giá trị — tách ra, đặt tên, TEST RIÊNG được.

// ✅ ② RECORD cho kết quả trung gian — thay vì Map.Entry / Object[] / Pair<A, B>
record DoanhThuThang(String khach, YearMonth thang, BigDecimal tien) {}
List<DoanhThuThang> bangXepHang = doanhThu.entrySet().stream()
    .flatMap(e -> e.getValue().entrySet().stream()
        .map(t -> new DoanhThuThang(e.getKey(), t.getKey(), t.getValue())))
    .sorted(comparing(DoanhThuThang::tien).reversed())
    .toList();
// ➜ e.getValue().getKey() không nói gì cả; dt.thang() thì nói tất cả.

// ✅ ③ VÒNG FOR khi có nhiều nhánh, cần break, hay phải ném checked exception (Ngày 52)
Map<String, BigDecimal> congNo = new HashMap<>();
for (DonHang d : donHang) {
    if (d.trangThai() != TrangThai.MOI) continue;
    if (d.ngay().isBefore(hanChot)) guiNhacNo(d);            // throws IOException — lambda KHÔNG ném được
    congNo.merge(d.khachHang(), d.tong(), BigDecimal::add);
}
// ⭐ Stream thắng khi mỗi bước là MỘT phép biến đổi thuần. Vòng for thắng khi có trạng thái, nhánh, ngoại lệ.

// ❗ peek để log trong production: chạy hay không tùy phép kết thúc (Ngày 56) — dùng map hoặc forEach rõ ràng.
// ❗ ĐỪNG kéo một triệu dòng lên chỉ để groupingBy: GROUP BY của DB nhanh hơn hàng trăm lần và không ăn heap.
//    Stream dành cho dữ liệu ĐÃ nằm trong bộ nhớ: một trang kết quả, cache, file nhỏ, response của API.

// 🔍 DEBUG một pipeline: tách ra biến, toList() ở từng bước, đặt breakpoint — rồi ghép lại sau
var daLoc     = donHang.stream().filter(d -> d.trangThai() == TrangThai.DA_THANH_TOAN).toList();  // xem daLoc
var theoKhach = daLoc.stream().collect(groupingBy(DonHang::khachHang));                          // xem theoKhach
// IntelliJ: breakpoint trong pipeline → "Trace Current Stream Chain" hiện từng phần tử đi qua mỗi bước.

• Quy tắc ba cấp: collector lồng quá hai cấp thì tách cấp trong cùng thành method có tên — theoThang(), tongTien() — pipeline còn lại đọc như một câu.

• Record thay tuple — e.getValue().getKey() không nói gì, dt.thang() nói tất cả; record ba trường tốn một dòng khai báo.

• Vòng for không phải bước lùi: có trạng thái tích lũy, có nhánh, có checked exception thì vòng for ngắn hơn và debug được; stream chỉ thắng khi mỗi bước là một phép biến đổi thuần.

• Stream không thay database: kéo một triệu dòng lên rồi groupingBy là ăn hết heap; GROUP BY chạy trên index, trả về vài trăm dòng — stream dành cho dữ liệu đã trong bộ nhớ.

Từ CSV đến báo cáo — bài toán khép lại Giai đoạn 6

Một bài toán gom cả chín ngày: đọc CSV bằng Files.lines lười trong try-with-resources, parse từng dòng thành record, ghép dòng hàng thành đơn bằng groupingBy theo mã, rồi mỗi con số thống kê là một pipeline ngắn có tên — summaryStatistics cho đếm/tổng/trung bình, groupingBy với collector đã đặt tên cho doanh thu theo tháng, max trả Optional cho khách VIP — và cuối cùng in bằng String.formatted với joining có prefix và suffix.

// TỪ FILE CSV ĐẾN BÁO CÁO — gói cả Giai đoạn 6 vào một bài toán.
// don-hang.csv:  ma,khach,trangThai,ngay,sanPham,soLuong,donGia   (mỗi dòng CSV = MỘT dòng hàng)

record DongCsv(String ma, String khach, TrangThai tt, LocalDate ngay, String sp, int sl, BigDecimal gia) {
    static DongCsv parse(String line) {
        String[] c = line.split(",");
        return new DongCsv(c[0], c[1], TrangThai.valueOf(c[2]), LocalDate.parse(c[3]),
                           c[4], Integer.parseInt(c[5]), new BigDecimal(c[6]));
    }
}

// ① ĐỌC & PARSE — Files.lines là stream LƯỜI; try-with-resources để đóng file (Ngày 44, 46)
List<DongCsv> dong;
try (Stream<String> lines = Files.lines(Path.of("don-hang.csv"))) {
    dong = lines.skip(1)                                        // bỏ dòng tiêu đề
                .filter(l -> !l.isBlank())
                .map(DongCsv::parse)                            // ❗ NumberFormatException? để nó nổ — dữ liệu hỏng thì báo cáo sai
                .toList();
}

// ② GHÉP dòng hàng thành đơn — groupingBy theo mã (LinkedHashMap giữ thứ tự file), rồi map từng nhóm
List<DonHang> donHang = dong.stream()
    .collect(groupingBy(DongCsv::ma, LinkedHashMap::new, toList()))
    .values().stream()
    .map(ds -> new DonHang(ds.get(0).ma(), ds.get(0).khach(), ds.get(0).tt(), ds.get(0).ngay(),
                           ds.stream().map(d -> new DongHang(d.sp(), d.sl(), d.gia())).toList()))
    .toList();

// ③ THỐNG KÊ — mỗi con số một pipeline NGẮN, có tên
LongSummaryStatistics tk = donHang.stream()
    .filter(d -> d.trangThai() == TrangThai.DA_THANH_TOAN)
    .mapToLong(d -> d.tong().longValue())
    .summaryStatistics();                                       // count, sum, average, min, max — một lần duyệt (Ngày 56)

Map<YearMonth, BigDecimal> theoThang = donHang.stream()
    .filter(d -> d.trangThai() == TrangThai.DA_THANH_TOAN)
    .collect(groupingBy(d -> YearMonth.from(d.ngay()), TreeMap::new, tongTien()));   // tongTien() ở phần trên

Optional<String> khachVip = donHang.stream()
    .filter(d -> d.trangThai() == TrangThai.DA_THANH_TOAN)
    .collect(groupingBy(DonHang::khachHang, tongTien()))
    .entrySet().stream()
    .max(Map.Entry.comparingByValue())
    .map(Map.Entry::getKey);                                    // Optional — có thể không có đơn nào (Ngày 58)

// ④ IN BÁO CÁO — String.formatted + joining với prefix/suffix; KHÔNG nối chuỗi bằng reduce (Ngày 56)
String baoCao = theoThang.entrySet().stream()
    .map(e -> "%s   %,14d đ".formatted(e.getKey(), e.getValue().longValue()))
    .collect(joining("\n",
                     "DOANH THU THEO THÁNG\n",
                     "\n— %d đơn, tổng %,d đ, khách VIP: %s".formatted(tk.getCount(), tk.getSum(), khachVip.orElse("(chưa có)"))));
System.out.println(baoCao);
// DOANH THU THEO THÁNG
// 2026-07        1,250,000 đ
// 2026-08        1,080,000 đ
// — 3 đơn, tổng 2,330,000 đ, khách VIP: An

// ⭐ NHÌN LẠI GIAI ĐOẠN 6: lambda (51) → functional interface (52) → method reference (53) → pipeline lười (54)
//    → map/filter/flatMap (55) → collect/reduce (56) → Collectors (57) → Optional (58) → parallel (59).
//    Bài toán thật nào cũng chỉ là những mảnh này ghép lại — mỗi mảnh ngắn, có tên, và test được.

• Files.lines phải nằm trong try-with-resources — stream giữ file handle cho đến khi đóng; skip(1) bỏ tiêu đề, filter(not blank) bỏ dòng trống cuối file.

• Dữ liệu hỏng thì để nó nổ — NumberFormatException ở dòng 1.203 tốt hơn một báo cáo lệch mà không ai phát hiện; nếu muốn bỏ qua dòng lỗi, mapMulti với try/catch và log rõ ràng.

• Mỗi con số một pipeline — ba pipeline ngắn có tên dễ đọc hơn một pipeline dài làm cả ba việc; chi phí duyệt lại một list trong bộ nhớ là không đáng kể.

• joining(sep, prefix, suffix) với %,d cho dấu phân cách hàng nghìn — báo cáo văn bản không cần thư viện nào cả.

Bài tập nhỏ

• Tạo 10 đơn hàng mẫu, tính doanh thu theo khách theo tháng bằng groupingBy hai cấp; đổi TreeMap::new thành mặc định và in ra để thấy tháng lộn xộn.

• Viết toMap lấy đơn mới nhất của mỗi khách mà không truyền hàm merge — đọc thông điệp IllegalStateException, rồi thêm merge.

• Dùng teeing để lấy đồng thời số đơn đã thanh toán và số đơn đã hủy trong một lần duyệt, trả về một record.

• Lấy một pipeline lồng ba cấp bạn từng viết (hoặc donLonNhat ở trên), tách cấp trong cùng thành method tĩnh có tên và viết một unit test cho riêng collector đó.

• Hoàn thiện báo cáo CSV: thêm cột "sản phẩm bán chạy nhất" mỗi tháng — gợi ý: flatMap xuống dòng hàng, groupingBy tháng rồi sản phẩm, collectingAndThen lấy max.

Kết lại

Bốn ý gói lại hôm nay: dữ liệu thật cần groupingBy nhiều cấp — lọc trước, TreeMap khi cần thứ tự, flatMap xuống dòng hàng, collectingAndThen và teeing cho top N và min/max; tiền là BigDecimal và toMap luôn có hàm merge; khi pipeline khó đọc thì đặt tên collector, dùng record thay tuple, hoặc quay về vòng for — và đừng bắt stream làm việc của GROUP BY; cuối cùng, một báo cáo thật chỉ là nhiều pipeline ngắn có tên ghép lại. Đó cũng là lời kết cho Giai đoạn 6: lập trình hàm trong Java không phải cú pháp cho đẹp, mà là cách chia bài toán thành những phép biến đổi nhỏ, thuần, và test được. Ngày 61 mở Giai đoạn 7 — Concurrency — bằng thứ nền tảng nhất: Thread và Runnable, vòng đời của một luồng, sleep, join và interrupt. Hẹn gặp lại!