1. CountDownLatch: khởi động theo tín hiệu
Trong thế giới đa luồng, ta thường cần sắp xếp cho một nhóm luồng hoạt động nhịp nhàng — để tất cả cùng bắt đầu, cùng kết thúc hoặc cùng chuyển sang giai đoạn kế tiếp. Ví dụ:
Hãy hình dung một cuộc đua. Xe đứng ở vạch xuất phát — có xe đã làm ấm động cơ, có xe vẫn đang kiểm tra lốp. Nhưng cho đến khi trọng tài phất cờ, không ai được xuất phát. Đó chính là bài toán phối hợp.
Hoặc ví dụ khác: bạn nấu bữa tối với bạn bè — người thì cắt rau, người thì đun nước, người thì đi tìm xem muối biến đâu. Quan trọng là mọi người phải chuẩn bị xong trước khi bắt đầu nấu.
Cho những trường hợp như vậy, Java cung cấp các công cụ đồng bộ hóa sẵn có — an toàn, dễ hiểu và không còn đau đầu vì wait() và notify(). Một trong những công cụ hữu ích nhất là CountDownLatch. Nó hoạt động như một khóa-đếm: khi số đếm chưa về 0, “cánh cửa” đóng và không ai đi tiếp. Khi tất cả đã điểm danh — latch mở ra và các luồng đồng loạt lao vào cuộc.
CountDownLatch
CountDownLatch là “van một lần” cho phép một hoặc nhiều luồng chờ đến khi các luồng khác hoàn thành một số lượng thao tác nhất định.
Giống như xuất phát marathon: tất cả vận động viên đứng ở vạch, chờ tiếng súng lệnh. Ngay khi trọng tài bóp cò (countdown về 0) — tất cả bắt đầu chạy.
Nó hoạt động thế nào
CountDownLatch giống như còi xuất phát cho các luồng. Khi tạo, bạn chỉ định một con số — ví dụ, 3. Đó là ba tín hiệu cần nhận trước khi cuộc đua bắt đầu.
Những luồng cần chờ khởi động sẽ gọi await(). Họ đứng ở vạch sẵn sàng lao đi, nhưng vẫn còn “đạp phanh”. Các luồng khác, khi hoàn tất chuẩn bị, lần lượt gọi countDown() — như thể gửi tín hiệu: “Tôi đã sẵn sàng!”.
Khi bộ đếm về 0 — bùm! — tất cả các luồng đang chờ đồng thời xuất phát.
Nhưng hãy nhớ: CountDownLatch là thứ dùng một lần. Khi bộ đếm đã về 0, không thể đặt lại. Nó không phải ổ quay mà là pháo giấy: nổ một cái — hết.
Ví dụ: chờ N tác vụ hoàn thành
import java.util.concurrent.CountDownLatch;
public class LatchDemo {
public static void main(String[] args) throws InterruptedException {
int workers = 3;
CountDownLatch latch = new CountDownLatch(workers);
for (int i = 1; i <= workers; i++) {
int id = i;
new Thread(() -> {
System.out.println("Nhân viên " + id + " bắt đầu làm việc");
try { Thread.sleep(500 + id * 200); } catch (InterruptedException ignored) {}
System.out.println("Nhân viên " + id + " đã hoàn thành công việc");
latch.countDown(); // giảm bộ đếm
}).start();
}
System.out.println("Luồng chính đang đợi tất cả nhân viên hoàn thành...");
latch.await(); // chờ cho đến khi tất cả nhân viên xong
System.out.println("Tất cả nhân viên đã xong! Tiếp tục công việc chính.");
}
}
Kết quả:
Luồng chính đang đợi tất cả nhân viên hoàn thành...
Nhân viên 1 bắt đầu làm việc
Nhân viên 2 bắt đầu làm việc
Nhân viên 3 bắt đầu làm việc
Nhân viên 1 đã hoàn thành công việc
Nhân viên 2 đã hoàn thành công việc
Nhân viên 3 đã hoàn thành công việc
Tất cả nhân viên đã xong! Tiếp tục công việc chính.
Ví dụ: xuất phát đồng thời “theo tín hiệu”
CountDownLatch startSignal = new CountDownLatch(1);
for (int i = 0; i < 5; i++) {
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + " đang chờ tín hiệu xuất phát");
startSignal.await(); // chờ tín hiệu
System.out.println(Thread.currentThread().getName() + " xuất phát!");
} catch (InterruptedException ignored) {}
}).start();
}
Thread.sleep(1000);
System.out.println("Tín hiệu xuất phát!");
startSignal.countDown(); // tất cả các luồng xuất phát đồng thời
2. CyclicBarrier: nhiều phase, hành động tại barrier
CyclicBarrier: gặp nhau bên đống lửa trại
CyclicBarrier là nơi gặp nhau của các luồng. Mỗi luồng chạy theo lộ trình riêng, làm việc của mình, rồi tất cả cùng tụ về “barrier” — như bên đống lửa trên núi. Khi mọi người đã có mặt, barrier mở ra, cả nhóm cùng đi tiếp.
Khác biệt chính so với CountDownLatch — barrier này có thể dùng lặp lại. Sau mỗi lần dừng chung, nó “nạp lại” và đội có thể tiếp tục sang giai đoạn tiếp theo.
Hãy tưởng tượng: Một nhóm phượt thủ đi đường dài. Mỗi người đi theo nhịp riêng: có người chụp ảnh bướm, có người săn Wi‑Fi. Nhưng ở mỗi con đèo, họ gặp nhau bên đống lửa, chờ nhau và quyết định đi tiếp. Đó chính là CyclicBarrier đang hoạt động.
Cách hoạt động
Bạn tạo một barrier và chỉ định số lượng thành viên cần tụ lại, ví dụ 4. Mỗi luồng khi tới điểm kiểm soát sẽ gọi await() — và chờ những người khác. Khi đủ bốn người, barrier “nhả” và cho tất cả đi tiếp.
Bạn thậm chí có thể đặt “hành động tại barrier” — một đoạn mã chạy đúng một lần khi nhóm đã tụ đủ. Ví dụ, nhóm đốt lửa hoặc ghi log: “Hoàn thành giai đoạn, đi tiếp”. Để làm vậy, truyền một Runnable vào constructor.
Quan trọng: khác với CountDownLatch dùng một lần, CyclicBarrier dùng nhiều lần. Sau mỗi lần “tụ họp”, nó lại sẵn sàng cho giai đoạn kế tiếp — như một đống lửa trại có thể nhóm lại nhiều lần.
Ví dụ: đồng bộ hóa các phase
import java.util.concurrent.CyclicBarrier;
public class BarrierDemo {
public static void main(String[] args) {
int parties = 3;
CyclicBarrier barrier = new CyclicBarrier(parties, () -> {
System.out.println("Mọi người đã đến barrier! Bắt đầu phase mới.");
});
for (int i = 1; i <= parties; i++) {
int id = i;
new Thread(() -> {
try {
System.out.println("Luồng " + id + " làm việc ở phase 1");
Thread.sleep(300 + id * 200);
System.out.println("Luồng " + id + " chờ barrier");
barrier.await(); // chờ những người còn lại
System.out.println("Luồng " + id + " làm việc ở phase 2");
Thread.sleep(200 + id * 100);
System.out.println("Luồng " + id + " chờ barrier (2)");
barrier.await(); // lại chờ
System.out.println("Luồng " + id + " đã hoàn thành công việc");
} catch (Exception e) {
System.out.println("Lỗi: " + e);
}
}).start();
}
}
}
Kết quả:
Luồng 1 làm việc ở phase 1
Luồng 2 làm việc ở phase 1
Luồng 3 làm việc ở phase 1
Luồng 1 chờ barrier
Luồng 2 chờ barrier
Luồng 3 chờ barrier
Mọi người đã đến barrier! Bắt đầu phase mới.
Luồng 1 làm việc ở phase 2
...
Hành động tại barrier
Bạn có thể truyền vào constructor của CyclicBarrier một hành động (Runnable) sẽ chạy một lần khi tất cả các luồng đã đến barrier (ví dụ, cập nhật trạng thái, in log).
Bẫy: nếu một luồng bị lỗi thì sao?
Nếu một trong các luồng ném ra ngoại lệ hoặc không đến được barrier, các luồng còn lại sẽ chờ mãi — hoặc nhận BrokenBarrierException. Barrier bị “hỏng” và cần tạo lại.
Đây là cách có thể viết lại phần này sinh động, giàu hình ảnh và gần gũi hơn — để nó như một phần tiếp nối tự nhiên của chủ đề “dàn nhạc”:
3. Phaser: người nhạc trưởng tài tình của buổi hòa nhạc lớn
Phaser — như một “siêu barrier”. Nó kết hợp những điểm hay của CountDownLatch và CyclicBarrier, đồng thời linh hoạt hơn nhiều. Hãy coi như một dàn nhạc: nhạc công có thể ra vào giữa các phần của buổi diễn, nhưng nhạc trưởng vẫn đảm bảo mỗi phần chỉ bắt đầu khi mọi người đã sẵn sàng.
Khác với barrier thông thường, Phaser làm việc theo từng giai đoạn — các phase nối tiếp nhau. Có người chỉ chơi ở phần đầu, có người vào sau, có người rời sớm — Phaser xử lý tất cả điều đó một cách bình thản.
Nó hoạt động thế nào
Đầu tiên tạo Phaser, thường với số lượng thành viên đã biết — parties. Mỗi luồng đăng ký (register()), thực hiện phần việc của mình và cuối phase gọi arriveAndAwaitAdvance() — báo đã xong và chờ những người khác. Khi mọi người đến điểm này, Phaser chuyển sang phase tiếp theo và lặp lại quy trình.
Nếu một thành viên không cần nữa — họ có thể “cúi chào” và rời sân khấu qua arriveAndDeregister(). Ngược lại, thành viên mới có thể tham gia ngay trong buổi diễn — qua register().
Khi nào Phaser tốt hơn Barrier
Phaser đáng chọn nếu chương trình của bạn không chỉ có một nhịp điệu duy nhất mà có nhiều nhịp:
- số lượng luồng thay đổi khi chạy,
- có nhiều giai đoạn và không phải mọi thành viên đều tham gia tất cả,
- hoặc đơn giản là muốn tối đa linh hoạt mà không phải tự đồng bộ thủ công rườm rà.
Về bản chất, Phaser là nhạc trưởng. Nó không chỉ vung đũa mà còn thích ứng với thành phần dàn nhạc, số phần của buổi diễn, thậm chí cả việc ai đó đến muộn hoặc rời sớm.
Ví dụ: xử lý theo giai đoạn với số luồng động
import java.util.concurrent.Phaser;
public class PhaserDemo {
public static void main(String[] args) {
Phaser phaser = new Phaser(1); // luồng chính
for (int i = 1; i <= 3; i++) {
phaser.register(); // đăng ký participant
int id = i;
new Thread(() -> {
for (int phase = 1; phase <= 2; phase++) {
System.out.println("Luồng " + id + " làm việc ở phase " + phase);
try { Thread.sleep(200 + id * 100); } catch (InterruptedException ignored) {}
phaser.arriveAndAwaitAdvance(); // chờ những người còn lại
}
System.out.println("Luồng " + id + " đã hoàn thành công việc");
phaser.arriveAndDeregister(); // rời khỏi phaser
}).start();
}
// Luồng chính cũng tham gia các phase
for (int phase = 1; phase <= 2; phase++) {
phaser.arriveAndAwaitAdvance();
System.out.println("Luồng chính: đã hoàn thành phase " + phase);
}
phaser.arriveAndDeregister();
System.out.println("Tất cả các phase đã hoàn thành!");
}
}
Đặc điểm:
- Có thể thêm/xóa thành viên ngay trong lúc chạy.
- Có thể lấy số phase hiện tại: phaser.getPhase().
- Có thể kết thúc phaser: phaser.forceTermination().
4. Exchanger: trao đổi dữ liệu theo cặp giữa các luồng
Exchanger<T> là bộ đồng bộ cho việc trao đổi dữ liệu giữa hai luồng. Mỗi luồng gọi exchange(data), và khi cả hai gặp nhau, họ hoán đổi dữ liệu.
Ví von: Hai người giao hàng gặp nhau ở ngã tư và trao đổi gói hàng.
Hoạt động thế nào?
- Một luồng gọi exchange(data1) — chờ luồng thứ hai.
- Luồng thứ hai gọi exchange(data2) — cả hai nhận dữ liệu của nhau.
- Nếu luồng thứ hai chưa đến — luồng thứ nhất sẽ chờ (có thể đặt timeout).
Ví dụ: trao đổi buffer giữa producer và consumer
import java.util.concurrent.Exchanger;
public class ExchangerDemo {
public static void main(String[] args) {
Exchanger<String> exchanger = new Exchanger<>();
// Producer
new Thread(() -> {
String data = "Dữ liệu từ producer";
try {
System.out.println("Producer: gửi dữ liệu");
String response = exchanger.exchange(data);
System.out.println("Producer: nhận phản hồi: " + response);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// Consumer
new Thread(() -> {
try {
String received = exchanger.exchange("Phản hồi từ consumer");
System.out.println("Consumer: nhận dữ liệu: " + received);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
Kết quả:
Producer: gửi dữ liệu
Consumer: nhận dữ liệu: Dữ liệu từ producer
Producer: nhận phản hồi: Phản hồi từ consumer
Ứng dụng:
- Trao đổi buffer giữa các luồng (ví dụ, một luồng đọc từ file, luồng khác ghi ra mạng).
- Đồng bộ các phase giữa hai luồng.
5. Thực hành: xử lý pipeline song song
Bài toán: “tick” trò chơi (các phase)
Giả sử có vài luồng, mỗi luồng phụ trách một phần của thế giới trò chơi (ví dụ vật lý, AI, render). Tất cả cần đồng bộ ở mỗi “tick” (phase) để tránh lệch pha.
Giải pháp: Dùng CyclicBarrier hoặc Phaser.
import java.util.concurrent.CyclicBarrier;
public class GameTickDemo {
public static void main(String[] args) {
int subsystems = 3;
CyclicBarrier barrier = new CyclicBarrier(subsystems, () -> {
System.out.println("Tất cả các subsystem đã hoàn thành tick. Bắt đầu tick tiếp theo.");
});
for (int i = 1; i <= subsystems; i++) {
int id = i;
new Thread(() -> {
for (int tick = 1; tick <= 5; tick++) {
System.out.println("Subsystem " + id + " làm việc ở tick " + tick);
try { Thread.sleep(100 + id * 50); } catch (InterruptedException ignored) {}
try {
barrier.await();
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
}
}
Bài toán: “van” cho số lượng lớn worker
Giả sử có 100 luồng worker cần xuất phát đồng thời sau khi chuẩn bị xong (ví dụ, kiểm thử tải).
Giải pháp: Dùng CountDownLatch.
import java.util.concurrent.CountDownLatch;
public class MassStartDemo {
public static void main(String[] args) throws InterruptedException {
int workers = 100;
CountDownLatch ready = new CountDownLatch(workers);
CountDownLatch start = new CountDownLatch(1);
for (int i = 0; i < workers; i++) {
new Thread(() -> {
System.out.println("Luồng sẵn sàng xuất phát");
ready.countDown(); // báo hiệu đã sẵn sàng
try {
start.await(); // chờ tín hiệu chung
System.out.println("Luồng xuất phát!");
} catch (InterruptedException ignored) {}
}).start();
}
ready.await(); // chờ cho đến khi tất cả luồng sẵn sàng
System.out.println("Mọi người đã sẵn sàng! BẮT ĐẦU!");
start.countDown(); // phát tín hiệu xuất phát
}
}
6. Lỗi thường gặp khi làm việc với các bộ đồng bộ
Lỗi №1: Dùng CountDownLatch như một barrier dùng nhiều lần.
CountDownLatch là công cụ dùng một lần! Sau khi về 0 thì không thể “nạp lại”. Với các phase dùng nhiều lần, hãy dùng CyclicBarrier hoặc Phaser.
Lỗi №2: Không xử lý ngoại lệ (InterruptedException, BrokenBarrierException).
Các phương thức await() có thể ném ngoại lệ — luôn xử lý chúng, nếu không luồng có thể “treo” hoặc kết thúc với lỗi. Theo dõi InterruptedException và BrokenBarrierException.
Lỗi №3: Một trong các luồng không đến barrier.
Nếu một luồng “rơi” hoặc không gọi await(), các luồng còn lại sẽ chờ mãi (hoặc nhận BrokenBarrierException). Hãy đảm bảo mọi thành viên đều đến barrier.
Lỗi №4: Quên gọi deregister() trong Phaser.
Nếu luồng đã xong nhưng không gọi arriveAndDeregister(), Phaser sẽ chờ một thành viên “chết”. Luôn xóa luồng khỏi Phaser một cách đúng đắn.
Lỗi №5: Dùng Exchanger cho nhiều hơn hai luồng.
Exchanger chỉ làm việc cho trao đổi giữa hai luồng. Nếu nhiều hơn — sẽ dẫn đến deadlock.
Lỗi №6: Trộn nhiều bộ đồng bộ khác nhau mà không hiểu rõ cách hoạt động.
Không nên đồng thời dùng nhiều barrier/latch khác nhau cho cùng một nhóm luồng — điều này có thể dẫn tới rối rắm và treo.
GO TO FULL VERSION