Tổng quan về công cụ đồng bộ trong JUC

Thư viện java.util.concurrent (JUC) cung cấp một tập hợp các lớp chuyên dụng để xử lý các vấn đề đồng bộ hóa phức tạp. Trong đó, CountDownLatch, CyclicBarrier và Semaphore là ba nguyên mẫu phổ biến nhất, mỗi loại giải quyết một kịch bản điều phối luồng khác nhau mà việc chỉ dùng synchronized hay wait/notify thuần túy sẽ trở nên cồng kềnh.

1. CountDownLatch: Bộ đếm ngược đồng bộ

1.1 Kịch bản sử dụng

CountDownLatch hoạt động như một cửa chặn kỹ thuật số. Nó cho phép một hoặc nhiều luồng mẹ bị đình chỉ cho đến khi một tập hợp các luồng con hoàn tất công việc. Minh họa bằng ví dụ ghép trận đấu: hệ thống chỉ khởi chạy khi đủ 5 người chơi nhấn sẵn sàng. Nếu chưa đủ, luồng chính sẽ bị treo trạng thái chờ.

Dưới đây là cách triển khai mô phỏng:

public static void minhHoaBoDem() throws Exception {
    // Khởi tạo với giá trị khởi tạo là 5
    CountDownLatch dongBoDem = new CountDownLatch(5);
    
    new Thread(() -> {
        for (int i = 0; i < 12; i++) {
            try { Thread.sleep(1000L); } catch (InterruptedException e) { e.printStackTrace(); }
            new Thread(() -> {
                long soLuongTruoc = dongBoDem.getCount();
                dongBoDem.countDown(); // Giảm bộ đếm khi người chơi sẵn sàng
                if (soLuongTruoc > 0) {
                    System.out.println("Luồng " + Thread.currentThread().getName() + " đã vào phòng, còn thiếu " + dongBoDem.getCount() + " người.");
                } else {
                    System.out.println("Luồng " + Thread.currentThread().getName() + " đến trễ, phòng đã đóng.");
                }
            }).start();
        }
    }).start();
    
    new Thread(() -> {
        try {
            System.out.println("Hệ thống đang chờ đủ điều kiện...");
            dongBoDem.await(); // Chặn luồng cho đến khi đếm về 0
            System.out.println("Phòng đã khóa, bắt đầu ghép trận!");
        } catch (InterruptedException e) { e.printStackTrace(); }
    }).start();
    
    System.out.println("Đang mở đăng ký...");
    dongBoDem.await();
    System.out.println("Đang tiến hành matching...");
}

Khi hai luồng độc lập gọi await(), cả hai đều sẽ bị chặn. Chỉ khi đúng 5 luồng gọi countDown(), trạng thái đồng bộ mới được đáp ứng. Lưu ý rằng CountDownLatch được thiết kế cho một chu kỳ duy nhất và không hỗ trợ reset.

1.2 Phân tích triển khai

Cốt lõi của lớp này nằm ở lớp nội bộ Sync kế thừa từ AbstractQueuedSynchronizer (AQS). Khi khởi tạo, tham số đếm được gán trực tiếp vào trường state của AQS. Mỗi lệnh countDown() thực chất là một lệnh giải phóng trạng thái chia sẻ (releaseShared), sử dụng phép CAS để giảm state đi 1 an toàn. Phương thức await() gọi acquireSharedInterruptibly(1), sẽ chỉ trả về thành công khi tryAcquireShared() trả về giá trị dương (điều kiện state == 0). Nếu chưa đạt, luồng sẽ được đẩy vào hàng đợi chờ và vòng lại liên tục cho đến khi điều kiện thỏa mãn.

2. CyclicBarrier: Rào chắn tuần hoàn

2.1 Kịch bản sử dụng

Khác với bộ đếm ngược chỉ chặn luồng chủ, CyclicBarrier tạo ra một điểm hội tụ chung. Tất cả các luồng tham gia đều phải gọi await() để tự chặn mình tại rào chắn. Chỉ khi luồng cuối cùng cũng chạm đến điểm này, rào chắn mới bung ra, giải phóng toàn bộ luồng đang ngủ. Đặc tính nổi bật nhất là nó tự động reset và sẵn sàng cho chu kỳ tiếp theo.

public static void minhHoaRanChong() throws Exception {
    CyclicBarrier ranTuanHoan = new CyclicBarrier(5);
    for (int i = 0; i < 12; i++) {
        Thread.sleep(1000L);
        new Thread(() -> {
            try {
                int daCho = ranTuanHoan.getNumberWaiting() + 1;
                System.out.println("Luồng " + Thread.currentThread().getName() + " chuẩn bị, tổng " + daCho + " người.");
                ranTuanHoan.await(); // Tự chặn tại rào chắn
                System.out.println("Luồng " + Thread.currentThread().getName() + " bắt đầu xử lý.");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

Kết quả thực thi cho thấy các luồng sẽ bị giữ lại cho đến khi đủ 5 đơn vị. Ngay sau khi nhóm đầu tiên được giải phóng, biến đếm tự động quay về trạng thái ban đầu để đón nhóm thứ hai.

2.2 Phân tích triển khai

CyclicBarrier không sử dụng AQS mà được xây dựng dựa trên ReentrantLock và Condition. Khi await() được gọi, luồng sẽ cố gắng nắm giữ khóa, sau đó giảm biến count. Nếu count còn lớn hơn 0, luồng sẽ gọi trip.await() để vào trạng thái blocking. Khi count đạt 0, luồng cuối cùng sẽ thực thi barrierAction (nếu được cấu hình), sau đó gọi trip.signalAll() để đánh thức toàn bộ luồng chờ. Đồng thời, nó tạo một đối tượng Generation mới để phân biệt các chu kỳ đồng bộ khác nhau.

2.3 Mô phỏng bằng cơ chế cơ bản

Cặp đôi ReentrantLock + Condition về mặt ngữ nghĩa tương đương với synchronized + wait/notify. Chúng ta có thể tái tạo lại hành vi này bằng cách quản lý thủ công:

public class VanDaiDongBo {
    private final int soLuongYeuCau;
    private int conLai;

    public VanDaiDongBo(int soLuongYeuCau) {
        this.soLuongYeuCau = soLuongYeuCau;
        this.conLai = soLuongYeuCau;
    }

    public int getSoLuongDaCho() {
        return soLuongYeuCau - conLai;
    }

    public int choDenKhiDuy() throws InterruptedException {
        synchronized (this) {
            int soCon = --conLai;
            if (soCon == 0) {
                this.conLai = soLuongYeuCau; // Reset bộ đếm
                this.notifyAll(); // Đánh thức tất cả luồng chờ
                return 0;
            }
            this.wait(); // Luồng hiện tại nhường CPU và chờ tín hiệu
        }
        return conLai;
    }
}

Đoạn mã trên chứng minh rằng bản chất của rào chắn tuần hoàn chỉ là việc đếm ngược kết hợp với cơ chế chờ/thông báo tập thể, không đòi hỏi kiến trúc quá phức tạp.

3. Semaphore: Bộ kiểm soát tài nguyên

3.1 Kịch bản sử dụng

Semaphore đóng vai trò như một máy cấp phát vé thông hành. Số vé ban đầu quy định số lượng luồng tối đa được phép thực thi song song. Khi vé cạn, các luồng mới sẽ bị chặn trong hàng đợi cho đến khi có luồng trả vé (release()). Đây là giải pháp chuẩn cho bài toán giới hạn kết nối database, điều phối thread pool, hoặc bảo vệ tài nguyên vật lý.

public static void minhHoaKiemSoatLuong() throws InterruptedException {
    int slotCapNhat = 5;
    Semaphore heThongHao = new Semaphore(slotCapNhat);
    System.out.println("Khởi tạo " + slotCapNhat + " slot xử lý.");
    
    for (int i = 0; i < 10; i++) {
        Thread.sleep(1000L);
        new Thread(() -> {
            try {
                System.out.println("Khách " + Thread.currentThread().getName() + " đang xin slot.");
                heThongHao.acquire(); // Chờ nhận vé
                System.out.println("Khách " + Thread.currentThread().getName() + " đang thực hiện nghiệp vụ.");
                Thread.sleep(5000L); // Giả lập thời gian xử lý
                System.out.println("Khách " + Thread.currentThread().getName() + " hoàn thành, trả slot.");
                heThongHao.release(); // Giải phóng vé
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

5 luồng đầu tiên sẽ chiếm trọn tài nguyên và chạy song song. Các luồng tiếp theo phải đợi trong hàng đợi FIFO (nếu dùng chế độ fair). Khi một luồng hoàn thành và gọi release(), một luồng đang chờ sẽ nhận được vé và kích hoạt. Semaphore hỗ trợ cả chế độ công bằng (Fair) và không công bằng (Nonfair) tùy theo constructor.

3.2 Phân tích triển khai

Tương tự CountDownLatch, Semaphore cũng kế thừa cơ chế AQS. Trường state đại diện cho số vé còn lại.

• Chế độ công bằng (FairSync): Trước khi thử chiếm vé, luồng sẽ kiểm tra hasQueuedPredecessors(). Nếu có luồng khác đứng trước trong hàng đợi, nó sẽ từ chối CAS và tự động đi vào hàng đợi chờ.
• Chế độ không công bằng (NonfairSync): Bỏ qua bước kiểm tra hàng đợi, thử CAS giảm state ngay lập tức. Nếu thành công hoặc state vẫn đủ, luồng được phép vào. Nếu không, mới đi chờ.

Phương thức release() thực hiện việc cộng ngược lại vào state thông qua CAS, sau đó gọi doReleaseShared() của AQS để quét và đánh thức các luồng đang block trong hàng đợi đầu vào.

4. Phân biệt đặc điểm kỹ thuật

• Đối tượng bị chặn: CountDownLatch chỉ chặn các luồng gọi await(); CyclicBarrier chặn tất cả các luồng tham gia gọi await().
• Chiến lược đếm: CountDownLatch đếm ngược và dừng ở 0; CyclicBarrier đếm ngược nhưng tự động reset về giá trị ban đầu sau mỗi lần đạt ngưỡng.
• Tính tái sử dụng: CountDownLatch dùng một lần; CyclicBarrier và Semaphore có thể tái sử dụng xuyên suốt vòng đời ứng dụng.
• Cơ chế nền tảng: CountDownLatch và Semaphore xây dựng trên AQS; CyclicBarrier xây dựng trên ReentrantLock và Condition.
• Phân bổ nhiệm vụ: CountDownLatch cho phép một luồng gọi countDown() nhiều lần; CyclicBarrier yêu cầu mỗi luồng phải gọi await() đúng một lần cho mỗi chu kỳ đồng bộ.