Synchronizers: Điều phối tiến độ giữa nhiều thread

TL;DR: Synchronizer điều phối tiến độ giữa các thread — ai chờ ai, ai được đi — khác với lock/atomic vốn bảo vệ dữ liệu. CountDownLatch cho sự kiện một lần: đếm về 0 rồi mở vĩnh viễn, không reset. CyclicBarrier cho một nhóm cố định hẹn nhau lặp lại tại các mốc, kèm barrier action giữa hai pha; một thành viên hỏng thì cả nhóm nhận BrokenBarrierException. Semaphore giới hạn số lượng truy cập đồng thời bằng permit không có chủ — acquire/release phải đi cặp qua try/finally kẻo permit rò rỉ. Phaser tổng quát hóa cả latch lẫn barrier với số party động. Pitfall lớn nhất: họ method arrive* của Phaser chỉ báo hiệu rồi return ngay, không chờ — muốn chờ phải gọi awaitAdvance.

1. Điều phối tiến độ khác bảo vệ dữ liệu thế nào?

Bài trước khép lại ở BlockingQueue, một component mà giá trị nằm ở chỗ nó block: put chờ khi hàng đầy, take chờ khi rỗng. Cái queue ấy làm hai việc cùng lúc — vừa chứa dữ liệu an toàn, vừa điều phối tiến độ, quyết định thread nào đi tiếp, thread nào dừng đợi. Hai việc đó là hai bài toán khác nhau; BlockingQueue chỉ tình cờ gộp cả hai.

Từ bài Thread safety tới giờ ta dồn sức cho loại thứ nhất: nhiều thread cùng chạm một mẩu dữ liệu mà không phá invariant — lock, atomic, immutable, concurrent collection đều canh giữ shared mutable state. Loại thứ hai khác hẳn: cái cần sắp xếp không phải một mẩu dữ liệu, mà là nhịp đi của các thread. Một thread đợi ba thread khác khởi tạo xong; năm thread phải cùng có mặt tại một mốc trước khi ai được đi tiếp; hệ thống chỉ cho mười request vào tài nguyên hiếm cùng lúc. Không có biến chung nào bị tranh chấp; cái được điều phối là tiến độ.

Đó là việc của synchronizer: đối tượng điều tiết luồng đi của các thread dựa trên trạng thái của chính nó. Bài này đi qua năm cái trong java.util.concurrent — CountDownLatch, CyclicBarrier, Semaphore, Exchanger, Phaser — mỗi cái trả lời một câu hỏi điều phối, đều là component đã kiểm chứng, không phải thứ ta nên tự dựng bằng wait/notify.

2. CountDownLatch chờ N sự kiện hoàn tất thế nào?

Hình dung một buổi đua: mọi vận động viên đã vào vị trí, nhưng cuộc đua chưa bắt đầu chừng nào trọng tài chưa kiểm xong từng làn — mỗi hạng mục xong, một ngọn đèn tắt. Đèn cuối tắt, cổng mở, tất cả cùng xuất phát. Cái bảng đèn đó là một latch: giữ cổng đóng tới khi một số điều kiện đều thỏa, rồi mở vĩnh viễn.

CountDownLatch khởi tạo với một số đếm. Thread gọi await() block tới khi số đếm về 0; mỗi sự kiện hoàn tất, ai đó gọi countDown() giảm số đếm đi một. Chạm 0, mọi thread đang chờ được thả cùng lúc, và mọi await() về sau trả về ngay. Chữ quan trọng nhất là "vĩnh viễn": latch dùng một lần, đã về 0 thì không reset — nó mô hình hóa một sự kiện chỉ xảy ra một lần trong đời ("khởi tạo đã xong"), không phải mốc lặp lại.

Có hai cách dùng đối xứng. Cách thứ nhất: một hay nhiều thread chờ đến khi N việc khởi tạo hoàn tất.

CountDownLatch ready = new CountDownLatch(3);   // 3 buoc warm-up
for (Runnable step : List.of(this::loadEvents, this::primeCache, this::verifyDependencies)) {
    Thread.startVirtualThread(() -> {
        try { step.run(); }
        finally { ready.countDown(); }           // luon dem, ke ca khi loi
    });
}
ready.await();                                   // block toi khi ca 3 dem xong

Đặt countDown() trong finally không phải tiểu tiết: nếu một bước warm-up ném exception mà bỏ qua nó, số đếm không bao giờ về 0 và thread await() treo vĩnh viễn. Latch không có khái niệm "thất bại"; nó chỉ đếm — ta phải bảo đảm nó luôn đếm đủ.

Cách thứ hai đảo vai: dùng latch như "súng lệnh" để N thread cùng bắt đầu tại một thời điểm — mẹo kinh điển viết test đo contention, để mọi worker đồng loạt lao vào cùng lúc thay vì thread đầu chạy xong trước khi thread cuối kịp khởi động.

CountDownLatch startGun = new CountDownLatch(1);            // cong xuat phat
CountDownLatch finished = new CountDownLatch(workerCount);  // vach dich
// moi worker chay: startGun.await(); doWork(); finished.countDown();
startGun.countDown();   // ban — tat ca worker xuat phat cung luc
finished.await();       // cho tat ca ve dich

Hai latch làm hai việc tách bạch: startGun (đếm từ 1) là cổng xuất phát, finished (đếm từ workerCount) là vạch đích. Đó là toàn bộ vốn từ vựng của CountDownLatch, và phần lớn bài toán "chờ một nhóm sự kiện một lần" chỉ cần chừng đó.

3. CyclicBarrier: N thread hẹn gặp nhau tại một điểm

Latch có một bất đối xứng tinh tế: thread gọi await() và thread gọi countDown() thường khác nhau — người chờ không phải người báo tin. CyclicBarrier xóa bỏ điều đó: mọi thread bình đẳng, cùng chạy đến một mốc, cùng gọi await(), cùng dừng đợi tới khi người cuối cùng tới, rồi rào mở và tất cả cùng đi tiếp. Hình dung một đoàn leo núi: mỗi trạm nghỉ cả đoàn phải tụ đủ mới đi tiếp; người tới sớm ngồi chờ, trạm kế lặp lại y hệt.

Chữ "cyclic" là khác biệt lớn nhất so với latch: sau khi tất cả vượt qua, barrier tự reset, sẵn sàng cho vòng tiếp. Đây là công cụ cho thuật toán lặp theo pha: chạy một pha, đồng bộ tại barrier để chắc mọi thread đã xong, rồi cùng bước vào pha sau.

// barrier action chay 1 lan moi khi ca nhom toi moc, TRUOC khi ai duoc di tiep
var barrier = new CyclicBarrier(workers, this::mergeAndPublishBoard);

Runnable worker = () -> {
    for (int step = 0; step < steps; step++) {
        computeRegion(myRegion);        // pha tinh toan doc lap tung vung
        try { barrier.await(); }        // cho moi vung tinh xong pha nay
        catch (InterruptedException | BrokenBarrierException e) { return; }
    }
};

CyclicBarrier cho ta thứ latch không có: barrier action — một Runnable truyền vào constructor, chạy đúng một lần mỗi khi rào mở, trên thread tới cuối cùng, trước khi bất kỳ thread nào được thả. Đó là khe an toàn để gộp kết quả pha vừa rồi: mọi thread đã xong việc, chưa ai sang pha mới, nên đọc gộp không bị chen ngang (ở trên, mergeAndPublishBoard hợp nhất kết quả các vùng thành bàn cờ mới).

Cái tên BrokenBarrierException kể đúng bản chất: barrier là giao ước nhóm nên mong manh theo nhóm. Nếu một thread đang đợi bị interrupt, một await(timeout, unit) hết hạn, hoặc barrier action ném exception, barrier rơi vào trạng thái "broken" và mọi thread khác đang chờ lập tức bị đánh thức bằng BrokenBarrierException. Hợp lý thôi — nếu một thành viên không bao giờ tới mốc, cả nhóm không thể đợi mãi; thà báo giao ước đã đổ vỡ còn hơn treo vô hạn. Barrier vỡ phải reset() mới dùng lại được.

Đặt hai vòng đời cạnh nhau, khác biệt một-lần vs lặp-lại hiện rõ:

flowchart TB
    subgraph LATCH["CountDownLatch: dem ve 0 mot chieu"]
        L3["count = 3"] -->|"countDown()"| L2["count = 2"]
        L2 -->|"countDown()"| L1["count = 1"]
        L1 -->|"countDown()"| L0["count = 0: cong mo VINH VIEN"]
        LW["await() block"] -.->|"duoc tha khi ve 0"| L0
    end
    subgraph BAR["CyclicBarrier: hen nhau, lap vong"]
        B1["N thread lan luot await()"] --> B2["nguoi cuoi cham moc"]
        B2 --> B3["chay barrier action"]
        B3 --> B4["ca nhom cung di tiep"]
        B4 -->|"tu reset, vong moi"| B1
    end
    style L0 fill:#6EE7B7
    style B4 fill:#93C5FD

Trái là đường thẳng một chiều (đếm cạn rồi mở mãi), phải là vòng tròn (chạm mốc, action chạy, rào mở, lặp lại). Ranh giới chọn đã rõ — CountDownLatch khi sự kiện một lần và người chờ khác người báo; CyclicBarrier khi một nhóm cố định gặp nhau lặp lại và cần làm gì đó giữa các mốc.

4. Semaphore: giới hạn số thread truy cập đồng thời

Hai synchronizer vừa rồi nói về thời điểm: chờ tới khi, gặp nhau tại. Semaphore nói về số lượng: tại một thời điểm tối đa bao nhiêu thread được vào. Analogy chuẩn là bãi đỗ xe N chỗ có rào chắn: mỗi xe vào lấy một vé, hết chỗ thì rào đóng và xe sau xếp hàng, mỗi xe ra trả vé cho một xe chờ vào. Bãi chỉ giữ một bất biến: số xe bên trong không vượt N.

Semaphore quản lý một tập permit y như tập vé đó. acquire() lấy một permit, block nếu hết; release() trả permit về, đánh thức một thread đang chờ. Số permit ban đầu chính là số truy cập đồng thời tối đa.

public <T> T withConnection(Callable<T> action) throws Exception {
    permits.acquire();                  // cho toi khi co cho
    try {
        return action.call();
    } finally {
        permits.release();              // luon tra permit, ke ca khi action nem
    }
}

Cấu trúc acquire() ... try/finally ... release() gần như bắt buộc: nếu action ném mà ta quên release(), một permit biến mất vĩnh viễn. Lặp vài lần, semaphore cạn và mọi thread sau block mãi — resource leak âm thầm, khó truy hơn deadlock vì không có vòng chờ nào để soi.

Một đặc tính tinh tế: semaphore không gắn permit với thread. Khác intrinsic lock vốn nhớ thread đang giữ (và nhờ đó reentrant), permit không có chủ — thread A acquire() còn thread B release() vẫn hợp lệ, tổng permit chỉ là một con số đếm. Điều đó khiến nó linh hoạt hơn lock (mô hình "một bên cấp, bên khác thu") nhưng không bảo vệ ta khỏi lỗi đếm sai: release() nhiều hơn acquire() đẩy số permit vượt giá trị khởi tạo, phá vỡ chính cái giới hạn ta dựng nó để giữ.

Constructor còn tùy chọn fair: new Semaphore(permits, true) cấp permit cho thread chờ lâu nhất trước. Mặc định (non-fair) cho thread mới "chen ngang" khi permit vừa trả — throughput cao hơn nhưng có thể bỏ đói thread chờ lâu dưới tải dồn; resource pool phục vụ user thật thường đáng cái giá fair đó.

Ở front-end thường không muốn block khi hết chỗ: tryAcquire() cố lấy một permit rồi trả boolean tức thì, không bao giờ chờ — thà nói "hết suất, thử lại sau" còn hơn treo trình duyệt.

Công dụng kinh điển nhất là biến một collection thường thành bounded resource pool: permit bằng số tài nguyên, lấy trước khi mượn, trả sau khi trả. N = 1 cho một binary semaphore dùng như mutex — nhưng không reentrant và không có chủ, nên trừ khi cần "release từ thread khác", ReentrantLock thường đúng hơn cho loại trừ lẫn nhau thuần túy.

5. Exchanger: trao đổi dữ liệu giữa đúng hai thread

Các synchronizer tới giờ điều phối nhiều thread theo số đếm. Exchanger<T> là ca rất hẹp: điểm hẹn cho đúng hai thread đổi dữ liệu cho nhau. Method đáng kể duy nhất là exchange(x) — thread gọi block tới khi một thread thứ hai gọi exchange(y) trên cùng đối tượng; lúc đó bên đầu nhận y, bên kia nhận x, cả hai đi tiếp. Hình dung hai điệp viên trao vali ở ga: ai tới trước đứng chờ, trao đổi chỉ xong khi cả hai có mặt — một rendezvous đối xứng.

Thực dụng mà nói, Exchanger ít gặp nhất trong bộ năm: đa số bài toán "đưa dữ liệu từ thread này sang thread kia" được BlockingQueue ở bài trước phục vụ tốt và tổng quát hơn (không giới hạn hai thread, không bắt cả hai phía có mặt cùng lúc). Biết nó tồn tại, nhận ra hình dạng hợp với nó, là đủ.

6. Phaser: barrier đa pha với số party động

CountDownLatch đếm một lần rồi thôi; CyclicBarrier lặp lại nhưng số party cố định cứng từ lúc khởi tạo — cả hai đều đòi số "người tham gia" biết trước và không đổi. Phaser (Java 7) gỡ cả hai hạn chế: vừa lặp theo pha như barrier, vừa cho thread gia nhập và rời đi giữa chừng. Hình dung một cuộc họp nhiều phiên nơi người dự được đến muộn hoặc về sớm: mỗi phiên chỉ kết thúc khi mọi người đang có mặt phát biểu xong, mà danh sách đó thay đổi giữa các phiên. Số party là đại lượng động — register() để vào, arriveAndDeregister() để ra.

public class DynamicPipeline {
    private final Phaser phaser = new Phaser(1);    // "1" = thread dieu phoi tu dang ky
    public void addWorker(Runnable task, int phases) {
        phaser.register();                          // worker moi gia nhap, party + 1
        Thread.startVirtualThread(() -> {
            try {
                for (int p = 0; p < phases; p++) {
                    task.run();
                    phaser.arriveAndAwaitAdvance(); // bao "xong pha", cho party khac
                }
            } finally {
                phaser.arriveAndDeregister();       // rut khoi nhom, party - 1
            }
        });
    }
    public void awaitFirstPhase() {
        // arriveAndDeregister CHI bao "toi moc va rut" roi return NGAY, khong cho ai
        int phase = phaser.arriveAndDeregister();   // tra ve so pha hien tai
        phaser.awaitAdvance(phase);                 // block toi khi pha do khep lai
    }
}

Method awaitFirstPhase phơi ra điểm dễ hiểu lầm nhất của Phaser: API của nó tách đôi thứ mà CyclicBarrier.await() gộp làm một. Họ arrive* là nửa báo hiệu — đánh dấu "tôi đã tới mốc" rồi đi tiếp ngay, không chặn (riêng arriveAndAwaitAdvance gộp sẵn cả hai). Còn awaitAdvance(phase) là nửa chờ — block tới khi pha mang số phase khép lại. arriveAndDeregister() thuộc nửa báo hiệu, nên một method "await" mà thân chỉ gọi nó sẽ return ngay mà không chờ gì — bug ngữ nghĩa mà biên dịch vẫn xanh. Cặp đúng: hứng số pha arriveAndDeregister trả về rồi đưa cho awaitAdvance; lời gọi này chờ pha hiện tại khép lại (không phải cả pipeline), và phaser tự terminate khi worker cuối đưa sĩ số về 0.

Vốn từ của Phaser ánh xạ thẳng sang cái đã biết: arriveAndAwaitAdvance() chính là CyclicBarrier.await(), còn register() / arriveAndDeregister() là phần barrier không có (đổi sĩ số nhóm tại chỗ). Nó cũng tái hiện được hành vi kiểu CountDownLatch — nên là thứ gói được cả hai, đổi lại bằng một API rậm rạp hơn.

Còn một nấc nữa: override onAdvance(int phase, int registeredParties) để can thiệp "khi nào pha kết thúc" — trả về true sẽ kết liễu phaser, diễn đạt điều kiện dừng của thuật toán lặp ("chạy tới khi hội tụ" thay vì "đúng K pha"), điểm linh hoạt mà latch lẫn barrier đều không cho.

Khi nào với tay tới Phaser? Khi số thread tham gia thay đổi trong vòng đời tính toán, hoặc điều kiện dừng phụ thuộc trạng thái chứ không phải số vòng cố định — nhóm cố định, bài toán đơn giản thì CountDownLatch/CyclicBarrier dễ đọc hơn, đừng trả giá tính động khi không cần.

7. Tie-in capstone: TicketFlow v2

TicketFlow v2 dùng hai synchronizer này ở đúng hai vai bài vừa mô tả. SalesGate là một CountDownLatch làm cổng mở bán một chiều: các task warm-up (nạp sự kiện, làm nóng cache, kiểm dependency) mỗi cái xong gọi countDown(), luồng phục vụ chờ ở await(), mở xong thì không quay lại được. HoldLimiter là Semaphore bounded resource pool per-event: permit ban đầu bằng capacity mỗi sự kiện nên số lượt giữ chỗ đồng thời không vượt sức chứa, và dùng tryAcquire() để hết slot thì trả false ngay thay vì treo trình duyệt. Hai câu hỏi: latch hỏi "đã đến lúc mở chưa?", semaphore hỏi "còn chỗ để giữ không?" — không cái nào bảo vệ dữ liệu, chúng điều phối tiến độ.

8. Liên hệ các bài khác

• Thread API và vòng đời — mọi lời gọi block của synchronizer (await, acquire, exchange, arriveAndAwaitAdvance) đều phản hồi interrupt; bài đó dạy cách xử lý InterruptedException cho đúng quanh chúng.
• ReentrantLock & Condition — phối hợp "chờ điều kiện rồi được đánh thức" mà ta từng dựng tay bằng lock + condition; synchronizer là tầng đóng gói bên trên cho các hình dạng phối hợp hay gặp nhất.
• ReadWriteLock, StampedLock & AQS — CountDownLatch và Semaphore đều là lớp vỏ mỏng quanh AbstractQueuedSynchronizer; đọc lại bài đó để thấy hàng đợi chờ chung bên dưới.
• Blocking Queues & Producer–Consumer — BlockingQueue cũng là một synchronizer, nhưng kiêm thêm vai chứa dữ liệu; bài này tách riêng phần điều phối tiến độ thuần túy.
• Executor & thread pool — bài kế tiếp mở Phần B: thôi tự tạo thread, bắt đầu tổ chức công việc thành task.

9. Tóm tắt

Nửa series vừa qua ta xoay quanh giữ state đúng khi nhiều thread cùng chạm (lock, atomic, immutability, safe publication, concurrent collection, blocking queue). Bài này tách ra một họ công cụ trả lời câu hỏi khác: không phải "bảo vệ dữ liệu nào", mà "sắp xếp nhịp đi của các thread ra sao".

Năm synchronizer ánh xạ năm hình dạng điều phối; điểm chung: đều là component đã kiểm chứng trong java.util.concurrent, việc của ta là chọn đúng cái cho hình dạng bài toán, không dựng lại bằng wait/notify.

Đến đây khép lại Phần A của series. Nhưng tới giờ ta vẫn tự tay tạo thread khắp nơi (Thread.startVirtualThread(...) rải trong gần như mọi ví dụ). Câu hỏi tiếp theo không còn là "giữ state đúng" mà là "tổ chức và chạy công việc concurrent có kỷ luật" — ai tạo thread, bao nhiêu, tái dùng và dừng thế nào. Đó là Phần B, mở màn bằng Executor và thread pool: thôi nghĩ về thread, bắt đầu nghĩ về task.

10. Tự kiểm tra

Bài tiếp theo: Executor Framework — thread pool, lập lịch & shutdown