java.util.concurrent Package đơn giản và dễ hiểu (Phần 2)

Tiếp nối phần 1 chúng ta hãy cùng đến với những lớp còn lại của gói java.util.concurrent

6. CyclicBarrier

CyclicBarrier là một lớp trong gói java.util.concurrent của Java, được sử dụng để đồng bộ hóa một nhóm các luồng (threads) lại với nhau, sao cho tất cả các luồng trong nhóm đều phải chờ đợi cho đến khi tất cả đều đạt đến một điểm đồng bộ chung (gọi là barrier hay tiếng việt là rào chắn) trước khi tiếp tục thực hiện. Sau khi tất cả các luồng đạt đến rào chắn, rào chắn sẽ được tái sử dụng (cyclic), do đó lớp này hữu ích trong các tình huống cần đồng bộ hóa định kỳ.

Đặc điểm của CyclicBarrier

• Đồng bộ hóa các luồng: Các luồng phải chờ nhau tại rào chắn trước khi tất cả tiếp tục thực hiện.

• Tái sử dụng rào chắn: Sau khi tất cả các luồng đã đạt đến rào chắn, nó có thể được tái sử dụng cho chu kỳ tiếp theo.

• Hỗ trợ hành động khi rào chắn được kích hoạt: Có thể chỉ định một hành động sẽ được thực hiện bởi một luồng khi tất cả các luồng đạt đến rào chắn.

• Phù hợp với các thuật toán đồng bộ phức tạp: Thường được sử dụng trong các bài toán như xử lý song song hoặc các ứng dụng cần phối hợp giữa nhiều luồng.

Điểm mạnh của CyclicBarrier là nó có thể tái sử dụng sau khi các luồng đã vượt qua barrier, giúp cho việc đồng bộ hóa trong các vòng lặp hoặc các bước xử lý lặp lại trở nên dễ dàng hơn.

Các phương thức chính của CyclicBarrier

• CyclicBarrier(int parties): Tạo một rào chắn cho số lượng luồng cụ thể.
• CyclicBarrier(int parties, Runnable barrierAction): Tạo một rào chắn và chỉ định một hành động thực thi khi tất cả các luồng đạt đến rào chắn.
• await(): Luồng gọi phương thức này sẽ chờ đến khi tất cả các luồng khác gọi await().
• getParties(): Trả về số lượng luồng ban đầu được thiết lập cho rào chắn.
• getNumberWaiting(): Trả về số lượng luồng hiện đang chờ tại rào chắn.
• reset(): Đặt lại trạng thái rào chắn, hủy bỏ tất cả các luồng đang chờ.

Ví dụ về CyclicBarrier

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier; public class CyclicBarrierExample { public static void main(String[] args) { final int NUM_THREADS = 3; // Tạo CyclicBarrier với số lượng luồng cần đồng bộ hóa và hành động thực hiện khi tất cả luồng đều chạm đến barrier CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS, () -> { System.out.println("Tất cả các luồng đã đến barrier và tiếp tục thực hiện."); }); // Tạo và khởi động các luồng for (int i = 0; i < NUM_THREADS; i++) { new Thread(new Task(barrier)).start(); } }
} class Task implements Runnable { private CyclicBarrier barrier; public Task(CyclicBarrier barrier) { this.barrier = barrier; } @Override public void run() { try { System.out.println(Thread.currentThread().getName() + " đang thực hiện công việc..."); Thread.sleep(1000); // Giả lập thời gian thực hiện công việc System.out.println(Thread.currentThread().getName() + " đã đến barrier"); // Đợi các luồng khác đến barrier barrier.await(); // Tiếp tục thực hiện sau khi tất cả các luồng đã đến barrier System.out.println(Thread.currentThread().getName() + " tiếp tục thực hiện công việc sau barrier"); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } }
}

Giải thích ví dụ

1. Tạo CyclicBarrier:

CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS, () -> { System.out.println("Tất cả các luồng đã đến barrier và tiếp tục thực hiện.");
});

Ở đây, chúng ta tạo một CyclicBarrier với số lượng luồng cần đồng bộ hóa là 3 (NUM_THREADS). Khi tất cả 3 luồng đều đạt đến barrier, hành động được chỉ định (in ra thông báo) sẽ được thực hiện.

2. Tạo và khởi động các luồng:

for (int i = 0; i < NUM_THREADS; i++) { new Thread(new Task(barrier)).start();
}

Chúng ta tạo và khởi động 3 luồng, mỗi luồng sẽ thực hiện công việc trong lớp Task.

3. Lớp Task

class Task implements Runnable { private CyclicBarrier barrier; public Task(CyclicBarrier barrier) { this.barrier = barrier; } @Override public void run() { try { System.out.println(Thread.currentThread().getName() + " đang thực hiện công việc..."); Thread.sleep(1000); // Giả lập thời gian thực hiện công việc System.out.println(Thread.currentThread().getName() + " đã đến barrier"); // Đợi các luồng khác đến barrier barrier.await(); // Tiếp tục thực hiện sau khi tất cả các luồng đã đến barrier System.out.println(Thread.currentThread().getName() + " tiếp tục thực hiện công việc sau barrier"); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } }
}

Trong lớp Task, mỗi luồng sẽ giả lập công việc bằng cách ngủ trong 1 giây, sau đó in ra thông báo cho biết nó đã đến barrier. Sau đó, nó sẽ gọi barrier.await() để đợi các luồng khác. Khi tất cả các luồng đều gọi await(), hành động được chỉ định ở bước 1 sẽ được thực hiện, và tất cả các luồng sẽ tiếp tục công việc của mình.

4. Kết quả

Khi chạy chương trình, bạn sẽ thấy rằng tất cả các luồng đều thực hiện công việc ban đầu của mình và đợi tại barrier. Khi tất cả đều đến barrier, thông báo "Tất cả các luồng đã đến barrier và tiếp tục thực hiện." sẽ được in ra, sau đó tất cả các luồng sẽ tiếp tục thực hiện phần công việc sau barrier.

Ví dụ này minh họa cách CyclicBarrier có thể được sử dụng để đồng bộ hóa các luồng trong Java, giúp đảm bảo rằng tất cả các luồng đều hoàn thành một phần công việc trước khi tiếp tục sang phần tiếp theo.

Hạn chế của CyclicBarrier

• Deadlock tiềm ẩn: Nếu một luồng không đạt đến rào chắn, các luồng khác sẽ chờ vô thời hạn, gây ra deadlock.

• Hủy bỏ phức tạp: Nếu rào chắn bị phá vỡ (BrokenBarrierException), cần xử lý lỗi để đảm bảo chương trình hoạt động bình thường.

7. Semaphore

Semaphore là một lớp trong gói java.util.concurrent của Java, được sử dụng để kiểm soát số lượng luồng truy cập vào một tài nguyên nhất định cùng một lúc. Nó hoạt động như một bộ đếm cho phép một số lượng nhất định các luồng truy cập vào các phần tài nguyên giới hạn. Semaphore có thể được sử dụng để giới hạn số lượng luồng đồng thời truy cập vào một tài nguyên nhất định hoặc để thực hiện các tác vụ đồng bộ hóa khác.

Đặc điểm của Semaphore

• Kiểm soát truy cập đồng thời: Hạn chế số lượng luồng có thể truy cập đồng thời vào một tài nguyên hoặc vùng code cụ thể.

• Giới hạn giấy phép: Giấy phép có thể được cấu hình khi khởi tạo Semaphore để giới hạn số luồng truy cập.

• Hai chế độ hoạt động:

  • Fair (Công bằng): Phân phối giấy phép theo thứ tự yêu cầu.
  • Non-fair (Không công bằng): Không đảm bảo thứ tự, thường nhanh hơn.

• Hỗ trợ đồng bộ hóa nâng cao: Semaphore có thể được sử dụng để xây dựng các cơ chế đồng bộ phức tạp như khóa nhị phân, hàng đợi hoặc hàng đợi giới hạn.

Các phương thức chính của Semaphore

• acquire(): Lấy một giấy phép, đợi nếu không có giấy phép khả dụng.
• acquire(int permits): Lấy một số lượng giấy phép, đợi nếu không đủ giấy phép khả dụng.
• tryAcquire(): Cố gắng lấy một giấy phép, trả về true nếu thành công, false nếu thất bại.
• tryAcquire(int permits, long timeout, TimeUnit unit): Cố gắng lấy một số giấy phép trong một khoảng thời gian, có thể bị ngắt.
• release(): Trả lại một giấy phép.
• release(int permits): Trả lại một số lượng giấy phép.
• availablePermits(): Trả về số lượng giấy phép hiện có.
• getQueueLength(): Trả về số lượng luồng đang chờ giấy phép.

Ví dụ về Semaphore

import java.util.concurrent.Semaphore; public class SemaphoreExample { public static void main(String[] args) { // Tạo một Semaphore với 2 permit (giấy phép truy cập). Semaphore semaphore = new Semaphore(2); // Tạo và khởi động 5 luồng. for (int i = 0; i < 5; i++) { new Thread(new Task(semaphore, "Task " + (i + 1))).start(); } }
} class Task implements Runnable { private Semaphore semaphore; private String name; public Task(Semaphore semaphore, String name) { this.semaphore = semaphore; this.name = name; } @Override public void run() { try { System.out.println(name + " đang chờ giấy phép truy cập..."); semaphore.acquire(); // Lấy một giấy phép truy cập System.out.println(name + " đã có giấy phép truy cập!"); // Giả lập thời gian truy cập tài nguyên Thread.sleep(2000); System.out.println(name + " trả lại giấy phép truy cập."); semaphore.release(); // Trả lại giấy phép truy cập } catch (InterruptedException e) { e.printStackTrace(); } }
}

1. Tạo Semaphore:

Semaphore semaphore = new Semaphore(2);

Ở đây, chúng ta tạo một Semaphore với 2 giấy phép truy cập (permit). Điều này có nghĩa là tối đa 2 luồng có thể truy cập vào tài nguyên cùng một lúc.

2. Tạo và khởi động các luồng:

for (int i = 0; i < 5; i++) { new Thread(new Task(semaphore, "Task " + (i + 1))).start();
}

Chúng ta tạo và khởi động 5 luồng, mỗi luồng sẽ thực hiện công việc trong lớp Task.

3. Lớp Task:

class Task implements Runnable { private Semaphore semaphore; private String name; public Task(Semaphore semaphore, String name) { this.semaphore = semaphore; this.name = name; } @Override public void run() { try { System.out.println(name + " đang chờ giấy phép truy cập..."); semaphore.acquire(); // Lấy một giấy phép truy cập System.out.println(name + " đã có giấy phép truy cập!"); // Giả lập thời gian truy cập tài nguyên Thread.sleep(2000); System.out.println(name + " trả lại giấy phép truy cập."); semaphore.release(); // Trả lại giấy phép truy cập } catch (InterruptedException e) { e.printStackTrace(); } }
}

Trong lớp Task, mỗi luồng sẽ cố gắng lấy một giấy phép truy cập từ Semaphore bằng cách gọi semaphore.acquire(). Nếu có giấy phép sẵn có, luồng sẽ tiếp tục thực hiện công việc của mình. Nếu không có giấy phép nào sẵn có, luồng sẽ chờ cho đến khi có một giấy phép được trả lại (bởi một luồng khác gọi semaphore.release()). Sau khi hoàn thành công việc, luồng sẽ trả lại giấy phép truy cập bằng cách gọi semaphore.release().

4. Kết quả

Khi chạy chương trình, bạn sẽ thấy rằng chỉ có tối đa 2 luồng có thể truy cập vào tài nguyên cùng một lúc. Các luồng khác sẽ phải chờ cho đến khi có giấy phép truy cập. Điều này minh họa cách Semaphore có thể được sử dụng để giới hạn số lượng luồng đồng thời truy cập vào một tài nguyên chung, giúp đảm bảo việc sử dụng tài nguyên một cách an toàn và hiệu quả trong môi trường đa luồng.

8. Lock

Lock là một giao diện trong gói java.util.concurrent.locks của Java, cung cấp các cơ chế kiểm soát truy cập đồng thời vào tài nguyên dùng chung, tương tự như synchronized, nhưng linh hoạt và mạnh mẽ hơn. Lock cho phép các luồng quản lý khóa (lock) theo cách thủ công, hỗ trợ các tính năng như cố gắng chiếm quyền (try-lock), chờ khóa trong một khoảng thời gian cụ thể, và ngắt quá trình chờ.

Đặc điểm của Lock

• Kiểm soát thủ công: Không giống synchronized, Lock yêu cầu mã rõ ràng để khóa (lock()) và mở khóa (unlock()). Điều này giúp kiểm soát chi tiết quá trình đồng bộ hóa.

• Linh hoạt hơn synchronized: Cung cấp khả năng kiểm soát cao hơn, như hỗ trợ timeout, khả năng thử chiếm quyền, hoặc phản ứng khi bị ngắt (interruptible).

• Nhiều loại Lock: Java cung cấp các loại khóa như ReentrantLock, ReadWriteLock và các biến thể khác để phù hợp với các yêu cầu cụ thể.

Các phương thức chính của Lock

• void lock(): Khóa ngay lập tức, đợi đến khi có khóa nếu nó đã bị khóa bởi luồng khác.
• void lockInterruptibly() throws InterruptedException: Giống lock() nhưng cho phép bị ngắt khi đang chờ.
• boolean tryLock(): Cố gắng lấy khóa, trả về true nếu thành công, false nếu thất bại.
• boolean tryLock(long time, TimeUnit unit) throws InterruptedException: Cố gắng lấy khóa trong một khoảng thời gian, có thể bị ngắt khi đang chờ.
• void unlock(): Mở khóa, phải được gọi sau khi khóa được chiếm giữ thành công.

Ví dụ về Lock

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; public class LockExample { private int counter = 0; private final Lock lock = new ReentrantLock(); public void increment() { lock.lock(); // Khóa trước khi truy cập tài nguyên dùng chung try { counter++; System.out.println(Thread.currentThread().getName() + " tăng counter: " + counter); } finally { lock.unlock(); // Đảm bảo mở khóa dù có xảy ra lỗi } } public static void main(String[] args) { LockExample example = new LockExample(); Runnable task = example::increment; Thread thread1 = new Thread(task, "Thread-1"); Thread thread2 = new Thread(task, "Thread-2"); thread1.start(); thread2.start(); }
}

Giải thích ví dụ

1. Khởi tạo Lock:

private final Lock lock = new ReentrantLock();

Sử dụng ReentrantLock để tạo một đối tượng Lock.

2. Sử dụng Lock để bảo vệ tài nguyên dùng chung:

• Trước khi truy cập counter, gọi lock.lock() để đảm bảo chỉ một luồng có thể truy cập tại một thời điểm.
• Sử dụng try-finally để đảm bảo khóa được mở ngay cả khi có lỗi.

3. Mở khóa:

Gọi lock.unlock() trong khối finally để đảm bảo khóa luôn được giải phóng, tránh deadlock.

Các loại Lock phổ biến

1. ReentrantLock:

• Một khóa tái nhập, cho phép luồng giữ khóa có thể khóa lại mà không bị deadlock.
• Cung cấp các phương thức bổ sung như tryLock() và lockInterruptibly().

2. ReadWriteLock:

• Cho phép nhiều luồng đọc đồng thời (khi không có ghi), nhưng chỉ một luồng có thể ghi.

• Được chia thành hai khóa con:

  • ReadLock
  • WriteLock

• StampedLock

  • Giới thiệu từ Java 8, hỗ trợ các cơ chế kiểm soát đồng thời tối ưu như optimistic locking (khóa lạc quan).

Kết quả khi chạy ví dụ

Khi chạy chương trình, bạn sẽ thấy các luồng thực thi theo thứ tự đảm bảo an toàn đồng thời:

Thread-1 tăng counter: 1
Thread-2 tăng counter: 2

Lợi ích của việc sử dụng Lock

1. Đồng bộ hóa linh hoạt hơn synchronized: Có thể thử chiếm quyền, hỗ trợ timeout, và phản ứng với ngắt luồng.

2. Hiệu suất tốt hơn: Trong nhiều trường hợp, Lock cung cấp hiệu suất cao hơn khi xử lý đồng thời phức tạp.

3. Hỗ trợ nhiều tính năng: Các loại Lock như ReadWriteLock hoặc StampedLock phù hợp với các kịch bản truy cập song song khác nhau.

Hạn chế của Lock

1. Phức tạp hơn synchronized: Yêu cầu mã rõ ràng để quản lý khóa, dễ dẫn đến lỗi như quên mở khóa (unlock).

2. Deadlock: Sử dụng không cẩn thận có thể gây ra deadlock, đặc biệt trong các hệ thống phức tạp.

Khi nào nên sử dụng Lock?

• Cần kiểm soát nâng cao, như timeout hoặc khả năng thử chiếm quyền.
• Tối ưu hóa truy cập song song, ví dụ: cho phép nhiều luồng đọc đồng thời với ReadWriteLock.
• Khi các thao tác đồng bộ hóa phức tạp và synchronized không đáp ứng tốt.

9. ReentrantLock

ReentrantLock là một lớp trong gói java.util.concurrent.locks của Java, cung cấp cơ chế khóa (lock) tương tự như từ khóa synchronized nhưng linh hoạt và mạnh mẽ hơn.

• "Reentrant" có nghĩa là một luồng đã giữ khóa có thể tái nhập vào cùng một khóa mà không bị chặn.
• Nó cung cấp các tính năng nâng cao như khả năng cố gắng giành khóa, thời gian chờ và khả năng gián đoạn.

Đặc điểm của ReentrantLock

1. Khóa tái nhập được (Reentrant): Một luồng có thể giành quyền sở hữu cùng một khóa nhiều lần mà không bị chặn.

2. Linh hoạt hơn synchronized: Có thể kiểm tra xem khóa đã được giữ hay chưa. Hỗ trợ tính năng cố gắng giành khóa hoặc hủy bỏ khi bị gián đoạn.

3. Hỗ trợ công bằng (Fairness):Cung cấp tùy chọn ưu tiên cho các luồng đang chờ lâu hơn (công bằng) hoặc theo thứ tự không xác định (mặc định).

4. Cần khóa và mở khóa thủ công: Khóa bằng phương thức lock() và mở khóa bằng unlock().

Các phương thức chính của ReentrantLock

• void lock(): Khóa, chờ đến khi khóa có sẵn nếu cần.
• void lockInterruptibly(): Khóa nhưng có thể bị gián đoạn.
• boolean tryLock(): Cố gắng giành khóa, trả về true nếu thành công, false nếu không.
• boolean tryLock(long time, TimeUnit unit): Cố gắng giành khóa trong khoảng thời gian chờ.
• void unlock(): Mở khóa.
• boolean isLocked(): Kiểm tra xem khóa hiện đang được giữ hay không.
• boolean isHeldByCurrentThread(): Kiểm tra xem khóa có đang được giữ bởi luồng hiện tại hay không.

Ví dụ về ReentrantLock

import java.util.concurrent.locks.ReentrantLock; public class ReentrantLockExample { private static final ReentrantLock lock = new ReentrantLock(); private static int sharedCounter = 0; public static void main(String[] args) { Thread thread1 = new Thread(new Task(), "Thread-1"); Thread thread2 = new Thread(new Task(), "Thread-2"); thread1.start(); thread2.start(); } static class Task implements Runnable { @Override public void run() { for (int i = 0; i < 5; i++) { lock.lock(); // Giành quyền truy cập vào đoạn mã try { System.out.println(Thread.currentThread().getName() + " đang tăng biến đếm."); sharedCounter++; System.out.println("Giá trị hiện tại của sharedCounter: " + sharedCounter); } finally { lock.unlock(); // Mở khóa để các luồng khác có thể truy cập } // Giả lập công việc khác ngoài đoạn mã đồng bộ try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } } }
}

Giải thích ví dụ

1. Khởi tạo ReentrantLock:

private static final ReentrantLock lock = new ReentrantLock();

• Ở đây, một đối tượng ReentrantLock được tạo ra và sử dụng chung cho tất cả các luồng.
• Nó sẽ đảm bảo rằng chỉ có một luồng được truy cập vào tài nguyên dùng chung sharedCounter tại một thời điểm.

2. 2. Khóa đoạn mã (lock):

lock.lock();
try { // Đoạn mã cần đồng bộ System.out.println(Thread.currentThread().getName() + " đang tăng biến đếm."); sharedCounter++; System.out.println("Giá trị hiện tại của sharedCounter: " + sharedCounter);
} finally { lock.unlock();
} 

• lock(): Khi một luồng gọi phương thức này, nó sẽ cố gắng giành quyền truy cập vào tài nguyên. Nếu tài nguyên đã bị khóa bởi luồng khác, nó sẽ chờ cho đến khi tài nguyên được mở khóa.
• Khối try-finally: Đảm bảo rằng khóa sẽ được giải phóng bằng cách gọi unlock(), ngay cả khi có lỗi xảy ra trong đoạn mã được bảo vệ.

3. Mở khóa (unlock)

finally { lock.unlock();
}

• Mỗi khi một luồng hoàn thành công việc với tài nguyên dùng chung, nó cần mở khóa để cho phép các luồng khác tiếp tục.
• Nếu không mở khóa, các luồng khác sẽ bị chặn mãi mãi, gây ra deadlock.

4. Giả lập công việc ngoài đoạn đồng bộ

try { Thread.sleep(500);
} catch (InterruptedException e) { e.printStackTrace();
}

• Thread.sleep(500): Giả lập công việc khác không liên quan đến tài nguyên dùng chung.
• Điều này cho thấy rằng các luồng chỉ đồng bộ hóa khi cần thao tác với tài nguyên dùng chung, giúp cải thiện hiệu suất.

Kết quả khi chạy ví dụ

Thread-1 đang tăng biến đếm.
Giá trị hiện tại của sharedCounter: 1
Thread-2 đang tăng biến đếm.
Giá trị hiện tại của sharedCounter: 2
Thread-1 đang tăng biến đếm.
Giá trị hiện tại của sharedCounter: 3
Thread-2 đang tăng biến đếm.
Giá trị hiện tại của sharedCounter: 4
...

Lợi ích của ReentrantLock

1. Linh hoạt và mạnh mẽ: Hỗ trợ các tính năng nâng cao như gián đoạn và thời gian chờ khi giành khóa.

2. Công bằng tùy chọn: Có thể cấu hình để đảm bảo thứ tự công bằng giữa các luồng chờ.

3. Thích hợp cho các tình huống phức tạp: Đặc biệt hữu ích khi cần kiểm soát chi tiết việc cấp phát khóa.

Hạn chế của ReentrantLock

1. Khóa và mở khóa thủ công: Lập trình viên phải đảm bảo gọi unlock(), nếu không sẽ dẫn đến deadlock.

2. Không tích hợp chặt với cú pháp của Java: Không thể sử dụng cùng với cấu trúc try-with-resources.

Khi nào nên sử dụng ReentrantLock?

• Khi cần kiểm soát linh hoạt hơn so với synchronized, chẳng hạn như:
  • Hủy bỏ giành khóa nếu cần.
  • Giới hạn thời gian chờ để giành khóa.
• Khi cần kiểm tra trạng thái của khóa (như kiểm tra xem khóa đã được giữ chưa).