Stream API & vòng lặp: báo cáo doanh thu theo khách hàng

Bài 1.11 đã có lambda. Giờ shop-backend cần một endpoint báo cáo: từ danh sách Order, lấy đơn PAID, tính tổng doanh thu, và gom doanh thu theo từng khách hàng.

1. Bài toán: viết bằng vòng lặp

BigDecimal revenue = BigDecimal.ZERO;
Map<String, BigDecimal> byCustomer = new HashMap<>();
for (Order o : orders) {
    if (o.status() != OrderStatus.PAID) continue;
    revenue = revenue.add(o.total());
    BigDecimal old = byCustomer.get(o.customerId());
    byCustomer.put(o.customerId(), old == null ? o.total() : old.add(o.total()));
}

2. Vấn đề

• Code nói làm thế nào (biến tạm, if null, put), không nói muốn gì. Ba yêu cầu trộn trong một vòng lặp.
• Biến mutable dùng chung; muốn chạy song song là phải tự lo đồng bộ.

3. Giải pháp: Stream pipeline

• Stream không phải collection, không lưu dữ liệu: là một pipeline xử lý dữ liệu từ nguồn (list, array, file...).
• Gồm source → intermediate ops (filter, map, flatMap, sorted, distinct: trả về stream, lazy) → terminal op (collect, reduce, forEach, count, findFirst: kích hoạt chạy).
• Không có terminal op thì không có gì chạy. Stream chỉ dùng được một lần.

4. Code từng bước

Bước 1: filter + map + reduce: tổng doanh thu

BigDecimal revenue = orders.stream()
        .filter(o -> o.status() == OrderStatus.PAID)
        .map(Order::total)
        .reduce(BigDecimal.ZERO, BigDecimal::add);

Bước 2: collect + groupingBy: doanh thu theo khách

Map<String, BigDecimal> byCustomer = orders.stream()
        .filter(o -> o.status() == OrderStatus.PAID)
        .collect(Collectors.groupingBy(
                Order::customerId,
                Collectors.reducing(BigDecimal.ZERO, Order::total, BigDecimal::add)));

Map<OrderStatus, Long> countByStatus = orders.stream()
        .collect(Collectors.groupingBy(Order::status, Collectors.counting()));

Bước 3: flatMap: mỗi đơn có nhiều item, lấy tất cả sản phẩm đã bán

// record Order(..., List<OrderItem> items)
Set<String> soldSkus = orders.stream()
        .flatMap(o -> o.items().stream())       // Stream<Order> → Stream<OrderItem>
        .map(OrderItem::sku)
        .collect(Collectors.toSet());

Bước 4: Top 5 đơn lớn nhất, map sang DTO

List<OrderSummary> top5 = orders.stream()
        .sorted(Comparator.comparing(Order::total).reversed())
        .limit(5)
        .map(o -> new OrderSummary(o.id(), o.total()))
        .toList();                               // Java 16+, list không sửa được

5. Cách hoạt động: lazy, từng phần tử đi hết pipeline

orders ──► filter(PAID) ──► map(total) ──► reduce(add)
            (chưa chạy)      (chưa chạy)     ▲ terminal: bắt đầu kéo dữ liệu

Không phải: filter cả list → rồi map cả list.
Mà là từng phần tử một:
  o1: filter ✔ → map → reduce
  o2: filter ✘ (dừng)
  o3: filter ✔ → map → reduce
Short-circuit: .filter(...).findFirst() dừng ngay khi tìm thấy phần tử đầu tiên.

6. So sánh

7. Khi nào nên dùng / không nên

• Stream: biến đổi/lọc/gom nhóm collection, logic dạng pipeline, không side effect.
• Vòng lặp: cần index, cần break phức tạp, ném checked exception, sửa nhiều biến cùng lúc, hoặc đoạn nóng (hot path) xử lý primitive.
• Không dùng stream().forEach() để add vào list bên ngoài; dùng collect/toList().
• Với dữ liệu trong DB, đừng findAll() rồi stream để group: để SQL GROUP BY làm.

Đúng về cơ bản: cả hai đều tuần tự, xử lý từng phần tử một, không tự tạo thread. Nhưng có một điểm cần sửa: for-each không chỉ dùng cho List. Nó dùng được với array và mọi thứ implement Iterable (List, Set, Queue...). Map không phải Iterable, phải đi qua entrySet().

for (int i = 0; i < orders.size(); i++)          // cần index, nhảy bước, chạy ngược
    System.out.println(i + ": " + orders.get(i).id());

for (Order o : orders) { ... }                     // chỉ cần phần tử; compile thành Iterator
for (Order o : orderArray) { ... }                 // array: compile thành for-i

Iterator<Order> it = orders.iterator();           // cần xóa trong lúc duyệt
while (it.hasNext()) if (it.next().status() == CANCELLED) it.remove();

orders.forEach(o -> log.info(o.id()));             // Iterable.forEach (Java 8), tuần tự
orders.stream().forEach(...);                      // tuần tự
orders.parallelStream().forEach(...);              // CÓ THỂ song song, thứ tự không đảm bảo

Nhớ đơn giản: for, for-each, stream().forEach() mặc định đều tuần tự; chỉ parallelStream() mới liên quan song song.

Tên gây hiểu nhầm: không cần nhiều thread. CME xảy ra khi collection bị sửa cấu trúc trong lúc đang có Iterator duyệt, bởi bất kỳ ai, kể cả chính thread đó.

for (Order o : orders) {
    if (o.status() == CANCELLED) orders.remove(o);   // ❌ CME ở lần next() sau
}

ArrayList có modCount (số lần sửa cấu trúc)
Iterator tạo ra ghi nhớ expectedModCount = modCount
orders.remove(o)  → modCount++
it.next()         → modCount != expectedModCount → throw CME   (fail-fast)

Cách đúng:

orders.removeIf(o -> o.status() == CANCELLED);           // ✅ gọn nhất (Java 8)

Iterator<Order> it = orders.iterator();                   // ✅ xóa qua iterator
while (it.hasNext()) if (it.next().status() == CANCELLED) it.remove();

List<Order> active = orders.stream()                     // ✅ tạo list mới, không mutate
        .filter(o -> o.status() != CANCELLED).toList();

Đa thread thật sự thì dùng CopyOnWriteArrayList / ConcurrentHashMap (iterator weakly consistent, không ném CME).

Không hẳn, thường là ngược lại trong ứng dụng web. Cần chỉnh mấy điểm:

• parallelStream() chạy trên ForkJoinPool.commonPool() dùng chung cả JVM (số thread ≈ số core − 1). Một request gọi DB/HTTP trong parallel stream sẽ chiếm pool đó, làm chậm mọi chỗ khác.
• Chi phí chia/gộp lớn: chỉ có lợi khi dữ liệu rất lớn (hàng trăm nghìn+), thao tác CPU-bound, nguồn chia tốt (ArrayList, array; LinkedList chia kém).
• Side effect gây sai: parallelStream().forEach(list::add) vào ArrayList là race condition. Dùng collect.
• forEach không giữ thứ tự; cần thứ tự thì forEachOrdered (mất lợi ích).
• Server đã có nhiều request song song, CPU vốn đã bận; song song thêm trong một request thường không có lợi.

// ❌ sai và nguy hiểm
List<Order> result = new ArrayList<>();
orders.parallelStream().filter(Order::isPaid).forEach(result::add);

// ✅ đúng
List<Order> result = orders.parallelStream().filter(Order::isPaid).toList();

Gọi nhiều service I/O song song thì dùng CompletableFuture với executor riêng hoặc virtual thread (Java 21), không phải parallel stream.

Nói Stream là gì, cấu trúc pipeline, tính lazy, rồi khi nào chọn vòng lặp.

Câu 1: Stream API là gì? Khác collection, khác vòng lặp thế nào?

🇻🇳 Tiếng Việt:

Stream là pipeline xử lý dữ liệu theo kiểu declarative, có từ Java 8. Nó không lưu dữ liệu và chỉ dùng được một lần. Pipeline gồm source, các intermediate operation như filter, map là lazy, và một terminal operation như collect, reduce mới kích hoạt chạy. So với vòng lặp, stream nói "muốn gì" thay vì "làm thế nào", ít biến mutable. Trong dự án em dùng filter đơn PAID rồi groupingBy(customerId) để tính doanh thu theo khách. Vòng lặp vẫn tốt hơn khi cần index, break, checked exception hoặc hot path.

🇬🇧 English:

A stream is a declarative data-processing pipeline introduced in Java 8. It doesn't store data and can be consumed only once. A pipeline has a source, lazy intermediate operations like filter and map, and one terminal operation like collect or reduce that triggers execution. Compared with loops, it says what you want rather than how, with less mutable state. In my project I filter paid orders and use groupingBy(customerId) to compute revenue per customer. A loop is still better when I need an index, break, checked exceptions or a hot path.

Câu 2: map và flatMap khác nhau gì?

🇻🇳 Tiếng Việt:

map biến mỗi phần tử thành đúng một phần tử. flatMap biến mỗi phần tử thành một stream rồi làm phẳng, ví dụ từ Stream<Order> sang Stream<OrderItem> bằng o -> o.items().stream().

🇬🇧 English:

map transforms each element into exactly one element. flatMap transforms each element into a stream and flattens the result, for example from Stream<Order> to Stream<OrderItem> with o -> o.items().stream().

Câu 3: ConcurrentModificationException là gì?

🇻🇳 Tiếng Việt:

Là lỗi fail-fast khi collection bị sửa cấu trúc trong lúc iterator đang duyệt, không cần đa thread; ví dụ gọi list.remove() trong for-each. Cách sửa là removeIf, iterator.remove(), tạo list mới bằng stream, hoặc dùng concurrent collection khi đa thread.

🇬🇧 English:

It's a fail-fast error thrown when a collection is structurally modified while an iterator is traversing it, even on a single thread, e.g. calling list.remove() inside a for-each. Fix it with removeIf, iterator.remove(), building a new list with a stream, or concurrent collections for multi-threaded code.

Câu 4: Khi nào dùng parallel stream?

🇻🇳 Tiếng Việt:

Hiếm khi trong web app. Nó dùng chung ForkJoinPool.commonPool(), chỉ có lợi với dữ liệu lớn, CPU-bound, nguồn chia tốt và không side effect. Không dùng cho I/O như gọi DB hay HTTP; khi đó dùng CompletableFuture với executor riêng. Luôn đo trước khi đổi.

🇬🇧 English:

Rarely in a web app. It uses the shared ForkJoinPool.commonPool() and only pays off for large, CPU-bound, easily splittable data without side effects. Never use it for I/O like DB or HTTP calls; use CompletableFuture with a dedicated executor instead. Always measure before switching.

Cách nhớ ngắn gọn

• Source → intermediate (lazy) → terminal (chạy). Dùng 1 lần.
• filter lọc, map đổi 1-1, flatMap làm phẳng, reduce gộp, collect/groupingBy gom.
• for, for-each, forEach đều tuần tự; for-each dùng cho array + Iterable.
• Xóa khi duyệt: removeIf / iterator.remove(), không thì CME.
• parallelStream = commonPool, không I/O, không side effect, đo trước.