iterator pattern

Trong bộ design pattern GoF, Iterator Pattern phổ biến tới mức dev nào cũng dùng hằng ngày mà không để ý: từ vòng for-each, java.util.Iterator tới các luồng dữ liệu vô hạn trong Stream API.

Về lý thuyết, mục đích của Iterator rất gọn: cho phép duyệt lần lượt các phần tử của một tập hợp mà không lộ cấu trúc dữ liệu bên dưới (dù bên dưới là Array, Doubly Linked List, Red-Black Tree hay B-Tree).

Nhưng trong hệ thống doanh nghiệp tải lớn, chạy đồng thời nhiều thread, Iterator có khá nhiều cái bẫy:

Vì sao vòng for-each tự dưng ném ConcurrentModificationException?
modCount chạy thế nào trong mã nguồn OpenJDK?
Vì sao Iterator kiểu cũ bó tay với tính toán song song, khiến Java phải đẻ ra Spliterator?
Làm sao duyệt hàng chục triệu bản ghi database mà không tràn heap (OOM)?

Mình sẽ lần lượt đi qua từng câu, nhìn từ góc của người làm hệ thống.

# fail-fast và biến `modCount` trong openjdk

Lỗi kinh điển mà dev Java nào cũng từng dính: xoá phần tử trong lúc đang duyệt bằng for-each:

List<String> servers = new ArrayList<>(List.of("srv-1", "srv-2", "srv-3"));
for (String server : servers) {
    if (server.equals("srv-1")) {
        servers.remove(server); // 🔴 NÉM RA ConcurrentModificationException!
    }
}

# vì sao lại thế?

for-each thật ra chỉ là syntactic sugar, compiler dịch nó thành một vòng lặp Iterator:

Iterator<String> it = servers.iterator();
while (it.hasNext()) {
    String server = it.next();
    if (server.equals("srv-1")) {
        servers.remove(server); // Gọi trực tiếp hàm remove() của List!
    }
}

Nhìn vào bên trong ArrayList.java của OpenJDK:

# `modCount` thật ra là gì

modCount (modification count) là một field int (khai báo trong AbstractList) đếm số lần collection bị thay đổi cấu trúc: thêm, xoá phần tử, clear()... Gán đè giá trị bằng set() thì không tính.
Lúc tạo Iterator, nó chụp lại: expectedModCount = modCount.

Mỗi lần gọi next() (và remove() của iterator), Iterator kiểm tra:

final void checkForComodification() {
    if (modCount != expectedModCount)
        throw new ConcurrentModificationException();
}

Cách làm an toàn: dùng iterator.remove(). Hàm này xoá phần tử thông qua Iterator và tự đồng bộ lại expectedModCount = modCount, nên không bị lỗi. Từ Java 8 còn gọn hơn: servers.removeIf(s -> s.equals("srv-1")).

Một cái bẫy nữa: hasNext() không kiểm tra modCount, nó chỉ so cursor != size. Thử đổi điều kiện thành xoá "srv-2" (phần tử áp chót) xem: sau khi xoá, size còn 2 đúng bằng cursor, vòng lặp kết thúc êm ru, không có exception nào, và "srv-3" bị bỏ qua không được duyệt. Đó là lý do Javadoc nói fail-fast chỉ là "best-effort": nó giúp phát hiện bug, đừng dựa vào nó để đảm bảo đúng.

# fail-fast vs fail-safe / weakly consistent khi chạy đa luồng

Khi chạy multi-threading, các collection thường (ArrayList, HashSet, HashMap) vốn không thread-safe: chỉ cần một thread ghi là các thread đang duyệt có thể dính ConcurrentModificationException, hoặc tệ hơn là đọc dữ liệu sai mà không báo gì (vì modCount cũng không được đồng bộ giữa các thread).

Trong java.util.concurrent, Java có hai hướng xử lý khác nhau:

# `CopyOnWriteArrayList` (fail-safe nhờ snapshot)

Khi thread ghi gọi add() hay remove(), nó copy cả mảng dữ liệu sang vùng nhớ mới.
Iterator duyệt trên mảng cũ tại lúc được tạo, nên không bị ảnh hưởng gì bởi thay đổi.
Trade-off: mỗi lần ghi là một lần copy O(N), tốn RAM và tạo rác cho GC nếu dữ liệu lớn. Chỉ hợp với kiểu đọc rất nhiều, ghi rất ít như danh sách listener hay route table. Iterator của nó cũng không hỗ trợ remove() (ném UnsupportedOperationException).

# `ConcurrentHashMap` (weakly consistent)

Không copy cả bảng băm.
Iterator duyệt thẳng qua các node của bucket table. Nếu thread khác thêm node vào sau vị trí con trỏ đang đứng, Iterator có thể thấy hoặc không thấy phần tử mới đó.
Không bao giờ ném ConcurrentModificationException, throughput đọc rất cao.
Cái giá: kết quả duyệt không phải ảnh chụp tại một thời điểm. Những thứ như size() hay tổng các giá trị tính trong lúc có thread khác ghi chỉ là con số gần đúng.

# tiến lên `Spliterator`: chạy parallel stream (Java 8+)

Iterator chỉ có hasNext() và next(), tức là chạy tuần tự từ đầu tới cuối. Muốn chia một Iterator cho 16 worker thread thì các thread phải tranh nhau gọi next() qua lock, hoặc một thread đọc hết rồi chia lại, đều tốn công đồng bộ.

Để gỡ giới hạn này cho Stream API và ForkJoinPool, Java 8 thêm Spliterator (Splitable Iterator):

# các method chính của `Spliterator`

boolean tryAdvance(Consumer<? super T> action): giống next() của Iterator, xử lý một phần tử.
Spliterator<T> trySplit(): phần quan trọng nhất cho chạy song song. Tách một phần dữ liệu hiện tại (lý tưởng là một nửa) ra một Spliterator mới; với nguồn ORDERED thì phần tách ra phải là phần đầu. Không chia được nữa thì trả null. Nguồn chia đều tốt như ArrayList hay mảng chạy song song rất hiệu quả; còn LinkedList hay một Iterator bọc lại thì chia rất kém, parallel stream trên chúng thường chậm hơn chạy tuần tự.
long estimateSize(): ước lượng số phần tử còn lại để ForkJoinPool quyết định có chia tiếp hay thôi.
int characteristics(): trả về một bitmask mô tả đặc điểm của dữ liệu, giúp stream engine tối ưu:
- ORDERED: dữ liệu có thứ tự xác định.
- DISTINCT: không có phần tử trùng.
- SORTED: đã sắp xếp sẵn.
- SIZED: biết chính xác kích thước trước khi duyệt.
- CONCURRENT: nguồn có thể bị sửa đồng thời một cách an toàn.

Bản thân một Spliterator không thread-safe. Ý tưởng là mỗi phần sau khi tách được đúng một thread xử lý, chứ không phải nhiều thread dùng chung một spliterator.

# tự viết cursor streaming Iterator: stream 10 triệu bản ghi không tràn ram

Bài toán hay gặp ở doanh nghiệp: export báo cáo từ một bảng có 10,000,000 giao dịch ra file CSV hoặc đẩy qua Kafka.

Nếu dùng Spring Data JPA kiểu:

List<Transaction> all = transactionRepository.findAll(); // 🔴 SẬP TOÀN BỘ SERVER VỚI OutOfMemoryError!

Heap 4GB rất dễ cạn, vì 10 triệu entity (cộng với persistence context của Hibernate giữ tham chiếu tới từng entity) dễ dàng chiếm nhiều GB bộ nhớ.

# cách xử lý: Keyset Pagination Iterator (streaming cursor)

Bọc keyset pagination (seek method) bên dưới một Iterator, mình có thể duyệt hàng chục triệu bản ghi mà RAM chỉ tốn cố định O(1) theo tổng số dòng (lúc nào cũng chỉ giữ đúng một batch trong bộ nhớ).

Khác với LIMIT ... OFFSET ..., câu WHERE id > :lastId ORDER BY id LIMIT n đi thẳng vào index tại vị trí cũ, nên trang thứ 5000 cũng nhanh như trang đầu. OFFSET thì database vẫn phải đọc rồi bỏ qua mọi dòng phía trước, càng về sau càng chậm. Điều kiện là cột dùng làm khoá phải có index và duy nhất (hoặc ghép thêm id vào cho duy nhất).

# code

package com.company.common.iterator;
 
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.function.Function;
 
public class KeysetPaginationIterator<T, ID extends Comparable<ID>> implements Iterator<T> {
 
    private final Function<ID, List<T>> pageFetcher;
    private final Function<T, ID> idExtractor;
    private final int batchSize;
 
    private List<T> currentBatch = Collections.emptyList();
    private int cursor = 0;
    private ID lastSeenId = null;
    private boolean isExhausted = false;
 
    public KeysetPaginationIterator(
            Function<ID, List<T>> pageFetcher,
            Function<T, ID> idExtractor,
            int batchSize) {
        this.pageFetcher = pageFetcher;
        this.idExtractor = idExtractor;
        this.batchSize = batchSize;
    }
 
    @Override
    public boolean hasNext() {
        if (cursor < currentBatch.size()) {
            return true;
        }
        if (isExhausted) {
            return false;
        }
 
        // Tải mẻ dữ liệu kế tiếp từ DB
        fetchNextBatch();
        return !currentBatch.isEmpty();
    }
 
    @Override
    public T next() {
        if (!hasNext()) {
            throw new NoSuchElementException("Đã duyệt hết dữ liệu!");
        }
        T item = currentBatch.get(cursor++);
        lastSeenId = idExtractor.apply(item);
        return item;
    }
 
    private void fetchNextBatch() {
        currentBatch = pageFetcher.apply(lastSeenId);
        cursor = 0;
        if (currentBatch.size() < batchSize) {
            // Khi số lượng bản ghi trả về nhỏ hơn batchSize -> Đã chạm đến đáy dữ liệu
            isExhausted = true;
        }
    }
}

# dùng trong pipeline export dữ liệu

@Service
public class TransactionExportService {
 
    @Autowired
    private JdbcTemplate jdbcTemplate;
 
    public void exportMillionsOfRecordsToCsv(OutputStream outputStream) throws IOException {
        var iterator = new KeysetPaginationIterator<TransactionRecord, Long>(
            // Lambda truy vấn DB theo batch với Seek Method (cực nhanh có Index)
            // Không dùng "(? IS NULL OR id > ?)": khi PostgreSQL chuyển sang generic plan,
            // điều kiện OR có thể khiến mỗi batch quét index từ đầu bảng.
            lastId -> jdbcTemplate.query("""
                SELECT id, account_id, amount, status, created_at
                FROM transactions
                WHERE id > ?
                ORDER BY id ASC
                LIMIT 2000
                """,
                new TransactionRowMapper(),
                lastId == null ? Long.MIN_VALUE : lastId
            ),
            TransactionRecord::id,
            2000
        );
 
        try (var writer = new BufferedWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8))) {
            writer.write("id,account_id,amount,status,created_at\n");
 
            // Xử lý từng phần tử: bộ nhớ chỉ phụ thuộc kích thước batch, không phụ thuộc tổng số dòng
            while (iterator.hasNext()) {
                TransactionRecord tx = iterator.next();
                // amount là BigDecimal: dùng toPlainString() thay vì "%.2f",
                // vì String.format theo Locale mặc định (vd vi_VN in "1234,50" làm vỡ CSV)
                writer.write(String.join(",",
                    String.valueOf(tx.id()), String.valueOf(tx.accountId()),
                    tx.amount().toPlainString(), String.valueOf(tx.status()),
                    String.valueOf(tx.createdAt())));
                writer.write("\n");
            }
            writer.flush();
        }
    }
}

Một lựa chọn khác là để JDBC driver tự stream bằng cursor phía server (setFetchSize, JdbcTemplate.queryForStream). Với PostgreSQL, driver chỉ stream khi connection tắt autocommit (tức là chạy trong transaction), nên cả lần export là một transaction dài giữ snapshot, có thể cản VACUUM trên bảng ghi nhiều. Keyset pagination thì mỗi batch là một query ngắn, đổi lại không có snapshot nhất quán cho toàn bộ file: dòng nào bị sửa trong lúc export thì có thể ra bản cũ hoặc bản mới.

# so sánh các cách duyệt trong Java

Tiêu chí	`java.util.Iterator`	`for-each` loop	`CopyOnWriteArrayList`	`ConcurrentHashMap`	`Spliterator`
Cơ chế	Con trỏ object	Cú pháp bọc ngoài Iterator	Copy mảng (snapshot)	Duyệt thẳng bucket	Chia để trị (Fork-Join)
Khi bị sửa	Fail-Fast (`modCount`)	Fail-Fast (`modCount`)	Fail-Safe	Weakly Consistent	Tuỳ data source
Tốn bộ nhớ	1 object Iterator nhỏ	1 object Iterator nhỏ	Rất cao khi có ghi	Rất thấp (O(1))	Thấp
Xoá phần tử	`iterator.remove()`	🔴 Cấm (gây crash)	🔴 Không hỗ trợ remove	Có hỗ trợ	Tuỳ cách cài đặt
Đa luồng	🔴 Không đồng bộ	🔴 Không đồng bộ	🟢 An toàn	🟢 An toàn	🚀 Chia phần cho từng thread

# tổng kết

Iterator Pattern trong Java đã đi một chặng khá dài:

Bắt đầu từ GoF Iterator để giấu cấu trúc dữ liệu.
Thêm fail-fast với modCount để giữ dữ liệu toàn vẹn.
Giải bài toán concurrency bằng fail-safe và weakly consistent.
Tận dụng CPU nhiều nhân với Spliterator từ Java 8.
Và thành công cụ rất tiện để stream dữ liệu cực lớn qua cursor iterator.

Nếu chỉ nhớ một điều để mang về code thì: đừng sửa collection trong lúc duyệt (dùng removeIf hoặc iterator.remove()), và với dữ liệu lớn từ database thì duyệt theo batch bằng keyset thay vì findAll().

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.

# fail-fast và biến modCount trong openjdk