Concurrency nâng cao

Từ happens-before đến executor sizing, synchronizer, concurrent data structure, Fork/Join, reactive streams và cancellation có cấu trúc.

1. Java Memory Model sâu hơn

Data race tồn tại khi hai thread truy cập cùng biến, ít nhất một write, không có ordering bởi happens-before. Happens-before không phải thời gian thực; nó là bảo đảm visibility/ordering của memory action. Các cạnh quan trọng gồm program order, monitor unlock→lock, volatile write→read, thread start và task completion/join.

Safe publication có thể đạt qua static initialization, volatile reference, final-field semantics đúng, lock hoặc concurrent collection. Publish reference của object trước khi constructor hoàn tất (“this escape”) có thể khiến thread khác thấy state chưa đúng.

2. Executor và queue policy

Thread pool không chỉ là số thread. Cần quyết định queue bounded/unbounded, rejection policy, task duration, blocking ratio, priority và shutdown. Unbounded queue che overload bằng latency/memory tăng; pool quá lớn gây context switching và downstream overload.

var executor = new ThreadPoolExecutor(
    8, 16, 30, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(500),
    new ThreadPoolExecutor.CallerRunsPolicy());

CallerRunsPolicy có thể tạo backpressure thô bằng cách làm producer chậm lại. Nhưng ở event-loop/request thread nó cũng có thể gây tác dụng phụ, nên phải hiểu caller.

Lifecycle và interruption

Interruption là cooperative cancellation signal. API blocking thường ném InterruptedException; code không xử lý thì nên restore flag bằng Thread.currentThread().interrupt() rồi propagate/return. Không dùng Thread.stop.

3. Locks và synchronizers

Condition wait luôn đặt trong loop vì spurious wakeup và predicate có thể không còn đúng khi reacquire lock.

lock.lock();
try {
    while (queue.isEmpty()) notEmpty.await();
    return queue.removeFirst();
} finally { lock.unlock(); }

4. Atomic, LongAdder và VarHandle

CAS đọc state kỳ vọng và chỉ write nếu chưa đổi. Under contention, retry có thể tốn CPU; lock-free không đồng nghĩa wait-free. AtomicReference hỗ trợ immutable state transition. LongAdder phân tán contention cho counter throughput cao nhưng sum() không phải atomic snapshot.

ABA xảy ra khi value đổi A→B→A khiến CAS chỉ nhìn value tưởng chưa đổi; AtomicStampedReference hoặc thiết kế version có thể xử lý. VarHandle cho library author kiểm soát memory ordering tinh hơn volatile.

5. Concurrent collections

ConcurrentHashMap không cho null để tránh ambiguity trong concurrent lookup. Iterator weakly consistent: không throw ConcurrentModificationException và có thể phản ánh một phần update. BlockingQueue kết hợp storage với wait/notification, hữu ích producer-consumer và backpressure; chọn bounded capacity khi cần giới hạn memory.

CopyOnWriteArrayList tối ưu read/iteration snapshot khi write rất hiếm; mỗi write copy toàn array nên không phù hợp mutation thường xuyên. ConcurrentLinkedQueue là non-blocking queue nhưng không tự cung cấp capacity/backpressure.

6. Fork/Join và parallelism

ForkJoinPool dùng work stealing cho recursive, CPU-bound tasks có thể chia nhỏ. Task quá nhỏ tăng overhead; blocking operation có thể làm worker thiếu hụt. Parallel stream chạy trên common pool theo mặc định, có thể tranh chấp với code khác trong process.

CPU-bound pool thường quanh số core; blocking workload cần ước lượng wait/compute ratio nhưng tốt hơn là đo. Với virtual threads, không sizing thread pool theo công thức cũ; giới hạn resource bên dưới.

7. Virtual threads: thuật ngữ và migration evidence

Thuật ngữ: virtual thread là lightweight thread do JVM quản lý; platform thread ánh xạ gần với OS thread; carrier thread là platform thread đang thực thi một virtual thread. Mô hình thread-per-request giữ một luồng logic riêng cho mỗi request mà không yêu cầu một OS thread riêng trong toàn bộ thời gian chờ I/O.

Theo Little's Law, concurrency xấp xỉ throughput nhân latency. Vì vậy virtual threads giúp duy trì nhiều request đang chờ, nhưng không làm CPU computation nhanh hơn và không tăng capacity của database hay downstream. Khi migration, hãy hỏi: latency tăng sẽ đòi bao nhiêu concurrency, và resource nào phải được bulkhead?

Khả năng giữ stack trace dễ đọc và tương thích với nhiều code dùng ThreadLocal là lợi ích migration quan trọng so với callback chains. Tuy nhiên vẫn phải audit lifecycle của ThreadLocal, giới hạn resource bên dưới và đo tail latency, memory, connection wait cùng throughput.

8. Flow và backpressure

java.util.concurrent.Flow định nghĩa Publisher, Subscriber, Subscription và Processor theo Reactive Streams semantics. Subscriber yêu cầu số item qua request(n); publisher không được phát quá demand. Backpressure là protocol kiểm soát tốc độ, khác với chỉ buffering vô hạn.

submissionPublisher.subscribe(new Flow.Subscriber<>() {
    public void onSubscribe(Flow.Subscription s) { subscription = s; s.request(1); }
    public void onNext(Event e) { try { handle(e); } finally { subscription.request(1); } }
    public void onError(Throwable error) { log(error); }
    public void onComplete() { }
});

9. Testing concurrency

Sleep-based test dễ flaky. Dùng latch/barrier để điều khiển interleaving, timeout để tránh test treo và chạy lặp stress. Assert invariant thay vì chỉ output cuối. Với low-level JMM/lock-free algorithm, cân nhắc jcstress; benchmark dùng JMH, không tự đo bằng một vòng nanoTime thiếu warmup.