Concurrent collections: khi HashMap bị nhiều thread dùng chung

Vẫn project Order Service trên Tomcat. Lần này là dữ liệu dùng chung giữa các request thread: bộ đếm đơn theo trạng thái, cache payment, hàng đợi gửi email.

1. Bài toán

@Service
class OrderStatService {
    private final Map<String, Integer> counts = new HashMap<>();

    void hit(String status) {                                    // "PAID", "FAILED"...
        counts.put(status, counts.getOrDefault(status, 0) + 1);  // đọc–tính–ghi
    }
}

Spring bean mặc định là singleton, nên 200 Tomcat thread dùng chung một HashMap. Mà HashMap không thread-safe.

2. Vấn đề

• Lost update: hai thread cùng đọc 5, cùng ghi 6.
• Hai thread cùng put khi map đang resize làm mất entry, hỏng cấu trúc bucket. Ở Java 7 còn có thể tạo vòng lặp trong linked list, get chạy 100% CPU mãi.
• Duyệt map trong lúc thread khác sửa: ConcurrentModificationException.

Chỗ rất dễ nhầm: "mình chạy qua thread pool rồi mà". Thread pool chỉ quản lý và tái sử dụng thread, nó không làm code thread-safe:

           shared counter / shared map
             ↑      ↑      ↑
          Thread-1 Thread-2 Thread-3     ← race condition

3. Giải pháp: concurrent collections

ConcurrentHashMap bên trong (Java 8+):

• Mảng bucket được đọc kiểu volatile, nên get không cần khóa.
• Bucket rỗng: chèn node đầu bằng CAS, không khóa.
• Bucket đã có node: synchronized trên node đầu của bucket đó. Thread sửa bucket khác vẫn chạy song song.
• Bucket dài (≥ 8 node, bảng ≥ 64) chuyển thành cây đỏ-đen như HashMap.
• Resize: nhiều thread cùng giúp chuyển dữ liệu sang bảng mới.
• Java 7 thì khác: chia 16 Segment, mỗi segment một lock.

Vì sao không cho null? Vì get trả null thì không phân biệt được "không có key" hay "giá trị null", mà đa luồng thì không thể gọi thêm containsKey một cách an toàn.

4. Code từng bước

Bước 1: Bộ đếm. Đổi sang ConcurrentHashMap và dùng thao tác gộp atomic:

@Service
class OrderStatService {
    private final ConcurrentHashMap<String, LongAdder> counts = new ConcurrentHashMap<>();

    void hit(String status) {
        counts.computeIfAbsent(status, k -> new LongAdder()).increment();
        // hoặc với Integer: counts.merge(status, 1, Integer::sum);
    }

    Map<String, Long> snapshot() {
        Map<String, Long> m = new HashMap<>();
        counts.forEach((k, v) -> m.put(k, v.sum()));
        return m;
    }
}

Bước 2: Cache thông tin payment method:

// SAI: hai thread cùng thấy null, cùng gọi payment-service
PaymentInfo p = cache.get(userId);
if (p == null) { p = paymentClient.methods(userId); cache.put(userId, p); }

// ĐÚNG: hàm load chỉ chạy một lần cho mỗi key
PaymentInfo p = cache.computeIfAbsent(userId, paymentClient::methods);

Lưu ý: hàm trong computeIfAbsent chạy khi đang giữ khóa bucket. Đừng làm việc quá lâu trong đó, và không sửa chính map đó bên trong (có thể treo hoặc ném IllegalStateException).

Bước 3: Hàng đợi email bằng BlockingQueue:

@Component
class EmailQueue {
    private final BlockingQueue<EmailJob> queue = new ArrayBlockingQueue<>(1000);
    private final ExecutorService workers = Executors.newFixedThreadPool(2);

    @PostConstruct
    void start() {
        for (int i = 0; i < 2; i++) workers.submit(this::loop);
    }

    boolean submit(EmailJob job) throws InterruptedException {
        return queue.offer(job, 200, TimeUnit.MILLISECONDS);   // đầy thì không chờ mãi
    }

    private void loop() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                EmailJob job = queue.take();                   // rỗng thì chờ
                mailer.send(job);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();            // giữ cờ interrupt rồi thoát
            }
        }
    }

    @PreDestroy
    void stop() { workers.shutdownNow(); }
}

Bước 4: Test: bắn 1000 request song song rồi so số đếm.

seq 1 1000 | xargs -P 50 -I{} curl -s -X POST localhost:8080/orders/{}/pay > /dev/null
curl -s localhost:8080/stats
# HashMap:           {"PAID": 962}   ← mất update
# ConcurrentHashMap: {"PAID": 1000}

5. Cách hoạt động

ConcurrentHashMap (Java 8+)
 table[] (volatile)
 [0] null          ← put: CAS, không khóa
 [1] A → B → C     ← put: synchronized(A), chỉ khóa bucket 1
 [2] D             ← thread khác ghi bucket 2 cùng lúc, không chờ
 [3] TreeBin       ← bucket dài → cây đỏ-đen
 get(): đọc thẳng, không khóa

BlockingQueue (producer–consumer)
 Tomcat threads --put/offer--> [ job | job | job | ... 1000 ] --take--> worker-1, worker-2
                 đầy → chờ/từ chối                 rỗng → worker chờ

6. So sánh

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

• Nên: state dùng chung trong bean singleton (counter, cache nhỏ, registry); producer–consumer trong một JVM.
• Không nên: nghĩ đổi sang concurrent collection là xong; chuỗi get rồi put vẫn sai. Cache lớn, có TTL thì dùng Caffeine. Việc quan trọng (payment, email bắt buộc) thì queue trong RAM mất khi restart, nên dùng Kafka/RabbitMQ.
• Nhiều instance Order Service thì mỗi instance một map riêng; số liệu toàn cục phải ở DB/Redis.

Cả hai đều thread-safe, khác ở cách khóa. synchronizedMap bọc mọi method bằng một khóa duy nhất. An toàn, nhưng mọi thread xếp hàng, kể cả khi chỉ đọc hoặc sửa hai key chẳng liên quan:

synchronizedMap:      T1 [put PAID] → T2 [get FAILED] → T3 [get PAID]   (xếp hàng)
ConcurrentHashMap:    T1 [put PAID]
                      T2 [get FAILED]                                   (song song)
                      T3 [get PAID]

Và khi duyệt synchronizedMap, bạn phải tự synchronized (map) { for ... }, không thì vẫn CME. ConcurrentHashMap duyệt được trong lúc thread khác ghi (weakly consistent: có thể thấy hoặc không thấy thay đổi mới, nhưng không lỗi).

Đúng, đổi class thôi là chưa đủ. get và put riêng lẻ thì an toàn, nhưng cặp getOrDefault rồi put vẫn là hai bước:

T1: getOrDefault(PAID) → 5
                              T2: getOrDefault(PAID) → 5
T1: put(PAID, 6)
                              T2: put(PAID, 6)        ← mất 1 lần đếm

Phải dùng thao tác gộp: merge(status, 1, Integer::sum) hoặc computeIfAbsent(...).increment() với LongAdder. Đây chính là compound action, sẽ đào kỹ ở Bài 3.12.

Tùy tỉ lệ đọc/ghi. Với list đọc nhiều ghi ít, dùng CopyOnWriteArrayList: mỗi lần ghi nó copy cả mảng ra mảng mới rồi thay tham chiếu. Người đọc luôn thấy một snapshot không đổi, không cần khóa, duyệt không bao giờ ném CME.

private final List<OrderPaidListener> listeners = new CopyOnWriteArrayList<>();

void register(OrderPaidListener l) { listeners.add(l); }             // hiếm, copy mảng
void publish(OrderPaid e) { listeners.forEach(l -> l.on(e)); }       // thường xuyên, không khóa

Không hợp với list ghi liên tục vì mỗi lần add tốn O(n).

Còn ConcurrentModificationException thì không chỉ do đa luồng. Một thread cũng gặp nếu sửa list khi đang for-each:

for (Order o : orders) {
    if (o.isCancelled()) orders.remove(o);   // CME
}
orders.removeIf(Order::isCancelled);        // đúng

Iterator của ArrayList/HashMap là fail-fast: kiểm tra modCount, thấy đổi thì ném lỗi. Iterator của concurrent collection là weakly consistent hoặc snapshot, không ném.

Producer–consumer thì dùng BlockingQueue như Bước 3, nó lo hết phần chờ và đánh thức, bạn không phải tự wait/notify:

• put: queue đầy thì producer chờ (tự tạo backpressure).
• take: queue rỗng thì consumer chờ.
• offer/poll có timeout nếu không muốn chờ mãi.

ThreadPoolExecutor chính là BlockingQueue + nhóm worker. Thứ tự là core → queue → max → reject, không phải core → max:

new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));

task 1–5     → tạo 5 core thread chạy ngay
task 6–105   → vào queue (100 chỗ)
task 106–110 → queue đầy, tạo thêm thread tới max=10
task 111–200 → queue đầy + đủ max → rejection policy

Chọn policy theo business: task như payment thì không được tùy tiện discard. Và queue trong RAM mất khi app restart; việc quan trọng thì đưa sang Kafka/RabbitMQ.

Q1. HashMap và ConcurrentHashMap khác nhau thế nào?

🇻🇳 Tiếng Việt:

HashMap không thread-safe: dùng chung giữa nhiều thread sẽ mất update, hỏng cấu trúc khi resize, và iterator ném ConcurrentModificationException. synchronizedMap an toàn nhưng khóa cả map. ConcurrentHashMap từ Java 8 đọc không khóa, ghi dùng CAS cho bucket rỗng và synchronized trên node đầu bucket, nên chỉ tranh chấp khi cùng bucket; nó không cho null key/value.

🇬🇧 English:

HashMap is not thread-safe: shared between threads it can lose updates, corrupt its structure during resize, and its iterator throws ConcurrentModificationException. synchronizedMap is safe but locks the whole map. Since Java 8, ConcurrentHashMap reads without locking and writes use CAS for empty buckets and synchronized on the first node of a bucket, so threads only contend on the same bucket. It does not allow null keys or values.

Q2. Dùng ConcurrentHashMap rồi thì code có chắc đúng không?

🇻🇳 Tiếng Việt:

Không chắc. Mỗi method an toàn không có nghĩa chuỗi method an toàn. get rồi put vẫn có khe hở, nên em dùng computeIfAbsent, merge, compute. Ví dụ đếm đơn theo trạng thái em dùng computeIfAbsent(status, k -> new LongAdder()).increment().

🇬🇧 English:

Not necessarily. Each method being thread-safe does not make a sequence of calls thread-safe. A get followed by a put still has a gap, so I use computeIfAbsent, merge or compute. For example, to count orders by status I use computeIfAbsent with a LongAdder and call increment.

Q3. CopyOnWriteArrayList và BlockingQueue dùng khi nào?

🇻🇳 Tiếng Việt:

CopyOnWriteArrayList cho list đọc rất nhiều, ghi rất ít như danh sách listener, vì mỗi lần ghi copy cả mảng. BlockingQueue cho producer–consumer; nên dùng queue có giới hạn để tạo backpressure và tránh OOM. ThreadPoolExecutor bên trong cũng là một BlockingQueue với thứ tự core, queue, max rồi reject.

🇬🇧 English:

CopyOnWriteArrayList is for lists that are read very often and rarely modified, such as listener lists, because every write copies the whole array. BlockingQueue is for producer-consumer; a bounded queue gives backpressure and avoids out-of-memory errors. ThreadPoolExecutor is itself a BlockingQueue plus workers, filling core threads, then the queue, then up to max, then rejecting.

Cách nhớ ngắn gọn

• HashMap: không an toàn. synchronizedMap: một khóa. CHM: khóa từng bucket, đọc không khóa.
• An toàn từng method ≠ an toàn cả chuỗi → computeIfAbsent, merge.
• Đọc nhiều ghi ít → CopyOnWrite. Producer–consumer → BlockingQueue có giới hạn.
• Pool: core → queue → max → reject.