Trong lập trình đồng thời (concurrent programming) với Java, việc quản lý và điều phối các luồng (threads) là một thách thức lớn. Java cung cấp một bộ công cụ mạnh mẽ trong gói java.util.concurrent để giúp các nhà phát triển giải quyết các vấn đề này một cách hiệu quả. Bài viết này sẽ đi sâu vào ba công cụ đồng bộ hóa phổ biến: CountDownLatch, CyclicBarrier và Semaphore, cùng với các ví dụ minh họa cụ thể để giúp bạn hiểu rõ cách chúng hoạt động và khi nào nên sử dụng.
CountDownLatch: Đợi Tất Cả Hoàn Thành
CountDownLatch là một cơ chế đồng bộ hóa cho phép một hoặc nhiều luồng chờ đợi cho đến khi một tập hợp các hoạt động được hoàn thành bởi các luồng khác. Nó hoạt động như một "cánh cổng" chỉ mở ra khi một bộ đếm nội bộ đếm ngược về 0. Bộ đếm này thường được khởi tạo với số lượng tác vụ cần hoàn thành.
Các Phương Thức Chính của CountDownLatch
public void await() throws InterruptedException: Khi một luồng gọi phương thức này, nó sẽ bị đình chỉ cho đến khi bộ đếm củaCountDownLatchđạt giá trị 0.public boolean await(long timeout, TimeUnit unit) throws InterruptedException: Tương tự nhưawait(), nhưng luồng sẽ chỉ chờ trong một khoảng thời gian nhất định. Nếu bộ đếm không về 0 trong thời gian đó, luồng sẽ tiếp tục thực thi.public void countDown(): Giảm giá trị bộ đếm xuống 1.
Ví Dụ Minh Họa CountDownLatch
Hãy xem xét một kịch bản đơn giản: có một số tác vụ cần được thực hiện bởi các luồng khác nhau, và một luồng chính chỉ có thể tiếp tục công việc của mình sau khi tất cả các tác vụ con này đã hoàn thành. Trong ví dụ dưới đây, luồng chính sẽ in ra thông báo "Tất cả nhiệm vụ đã hoàn tất! Luồng chính có thể tiếp tục." chỉ khi tất cả 5 tác vụ phụ đã gọi countDown().
Lưu ý: Nếu số lượng luồng thực hiện countDown() ít hơn giá trị khởi tạo của CountDownLatch, luồng gọi await() sẽ chờ vô thời hạn (hoặc cho đến khi hết thời gian nếu dùng await(timeout, unit)).
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class DoiHoanThanhTacVu {
public static void main(String[] args) {
// Khởi tạo CountDownLatch với 5 tác vụ cần hoàn thành
CountDownLatch congKetThuc = new CountDownLatch(5);
// Sử dụng ExecutorService để quản lý luồng tốt hơn
ExecutorService quanLyLuong = Executors.newFixedThreadPool(5);
System.out.println("Bắt đầu các nhiệm vụ con...");
// Khởi tạo và chạy 5 tác vụ
for (int i = 0; i < 5; i++) {
final int idTacVu = i + 1;
quanLyLuong.submit(() -> {
try {
System.out.println("Nhiệm vụ " + idTacVu + " đang thực hiện...");
Thread.sleep((long) (Math.random() * 1000)); // Giả lập thời gian làm việc
System.out.println("Nhiệm vụ " + idTacVu + " đã hoàn thành.");
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // Đặt lại cờ ngắt
System.err.println("Nhiệm vụ " + idTacVu + " bị gián đoạn.");
} finally {
congKetThuc.countDown(); // Giảm bộ đếm sau khi nhiệm vụ hoàn thành
}
});
}
try {
// Luồng chính chờ cho đến khi tất cả 5 tác vụ đã hoàn thành
System.out.println("Luồng chính đang chờ các nhiệm vụ con hoàn tất...");
congKetThuc.await();
System.out.println("Tất cả nhiệm vụ đã hoàn tất! Luồng chính có thể tiếp tục.");
} catch (InterruptedException e) {
System.err.println("Luồng chính bị gián đoạn trong khi chờ.");
Thread.currentThread().interrupt();
} finally {
quanLyLuong.shutdown(); // Đảm bảo thread pool được đóng
try {
if (!quanLyLuong.awaitTermination(60, TimeUnit.SECONDS)) { // Chờ tối đa 60s để các tác vụ kết thúc
System.err.println("Một số tác vụ chưa kết thúc. Buộc đóng pool.");
quanLyLuong.shutdownNow();
}
} catch (InterruptedException e) {
quanLyLuong.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
}
CyclicBarrier: Điểm Hẹn Chung Cho Các Luồng
Không giống như CountDownLatch chỉ sử dụng một lần, CyclicBarrier là một công cụ đồng bộ hóa có thể tái sử dụng, cho phép một nhóm các luồng chờ đợi lẫn nhau tại một "điểm hẹn" chung trước khi tiếp tục thực thi. Khi tất cả các luồng đã đến điểm hẹn (bằng cách gọi await()), rào cản sẽ được "mở" ra, cho phép tất cả chúng tiếp tục. Sau đó, rào cản có thể được "đặt lại" để sử dụng cho lần đồng bộ hóa tiếp theo.
Các Hàm Khởi Tạo của CyclicBarrier
public CyclicBarrier(int parties): Khởi tạo mộtCyclicBarriervới số lượng luồng tham gia cụ thể.public CyclicBarrier(int parties, Runnable barrierAction): Ngoài số lượng luồng tham gia, hàm này còn cho phép chỉ định một hành động (Runnable) sẽ được thực thi bởi luồng cuối cùng đến điểm hẹn, trước khi các luồng khác được giải phóng.
Các Phương Thức Chính của CyclicBarrier
public int await() throws InterruptedException, BrokenBarrierException: Luồng gọi phương thức này sẽ bị chặn cho đến khi tất cả các luồng khác đạt đến điểm hẹn. Phương thức này trả về thứ tự đến của luồng tại điểm hẹn.public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException: Tương tựawait()nhưng có giới hạn thời gian chờ.BrokenBarrierException: Được ném ra nếu rào cản bị hỏng, có thể do một luồng bị gián đoạn hoặc hết thời gian chờ.
Ví Dụ Minh Họa CyclicBarrier
Hãy tưởng tượng một đội gồm 4 người cần hoàn thành 2 giai đoạn của một dự án. Mỗi người phải hoàn thành công việc của mình trong một giai đoạn trước khi cả đội có thể bắt đầu giai đoạn tiếp theo. CyclicBarrier rất phù hợp cho kịch bản này, vì nó cho phép các luồng đồng bộ hóa ở nhiều điểm.
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class DiemHenDoiNhom {
private static final int SO_THANH_VIEN = 4;
public static void main(String[] args) {
// Khởi tạo CyclicBarrier với 4 thành viên và một hành động sau khi tất cả đến điểm hẹn
CyclicBarrier rapCanhChung = new CyclicBarrier(SO_THANH_VIEN, () -> {
System.out.println("\n--- TẤT CẢ THÀNH VIÊN ĐÃ SẴN SÀNG CHO GIAI ĐOẠN TIẾP THEO! ---");
});
ExecutorService quanLyDoiNhom = Executors.newFixedThreadPool(SO_THANH_VIEN);
System.out.println("Dự án bắt đầu!");
for (int i = 0; i < SO_THANH_VIEN; i++) {
final int idThanhVien = i + 1;
quanLyDoiNhom.submit(() -> {
try {
System.out.println("Thành viên " + idThanhVien + " đang làm việc ở GIAI ĐOẠN 1.");
Thread.sleep((long) (Math.random() * 2000 + 500)); // Giả lập thời gian làm việc
System.out.println("Thành viên " + idThanhVien + " đã hoàn thành GIAI ĐOẠN 1 và chờ đợi tại điểm hẹn.");
rapCanhChung.await(); // Chờ các thành viên khác
System.out.println("Thành viên " + idThanhVien + " đang làm việc ở GIAI ĐOẠN 2.");
Thread.sleep((long) (Math.random() * 2000 + 500)); // Giả lập thời gian làm việc
System.out.println("Thành viên " + idThanhVien + " đã hoàn thành GIAI ĐOẠN 2.");
rapCanhChung.await(); // Chờ các thành viên khác cho đến khi tất cả hoàn thành giai đoạn 2
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println("Thành viên " + idThanhVien + " bị gián đoạn.");
} catch (BrokenBarrierException e) {
System.err.println("Thành viên " + idThanhVien + ": Rào cản bị hỏng.");
}
});
}
quanLyDoiNhom.shutdown();
System.out.println("\nĐang chờ tất cả luồng hoàn thành và đóng pool...");
try {
if (!quanLyDoiNhom.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("Một số thành viên chưa hoàn thành. Buộc đóng pool.");
quanLyDoiNhom.shutdownNow();
}
} catch (InterruptedException e) {
quanLyDoiNhom.shutdownNow();
Thread.currentThread().interrupt();
}
System.out.println("Dự án đã kết thúc.");
}
}
Semaphore: Kiểm Soát Quyền Truy Cập Tài Nguyên
Semaphore là một công cụ giúp kiểm soát số lượng luồng có thể truy cập đồng thời vào một tài nguyên hoặc một phần mã giới hạn. Nó duy trì một tập hợp các "giấy phép" (permits). Mỗi luồng muốn truy cập tài nguyên phải acquire() một giấy phép. Khi luồng hoàn thành công việc, nó phải release() giấy phép đó, trả lại cho Semaphore. Nếu không có giấy phép nào khả dụng, luồng sẽ bị chặn cho đến khi có một giấy phép được giải phóng.
Các Phương Thức Thường Dùng của Semaphore
void acquire() throws InterruptedException: Yêu cầu một giấy phép từSemaphore. Luồng sẽ bị chặn nếu không có giấy phép nào khả dụng.void release(): Giải phóng một giấy phép, trả lại choSemaphore.int availablePermits(): Trả về số lượng giấy phép hiện có.boolean hasQueuedThreads(): Kiểm tra xem có luồng nào đang chờ lấy giấy phép hay không.
Ví Dụ Minh Họa Semaphore
Hãy xem xét một kịch bản có một khu vực giới hạn (ví dụ: bãi đỗ xe với số chỗ có hạn, hoặc một máy chủ có số lượng kết nối tối đa). Có nhiều luồng muốn truy cập khu vực này, nhưng chỉ một số lượng giới hạn có thể vào cùng lúc.
import java.util.concurrent.Semaphore;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class QuanLyTruyCapHanChe {
private static final int SO_MAY_TINH_KHA_DUNG = 3; // Ví dụ: 3 máy tính có thể sử dụng cùng lúc
private static final int SO_SINH_VIEN = 7; // 7 sinh viên muốn sử dụng máy tính
public static void main(String[] args) {
// Khởi tạo Semaphore với số lượng giấy phép là SO_MAY_TINH_KHA_DUNG
Semaphore quanLyMayTinh = new Semaphore(SO_MAY_TINH_KHA_DUNG);
ExecutorService phongMay = Executors.newFixedThreadPool(SO_SINH_VIEN);
System.out.println("Phòng máy có " + SO_MAY_TINH_KHA_DUNG + " máy tính sẵn sàng.");
System.out.println("Có " + SO_SINH_VIEN + " sinh viên muốn sử dụng máy.");
for (int i = 0; i < SO_SINH_VIEN; i++) {
final int idSinhVien = i + 1;
phongMay.submit(() -> {
try {
System.out.println("Sinh viên " + idSinhVien + " đang chờ máy tính...");
quanLyMayTinh.acquire(); // Yêu cầu một máy tính
System.out.println("Sinh viên " + idSinhVien + " đã có máy tính. Số máy còn trống: " + quanLyMayTinh.availablePermits());
Thread.sleep((long) (Math.random() * 3000 + 1000)); // Giả lập thời gian sử dụng máy
System.out.println("Sinh viên " + idSinhVien + " đã sử dụng xong máy tính.");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println("Sinh viên " + idSinhVien + " bị gián đoạn khi dùng máy.");
} finally {
quanLyMayTinh.release(); // Trả lại máy tính
System.out.println("Sinh viên " + idSinhVien + " đã trả máy tính. Số máy còn trống: " + quanLyMayTinh.availablePermits());
}
});
}
phongMay.shutdown();
System.out.println("\nĐang chờ tất cả sinh viên hoàn thành và đóng pool...");
try {
if (!phongMay.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("Một số sinh viên chưa trả máy. Buộc đóng pool.");
phongMay.shutdownNow();
}
} catch (InterruptedException e) {
phongMay.shutdownNow();
Thread.currentThread().interrupt();
}
System.out.println("Phòng máy đã đóng cửa.");
}
}