Exchanger và Phaser — điều phối tiến độ ngoài ba máy đếm

TL;DR: Ngoài ba máy đếm, java.util.concurrent còn hai synchronizer cho hình dạng hiếm hơn. Exchanger<T> là điểm hẹn cho đúng hai thread đổi dữ liệu cho nhau: mỗi bên gọi exchange(x), block tới khi bên kia tới, rồi cả hai nhận giá trị của nhau — hợp nhất với double buffering, còn lại BlockingQueue thường tổng quát hơn. Phaser (Java 7) tổng quát hóa cả latch lẫn barrier: lặp theo pha và cho thread gia nhập/rời đi giữa chừng bằng register/arriveAndDeregister. Bẫy lớn nhất: họ arrive* chỉ báo hiệu rồi return ngay, không chờ — muốn chờ phải gọi awaitAdvance(phase).

Bài trước mổ ba synchronizer đếm — CountDownLatch, CyclicBarrier, Semaphore — mỗi cái điều phối nhiều thread bằng một con số. Hai hình dạng không vừa khuôn đó: đổi dữ liệu giữa đúng hai thread, và một barrier mà số thành viên thay đổi giữa chừng. Đó là chỗ của Exchanger và Phaser.

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

Ba máy đếm ở bài trước đ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 blocking queues 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à đủ.

2. 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 xong việc, 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
            }
        });
    }
}

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.

Điểm dễ hiểu lầm nhất: API của Phaser 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.
• awaitAdvance(phase) là nửa chờ — block tới khi pha mang số phase khép lại.
• arriveAndAwaitAdvance() gộp sẵn cả hai (báo rồi chờ) — nên nó tương đương CyclicBarrier.await().

flowchart TB
    subgraph SIGNAL["Nua BAO HIEU: return NGAY, khong chan"]
        A1["arrive()"]
        A3["arriveAndDeregister()"]
    end
    subgraph WAIT["Nua CHO: block toi khi pha khep"]
        W1["awaitAdvance(phase)"]
    end
    COMBO["arriveAndAwaitAdvance()<br/>= bao + cho, gop san"]
    A1 -.->|"can cho thi hung phase roi goi"| W1
    A3 -.->|"can cho thi hung phase roi goi"| W1
    style COMBO fill:#93C5FD
    style W1 fill:#6EE7B7

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. Phaser cũng tự terminate khi worker cuối đưa sĩ số về 0.

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.

3. Vì sao một method "await" lại không hề chờ?

Ta vừa chốt: họ arrive* chỉ báo hiệu, awaitAdvance mới chờ. Giờ dùng phân biệt đó để bắt một bug ngữ nghĩa mà trình biên dịch vẫn cho qua. Một service muốn phơi ra method awaitFirstPhase() — chặn caller tới khi pha đầu tiên của pipeline khép lại — và ai đó viết thế này:

public int awaitFirstPhase() {
    return phaser.arriveAndDeregister();   // TODO: co that su cho khong?
}

arriveAndDeregister() thuộc nửa báo hiệu: nó đánh dấu "tôi tới mốc và rút khỏi nhóm" rồi return ngay, mang theo số pha hiện tại — không chờ ai. Nên method trên trả về tức thì, chưa chắc pha đầu đã khép. Đó là bug ngữ nghĩa: tên hứa "await" nhưng thân chỉ "arrive". Trình biên dịch không thể bắt vì kiểu trả về (int số pha) vẫn hợp lệ — chỉ hành vi lúc chạy mới sai.

Sửa thế nào? Hứng số pha mà arriveAndDeregister() trả về, rồi đưa cho awaitAdvance — lời gọi này mới chặn tới khi pha đó khép lại:

public int awaitFirstPhase() {
    int phase = phaser.arriveAndDeregister();   // bao "toi moc + rut", nhan so pha
    return phaser.awaitAdvance(phase);          // CHO toi khi pha do khep lai
}

Quy tắc rút ra để áp cho mọi lần dùng Phaser: khi cần chờ, phải có một lời gọi thuộc họ await (awaitAdvance, awaitAdvanceInterruptibly) hoặc dùng thẳng arriveAndAwaitAdvance(). Nếu code chỉ có arrive* mà kỳ vọng nó chặn, đó là bug — và nó sẽ không lộ ra tới khi một pha chạy nhanh hơn dự tính.

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

• Synchronizers: CountDownLatch, CyclicBarrier, Semaphore — ba máy đếm mà Phaser tổng quát hóa; đọc để thấy arriveAndAwaitAdvance() chính là CyclicBarrier.await() và vì sao Phaser gói được cả latch.
• Blocking Queues & Producer–Consumer — BlockingQueue là lựa chọn mặc định để chuyển dữ liệu giữa thread; Exchanger chỉ thắng ở ca đối xứng đúng hai bên (double buffering).
• Thread API và vòng đời — exchange, arriveAndAwaitAdvance, awaitAdvance đều là lời gọi block phản hồi interrupt; bài đó dạy xử lý InterruptedException cho đúng.
• 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.

5. Tóm tắt

Exchanger và Phaser phủ hai hình dạng điều phối mà ba máy đếm không vừa khuôn:

Exchanger hẹp và hiếm — nhớ nó tồn tại, nhận ra double buffering là hình dạng của nó, là đủ. Phaser mạnh nhưng API rậm: chỉ trả giá độ phức tạp khi số party động hoặc điều kiện dừng phụ thuộc trạng thái; và luôn nhớ họ arrive* chỉ báo hiệu, muốn chờ phải awaitAdvance.

Đến đây khép lại Phần A của series — năm synchronizer, năm hình dạng điều phối, đề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 chứ không dựng lại bằng wait/notify. 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.

6. Tự kiểm tra

Bài tiếp theo: Executor Framework — thread pool và thuật toán nhận task