Trong quá trình phát triển ứng dụng Java, mình thường bắt đầu với những đoạn code chạy tuần tự: gọi một method, nhận kết quả, sau đó thực hiện bước tiếp theo. Cách tiếp cận này tương đối dễ hiểu và dễ debug.
Nhưng khi ứng dụng cần xử lý hàng nghìn HTTP requests, truy vấn database, gọi nhiều API bên ngoài hoặc thực hiện các tác vụ nền cùng lúc, việc xử lý tuần tự sẽ trở thành một giới hạn lớn.
Đây là lúc mình cần hiểu sâu hơn về Multithreading và Concurrency.
Không chỉ biết cách tạo Thread hoặc sử dụng ExecutorService, mà còn hiểu vì sao dữ liệu có thể bị sai khi nhiều Thread cùng truy cập, JVM đảm bảo tính nhất quán của bộ nhớ như thế nào, cách thiết kế cơ chế đồng bộ hóa và cách lựa chọn mô hình xử lý đồng thời phù hợp với ứng dụng thực tế.
Phiên bản Java:
Các ví dụ nền tảng sử dụng Java 17 trở lên. Phần Virtual Threads sử dụng Java 21+, phiên bản đánh dấu sự ổn định của tính năng này.
1. Tổng quan về Multithreading và Concurrency
Trước khi tìm hiểu các API như synchronized, ReentrantLock hay CompletableFuture, mình cần xây dựng một mô hình tư duy đúng về cách chương trình được thực thi.
Điểm quan trọng nhất là: Concurrency không đơn giản chỉ là chạy nhiều Thread cùng lúc. Nó còn là bài toán phối hợp công việc, chia sẻ trạng thái, đảm bảo thứ tự quan sát dữ liệu và kiểm soát tài nguyên.
1.1. Process và Thread khác nhau như thế nào?
Process (tiến trình) là một chương trình đang được hệ điều hành thực thi. Một Process có không gian địa chỉ bộ nhớ và các tài nguyên hệ thống của riêng nó.
Thread (luồng) là một đơn vị thực thi bên trong Process. Một Process có thể chứa một hoặc nhiều Thread cùng tham gia thực hiện công việc.
Ví dụ, khi chạy một ứng dụng Java, hệ điều hành tạo ra một Process cho ứng dụng đó. Bên trong Process, JVM có thể sử dụng nhiều Thread để xử lý các công việc khác nhau.
Java Application Process
│
├── Shared Heap
│ ├── User Objects
│ ├── Order Objects
│ └── Shared Cache
│
├── Thread 1
│ ├── Stack
│ ├── Local Variables
│ └── Execution State
│
├── Thread 2
│ ├── Stack
│ ├── Local Variables
│ └── Execution State
│
└── Thread 3
├── Stack
├── Local Variables
└── Execution State
Các Thread trong cùng một Process thường chia sẻ Heap và những tài nguyên cấp Process. Tuy nhiên, mỗi Thread có Stack và trạng thái thực thi riêng.
Ví dụ:
public class Main {
public static void main(String[] args) {
User user = new User();
Thread thread1 = new Thread(() -> {
user.setName("Alice");
});
Thread thread2 = new Thread(() -> {
System.out.println(user.getName());
});
thread1.start();
thread2.start();
}
}Hai Thread cùng giữ quyền truy cập đến một đối tượng User. Vì vậy, nếu một Thread sửa đổi dữ liệu trong khi Thread khác đọc dữ liệu đó, mình cần quan tâm đến tính an toàn của thao tác đọc và ghi.
Ngoài ra, cần phân biệt việc chia sẻ reference với việc chia sẻ toàn bộ trạng thái của một biến. Local variable thuộc Stack của từng Thread, nhưng nếu local variable đó chứa reference đến cùng một object trên Heap thì object vẫn có thể được nhiều Thread truy cập.
Local variable không tự động làm cho object mà nó tham chiếu trở thành Thread-Safe. Điều cần quan tâm là object có bị chia sẻ giữa các Thread hay không và những quy tắc đồng bộ hóa nào bảo vệ trạng thái của object đó.
1.2. Concurrency và Parallelism
Hai khái niệm này thường được sử dụng cùng nhau nhưng không hoàn toàn giống nhau.
Concurrency (đồng thời) là khả năng tổ chức nhiều công việc có thể tiến triển xen kẽ nhau trong cùng một khoảng thời gian.
Parallelism (song song) là việc thực thi thực sự nhiều công việc tại cùng một thời điểm, thường nhờ nhiều CPU core hoặc nhiều đơn vị xử lý.
Một CPU Core │ ├── Task A: A1 → A2 ├── Task B: B1 → B2 └── Task C: C1 → C2 CPU luân phiên thực thi: A1 → B1 → C1 → A2 → B2 → C2
CPU Core 1: Task A ────────────── CPU Core 2: Task B ────────────── CPU Core 3: Task C ────────────── Các task có thể thực thi thực sự cùng một thời điểm.
Concurrency không bắt buộc phải có nhiều CPU core. Ngược lại, Parallelism đòi hỏi các công việc phải được thực thi đồng thời về mặt vật lý.
Trong Java, mình có thể sử dụng nhiều Thread để tổ chức công việc đồng thời. Việc chúng có thực thi song song hay không phụ thuộc vào số lượng tài nguyên xử lý, cách lập lịch của hệ điều hành hoặc JVM và đặc điểm của công việc.
1.2.1. CPU-bound và I/O-bound
Đây là cách phân loại rất hữu ích khi thiết kế hệ thống.
CPU-bound: Công việc tiêu tốn phần lớn thời gian để thực hiện tính toán, chẳng hạn mã hóa dữ liệu, xử lý ảnh hoặc tính toán số học lớn.
I/O-bound: Công việc dành phần lớn thời gian chờ database, HTTP API, đọc file hoặc các hoạt động I/O khác.
Đối với CPU-bound, tăng số lượng Thread vượt quá số CPU core có thể làm tăng chi phí chuyển đổi ngữ cảnh mà không cải thiện đáng kể throughput.
Đối với I/O-bound, nhiều công việc có thể chờ I/O cùng lúc. Một mô hình thực thi phù hợp có thể giúp hệ thống sử dụng tài nguyên hiệu quả hơn trong khoảng thời gian chờ này.
Đừng lựa chọn số lượng Thread chỉ dựa trên một con số cố định. Hãy dựa vào đặc điểm công việc, độ trễ, số CPU core, giới hạn kết nối database, giới hạn API bên ngoài và mức tài nguyên mà hệ thống có thể sử dụng.
1.3. Vòng đời của Thread trong Java
Java định nghĩa sáu trạng thái của Thread thông qua Thread.State:
new Thread()
│
▼
NEW
│
start()
│
▼
RUNNABLE
/ | \
/ | \
Lock busy wait() sleep(timeout)
│ │ │
▼ ▼ ▼
BLOCKED WAITING TIMED_WAITING
│ │ │
└────────┴──────────┘
│
Điều kiện được đáp ứng
│
▼
RUNNABLE
│
run() kết thúc
│
▼
TERMINATED
Cần lưu ý rằng sơ đồ trên chỉ mô tả một số chuyển đổi tiêu biểu. Trong thực tế, trạng thái thay đổi tùy thuộc vào thao tác mà Thread thực hiện.
1.3.1. NEW
Thread đã được tạo nhưng chưa được khởi động.
Thread thread = new Thread(() -> {
System.out.println("Hello Thread");
});
System.out.println(thread.getState());
// NEW1.3.2. RUNNABLE
Thread đã được khởi động và đang ở trạng thái có thể thực thi theo mô hình của JVM. Thread có thể đang thực sự chạy trên CPU hoặc đang chờ được hệ điều hành lập lịch.
Trong Java, RUNNABLE bao gồm cả trạng thái sẵn sàng chạy và đang chạy. Vì vậy, không có trạng thái RUNNING riêng trong Thread.State.
Thread thread = new Thread(() -> {
while (!Thread.currentThread().isInterrupted()) {
// Thực hiện công việc
}
});
thread.start();
System.out.println(thread.getState());Trong ví dụ trên, kết quả phụ thuộc vào thời điểm kiểm tra. Thread có thể đang ở RUNNABLE hoặc đã chuyển sang trạng thái khác nếu chương trình thay đổi cách thực thi.
1.3.3. BLOCKED
Thread đang chờ lấy một intrinsic monitor lock để đi vào hoặc tiếp tục thực thi một khối lệnh synchronized.
public class LockExample {
private static final Object LOCK = new Object();
public static void main(String[] args) {
Thread thread1 = new Thread(() -> {
synchronized (LOCK) {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
Thread thread2 = new Thread(() -> {
synchronized (LOCK) {
System.out.println("Acquired lock");
}
});
thread1.start();
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
thread2.start();
}
}Trong khoảng thời gian Thread 1 đang giữ monitor, Thread 2 có thể chuyển sang BLOCKED khi cố lấy cùng monitor đó.
Lưu ý: Thread.sleep() không giải phóng monitor mà Thread đang nắm giữ.
1.3.4. WAITING
Thread đang chờ một Thread khác thực hiện một hành động để nó có thể tiếp tục. Một số thao tác có thể đưa Thread vào trạng thái này là:
- Object.wait() không có timeout.
- Thread.join() không có timeout.
- LockSupport.park() trong một số trường hợp.
Ví dụ:
Thread worker = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
worker.start();
try {
worker.join();
System.out.println("Worker completed");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}Thread gọi join() chờ Thread worker kết thúc. Sau đó, nó có thể tiếp tục thực thi.
1.3.5. TIMED_WAITING
Thread đang chờ trong một khoảng thời gian giới hạn. Các thao tác phổ biến bao gồm:
Thread.sleep(milliseconds)
Object.wait(timeout)
Thread.join(timeout)
LockSupport.parkNanos(...)
Khi hết thời gian chờ, Thread có thể tiếp tục tranh chấp tài nguyên hoặc chuyển về trạng thái có thể thực thi. Điều đó không đồng nghĩa với việc Thread chắc chắn chạy ngay lập tức.
1.3.6. TERMINATED
Thread đã kết thúc khi phương thức run() hoàn thành hoặc kết thúc do một ngoại lệ không được xử lý.
Một Thread Java không thể được khởi động lại sau khi đã kết thúc. Nếu cần thực hiện công việc mới, hãy tạo một Thread khác hoặc gửi task mới cho Executor.
1.3.7. Phân biệt start() và run()
Đây là một lỗi rất phổ biến khi mới học Multithreading.
Thread thread = new Thread(() -> {
System.out.println(
Thread.currentThread().getName()
);
});
thread.run(); // Gọi run() như một method thông thường
thread.start(); // Yêu cầu khởi động Thread mớiGọi run() trực tiếp không tạo ra một luồng thực thi mới. Phương thức được gọi ngay trên Thread hiện tại.
Ngược lại, start() yêu cầu JVM khởi động một Thread mới để thực thi phần công việc được định nghĩa trong run().
Trong ứng dụng thực tế, nên ưu tiên giao task cho
ExecutorService thay vì tự tạo Thread cho từng công việc không giới hạn. Executor giúp quản lý vòng đời thực thi, giới hạn tài nguyên và xử lý việc từ chối task.
1.4. Java Memory Model (JMM)
Java Memory Model là một trong những nền tảng quan trọng nhất để hiểu Concurrency. Nếu chỉ hiểu Thread là các luồng chạy song song mà không hiểu JMM, mình sẽ rất dễ viết những chương trình có hành vi khó dự đoán.
JMM định nghĩa các quy tắc về việc đọc, ghi và quan sát dữ liệu giữa các Thread, cũng như những mối quan hệ đồng bộ hóa cần thiết để đảm bảo tính đúng đắn của chương trình.
Mình cần tránh một cách diễn giải quá đơn giản: JMM không quy định rằng mọi Thread đều phải có một bản sao bộ nhớ vật lý riêng, và cũng không yêu cầu JVM phải thực hiện mọi thao tác thông qua việc sao chép dữ liệu từ Main Memory sang Working Memory.
Khái niệm Main Memory và Working Memory thường được sử dụng như mô hình minh họa. Trong thực tế, JIT compiler, CPU registers, cache và các cơ chế tối ưu hóa có thể ảnh hưởng đến cách dữ liệu được đọc và ghi, nhưng chúng phải tuân thủ các quy tắc của JMM.
Shared State
(Objects / Fields)
│
┌────────┴────────┐
│ │
▼ ▼
Thread A Thread B
│ │
Read / Write Read / Write
│ │
▼ ▼
Local Execution Local Execution
and Optimizations and Optimizations
│ │
└────────┬────────┘
│
▼
JMM Rules
Visibility / Ordering
/ Synchronization
1.4.1. Vấn đề Visibility
Visibility (tính hiển thị) là khả năng một Thread quan sát được những thay đổi dữ liệu do Thread khác thực hiện.
Ví dụ:
public class VisibilityProblem {
private static boolean running = true;
public static void main(String[] args) {
Thread worker = new Thread(() -> {
while (running) {
// Chờ yêu cầu dừng
}
System.out.println("Stopped");
});
worker.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
running = false;
}
}Đoạn code trên có Data Race vì hai Thread truy cập biến running, trong đó có một thao tác ghi, nhưng không có cơ chế đồng bộ hóa phù hợp.
Thread worker không được đảm bảo sẽ quan sát thấy giá trị false theo cách mà người viết mong đợi. Compiler hoặc runtime có thể tối ưu hóa việc đọc theo những cách hợp lệ dưới mô hình bộ nhớ Java.
Đây không phải là ví dụ để chứng minh rằng chương trình chắc chắn chạy vô hạn. Kết quả cụ thể phụ thuộc vào cách thực thi; điều quan trọng là chương trình không có bảo đảm cần thiết.
1.4.2. Vấn đề Ordering
Ordering liên quan đến thứ tự mà các thao tác đọc và ghi có thể được quan sát giữa các Thread.
Trong một Thread, Java bảo đảm các quy tắc thực thi phù hợp với ngữ nghĩa của chương trình. Tuy nhiên, giữa nhiều Thread không có đồng bộ hóa, mình không thể mặc định rằng mọi thao tác sẽ được quan sát theo đúng thứ tự được viết trong source code.
Compiler và CPU có thể thực hiện các tối ưu hóa như sắp xếp lại một số thao tác, miễn là không vi phạm những bảo đảm của mô hình bộ nhớ.
Vì vậy, khi một Thread ghi dữ liệu và sau đó thông báo cho Thread khác rằng dữ liệu đã sẵn sàng, mình cần cơ chế tạo ra quan hệ đồng bộ hóa phù hợp.
1.4.3. Vấn đề Atomicity
Atomicity (tính nguyên tử) là đặc tính một thao tác được xem như xảy ra trọn vẹn đối với các Thread liên quan, thay vì để các Thread khác quan sát được trạng thái trung gian không mong muốn.
Ví dụ:
count++;Về mặt logic, phép tăng thường tương đương với ba bước:
1. Đọc giá trị count
│
▼
2. Tính count + 1
│
▼
3. Ghi giá trị mới vào count
Nếu hai Thread cùng đọc giá trị 10, sau đó cùng tính ra 11 và ghi trở lại, kết quả cuối cùng có thể là 11 thay vì 12.
Đây là lý do vì sao một phép toán nhìn có vẻ đơn giản vẫn có thể gây ra lỗi trong môi trường đa luồng.
1.5. Thread-Safety là gì?
Một class hoặc đoạn code được xem là Thread-Safe khi nó duy trì được các bất biến và hành vi đúng đắn theo hợp đồng của mình trong những điều kiện truy cập đồng thời được hỗ trợ.
Thread-Safety không chỉ có nghĩa là không xảy ra Exception. Một chương trình có thể chạy ổn định trong thời gian dài nhưng vẫn trả về kết quả sai do Race Condition.
Có một số chiến lược phổ biến để đạt được Thread-Safety:
1. Immutability: Sử dụng object bất biến sau khi khởi tạo. 2. Thread Confinement: Giới hạn việc sử dụng dữ liệu trong một Thread. 3. Synchronization: Bảo vệ trạng thái chia sẻ bằng khóa. 4. Atomic Variables: Sử dụng các thao tác nguyên tử phù hợp. 5. Concurrent Collections: Sử dụng cấu trúc dữ liệu được thiết kế cho truy cập đồng thời. 6. Message Passing: Các Thread trao đổi thông điệp thay vì tùy ý sửa đổi chung một trạng thái.
Ví dụ về Immutability:
public final class Money {
private final long amount;
private final String currency;
public Money(long amount, String currency) {
this.amount = amount;
this.currency = currency;
}
public long getAmount() {
return amount;
}
public String getCurrency() {
return currency;
}
}Object Money không cho phép thay đổi hai field sau khi khởi tạo. Nếu những dữ liệu tham chiếu bên trong cũng bất biến hoặc được bảo vệ đúng cách, việc chia sẻ object này giữa các Thread sẽ đơn giản hơn nhiều.
Lưu ý rằng final chỉ ngăn thay đổi reference hoặc giá trị field theo quy tắc của Java; nó không tự động biến mọi object được tham chiếu thành bất biến.
2. Synchronization và cơ chế khóa
Sau khi hiểu các vấn đề Visibility, Ordering và Atomicity, mình cần tìm hiểu những cơ chế mà Java cung cấp để kiểm soát việc truy cập dữ liệu chia sẻ.
Các công cụ chính gồm synchronized, volatile, Lock, Atomic Variables và những cơ chế đồng bộ hóa cấp cao hơn.
2.1. Race Condition và Atomicity
Race Condition xảy ra khi kết quả hoặc tính đúng đắn của chương trình phụ thuộc vào thứ tự thực thi không được kiểm soát giữa nhiều thao tác đồng thời.
Ví dụ, mình xây dựng một bộ đếm dùng chung:
public class Counter {
private int count = 0;
public void increment() {
count++;
}
public int getCount() {
return count;
}
}Nếu chỉ có một Thread gọi increment(), chương trình có thể hoạt động đúng như mong đợi. Nhưng nếu nhiều Thread cùng gọi method này, các phép tăng có thể ghi đè kết quả của nhau.
Giả sử ban đầu count = 0:
Thread A Thread B │ │ ├── Read count = 0 │ │ ├── Read count = 0 │ ├── Compute 0 + 1 ├── Compute 0 + 1 │ │ ├── Write count = 1 ├── Write count = 1 │ │ │ ▼ ▼ Expected: 2 Actual: 1
Để chứng minh vấn đề này trong thực tế, mình có thể chạy nhiều Thread và tăng cùng một biến rất nhiều lần.
import java.util.ArrayList;
import java.util.List;
public class RaceConditionDemo {
private static int count = 0;
public static void main(String[] args)
throws InterruptedException {
int threadCount = 10;
int incrementsPerThread = 100_000;
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < threadCount; i++) {
Thread thread = new Thread(() -> {
for (int j = 0; j < incrementsPerThread; j++) {
count++;
}
});
threads.add(thread);
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
System.out.println("Expected: "
+ threadCount * incrementsPerThread);
System.out.println("Actual: " + count);
}
}Kết quả mong đợi là 1_000_000. Kết quả thực tế có thể thấp hơn do các phép tăng bị mất.
Không nên yêu cầu kết quả của một lần chạy cụ thể phải luôn thấp hơn một triệu. Race Condition khiến chương trình không có bảo đảm cần thiết; mức độ sai lệch phụ thuộc vào cách thực thi.
2.2. synchronized và Monitor Lock
synchronized là cơ chế đồng bộ hóa tích hợp của Java. Nó sử dụng intrinsic lock, còn được gọi là monitor lock, để bảo vệ một vùng code hoặc một method.
Quy tắc cốt lõi là tại một thời điểm, chỉ một Thread có thể sở hữu cùng một monitor để thực thi vùng code được bảo vệ bởi monitor đó.
Ví dụ sửa bộ đếm:
public class Counter {
private int count = 0;
public synchronized void increment() {
count++;
}
public synchronized int getCount() {
return count;
}
}Khi Thread gọi increment(), nó phải lấy monitor của đối tượng this. Nếu Thread khác đang giữ cùng monitor, Thread gọi sẽ chờ.
Khi method kết thúc, monitor được giải phóng tự động, kể cả khi method kết thúc do ngoại lệ.
2.2.1. Synchronized Instance Method
Với một instance method được khai báo synchronized, monitor được sử dụng là đối tượng hiện tại, tức this.
public synchronized void update() {
// Khóa trên this
}Đoạn code trên tương đương về ý nghĩa đồng bộ hóa với:
public void update() {
synchronized (this) {
// Khóa trên this
}
}Điều này có nghĩa là hai instance khác nhau của cùng một class có hai monitor khác nhau. Hai Thread có thể thực thi đồng thời trên hai instance khác nhau, miễn là chúng không tranh chấp một tài nguyên chia sẻ khác.
2.2.2. Synchronized Static Method
Với static method được khai báo synchronized, monitor thuộc về đối tượng Class tương ứng.
public class Configuration {
public static synchronized void reload() {
// Khóa trên Configuration.class
}
}Về ý nghĩa đồng bộ hóa, đoạn code trên tương đương với:
public static void reload() {
synchronized (Configuration.class) {
// Khóa trên Configuration.class
}
}Không nên nhầm monitor của class với monitor của một instance. Chúng là hai đối tượng khóa khác nhau.
2.2.3. Synchronized Block
Synchronized block cho phép mình kiểm soát chính xác vùng code cần khóa, thay vì khóa toàn bộ method.
public class Account {
private final Object lock = new Object();
private long balance;
public void deposit(long amount) {
if (amount <= 0) {
throw new IllegalArgumentException();
}
synchronized (lock) {
balance += amount;
}
}
public long getBalance() {
synchronized (lock) {
return balance;
}
}
}Ở ví dụ này, cả thao tác cập nhật và đọc số dư đều sử dụng cùng một monitor. Nhờ đó, các Thread truy cập qua những method này tuân thủ cùng một cơ chế đồng bộ hóa.
Trong hệ thống thực tế, mình nên giữ vùng khóa ngắn và không thực hiện những thao tác I/O chậm trong khi giữ khóa nếu không thực sự cần thiết.
2.2.4. Reentrant Locking
Intrinsic lock của Java có tính chất reentrant. Nghĩa là một Thread đang giữ monitor có thể tiếp tục đi vào những vùng synchronized khác sử dụng cùng monitor đó.
public class ReentrantExample {
public synchronized void methodA() {
methodB();
}
public synchronized void methodB() {
System.out.println("Executing methodB");
}
}Khi một Thread gọi methodA(), nó lấy monitor của this. Khi tiếp tục gọi methodB(), Thread đó không bị chặn bởi chính monitor mà nó đang sở hữu.
Nếu intrinsic lock không reentrant, những lời gọi lồng nhau như vậy có thể tự chặn chính Thread đang thực thi.
2.2.5. Cơ chế hoạt động ở tầng JVM
Trong bytecode Java, synchronized block thường được biểu diễn bằng các lệnh monitorenter và monitorexit. Synchronized method được đánh dấu bằng cờ đồng bộ hóa trong metadata của method.
Đây là mô hình bytecode ở mức khái niệm:
Thread muốn vào vùng synchronized
│
▼
Yêu cầu monitor lock
│
┌───────┴────────┐
│ │
▼ ▼
Lock available Lock occupied
│ │
▼ ▼
Acquire lock Wait / contend
│ │
▼ │
Execute code │
│ │
▼ │
Release lock ◄───────┘
│
▼
Thread khác có cơ hội
lấy monitor
Ở HotSpot JVM, cơ chế triển khai monitor và tối ưu hóa khóa có thể thay đổi theo phiên bản, kiến trúc phần cứng và tình trạng tranh chấp. Vì vậy, không nên coi một sơ đồ triển khai nội bộ cụ thể là quy tắc bắt buộc của JVM Specification.
2.3. Happens-Before Relationship
Happens-Before là một khái niệm trung tâm của Java Memory Model. Nó xác định quan hệ thứ tự giữa các thao tác, từ đó tạo ra những bảo đảm về khả năng quan sát dữ liệu.
Nếu thao tác A happens-before thao tác B, những thay đổi được thực hiện trước A có thể được bảo đảm hiển thị với B theo các quy tắc của JMM, với điều kiện không có thao tác ghi xen giữa làm thay đổi giá trị liên quan.
Mình không nên hiểu Happens-Before đơn giản là thời điểm thực tế A xảy ra trước B trên đồng hồ. Đây là một quan hệ ngữ nghĩa về đồng bộ hóa và tính nhất quán của bộ nhớ.
2.3.1. Các quy tắc quan trọng
Program Order Rule: Trong một Thread, thứ tự thực thi theo ngữ nghĩa của chương trình tạo ra các quan hệ happens-before tương ứng.
Monitor Lock Rule: Một thao tác unlock trên monitor happens-before những lần lock tiếp theo thành công trên cùng monitor đó.
Volatile Rule: Một thao tác ghi vào một biến volatile happens-before những lần đọc tiếp theo của cùng biến đó theo thứ tự đồng bộ hóa.
Thread Start Rule: Những thao tác trước khi gọi Thread.start() happens-before những thao tác trong Thread được khởi động.
Thread Join Rule: Những thao tác của một Thread happens-before một Thread khác phát hiện rằng Thread đó đã kết thúc, chẳng hạn thông qua việc join() hoàn thành.
Transitivity Rule: Nếu A happens-before B và B happens-before C, thì A happens-before C.
2.3.2. Ví dụ với synchronized
public class SafePublication {
private int value = 0;
private final Object lock = new Object();
public void write() {
synchronized (lock) {
value = 42;
}
}
public int read() {
synchronized (lock) {
return value;
}
}
}Nếu Thread A ghi value = 42 trong vùng khóa và sau đó Thread B lấy chính monitor đó trước khi đọc, thao tác unlock của A và lock thành công của B tạo ra quan hệ happens-before.
Nhờ đó, Thread B có thể quan sát thay đổi tương ứng với việc ghi trước đó, miễn là không có thao tác ghi mới hơn làm thay đổi kết quả.
Điểm quan trọng là cả hai Thread phải phối hợp qua cùng cơ chế đồng bộ hóa. Nếu Thread B đọc trực tiếp mà không tuân thủ cơ chế đó, mình không thể dựa vào quan hệ monitor ở trên để suy ra mọi bảo đảm cần thiết.
2.4. volatile: Visibility nhưng không đảm bảo mọi Atomicity
Từ khóa volatile được sử dụng khi mình cần đảm bảo các Thread quan sát được giá trị mới nhất theo quy tắc của JMM mà không cần sử dụng một khóa monitor cho từng lần đọc hoặc ghi.
Ví dụ phổ biến nhất là một cờ thông báo dừng:
public class Worker {
private volatile boolean running = true;
public void run() {
while (running) {
doWork();
}
}
public void stop() {
running = false;
}
private void doWork() {
// Thực hiện một đơn vị công việc
}
}Thread thực thi run() đọc biến running, trong khi Thread khác có thể gọi stop() để ghi giá trị false.
Vì biến là volatile, thao tác ghi và các lần đọc biến đó tuân thủ quy tắc visibility của JMM. Thread đang chạy có thể nhận biết yêu cầu dừng mà không cần khóa monitor cho biến cờ.
2.4.1. Vì sao volatile không làm count++ trở thành nguyên tử?
private volatile int count = 0;
public void increment() {
count++;
}Ở đây, các lần đọc và ghi biến count đều tuân thủ quy tắc của volatile. Tuy nhiên, phép count++ vẫn gồm việc đọc, tính toán và ghi lại giá trị.
Hai Thread vẫn có thể đọc cùng một giá trị trước khi ghi kết quả, dẫn đến Lost Update.
Visibility: Một Thread quan sát được thay đổi dữ liệu.
Ordering: Các thao tác được quan sát theo những quan hệ thứ tự được bảo đảm.
Atomicity: Một thao tác hoặc một chuỗi thao tác không bị chia cắt theo cách làm sai kết quả.
volatile hỗ trợ Visibility và Ordering theo JMM, nhưng không biến một chuỗi thao tác như count++ thành một thao tác nguyên tử.
2.4.2. Khi nào nên sử dụng volatile?
Những trường hợp phù hợp bao gồm:
- Cờ dừng một Worker Thread. - Trạng thái đơn giản chỉ cần đọc và ghi độc lập. - Công bố một giá trị thay thế bất biến khi các điều kiện publication được đáp ứng.
Không nên dùng volatile như một giải pháp chung cho các biến cần cập nhật theo nhiều bước hoặc cần duy trì một bất biến giữa nhiều field.
2.5. ReentrantLock và Condition
ReentrantLock thuộc package java.util.concurrent.locks. Đây là một cơ chế khóa tường minh có khả năng reentrant, tương tự intrinsic lock về đặc tính tái nhập nhưng cung cấp thêm một số khả năng điều khiển.
Ví dụ:
import java.util.concurrent.locks.ReentrantLock;
public class SafeCounter {
private final ReentrantLock lock =
new ReentrantLock();
private int count;
public void increment() {
lock.lock();
try {
count++;
} finally {
lock.unlock();
}
}
public int getCount() {
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
}Điểm quan trọng nhất của đoạn code trên là unlock() được đặt trong finally. Nếu code bên trong phát sinh ngoại lệ, khóa vẫn được giải phóng.
Khác với synchronized, khóa tường minh đòi hỏi lập trình viên chịu trách nhiệm giải phóng khóa đúng cách.
2.5.1. tryLock()
tryLock() cho phép thử lấy khóa mà không nhất thiết phải chờ vô thời hạn.
if (lock.tryLock()) {
try {
// Thực hiện công việc
} finally {
lock.unlock();
}
} else {
// Không lấy được khóa
// Có thể thử lại hoặc xử lý theo chính sách khác
}Ngoài ra, tryLock(timeout, unit) cho phép giới hạn thời gian chờ và hỗ trợ phản ứng với interruption.
2.5.2. Lock có thể interrupt
lockInterruptibly() cho phép Thread bị gián đoạn trong quá trình chờ lấy khóa.
try {
lock.lockInterruptibly();
try {
// Vùng code được bảo vệ
} finally {
lock.unlock();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}Khả năng phản ứng với interruption rất hữu ích khi xây dựng hệ thống cần dừng tác vụ có kiểm soát.
2.5.3. Fair Lock và Non-Fair Lock
ReentrantLock cho phép lựa chọn chế độ công bằng thông qua constructor.
ReentrantLock fairLock =
new ReentrantLock(true);
ReentrantLock nonFairLock =
new ReentrantLock(false);Fair lock cố gắng ưu tiên Thread chờ lâu hơn theo chính sách của khóa. Non-fair lock có thể cho phép Thread mới lấy khóa trước Thread đang chờ.
Fairness không đồng nghĩa với việc hệ thống luôn nhanh hơn hoặc không bao giờ xảy ra starvation. Fair lock thường có thể làm giảm throughput trong một số workload.
2.5.4. Condition
Condition cho phép một khóa có nhiều điều kiện chờ riêng biệt. Đây là khả năng hữu ích trong những cấu trúc đồng bộ hóa phức tạp hơn.
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class SignalExample {
private final ReentrantLock lock =
new ReentrantLock();
private final Condition condition =
lock.newCondition();
private boolean ready = false;
public void awaitReady() throws InterruptedException {
lock.lockInterruptibly();
try {
while (!ready) {
condition.await();
}
System.out.println("Ready!");
} finally {
lock.unlock();
}
}
public void signalReady() {
lock.lock();
try {
ready = true;
condition.signalAll();
} finally {
lock.unlock();
}
}
}Phải kiểm tra điều kiện trong vòng lặp while vì Thread có thể thức dậy khi điều kiện chưa được đáp ứng hoặc có Thread khác thay đổi trạng thái trước khi nó tiếp tục.
Với Condition.await(), Thread giải phóng khóa tương ứng trong quá trình chờ và lấy lại khóa trước khi tiếp tục thực thi.
2.6. ReadWriteLock và StampedLock
Trong một số ứng dụng, số lần đọc dữ liệu nhiều hơn đáng kể so với số lần ghi. Khi đó, mình có thể cân nhắc sử dụng cơ chế Read-Write Lock.
ReadWriteLock cung cấp hai loại khóa:
Read Lock: Cho phép nhiều Reader cùng giữ khóa trong những điều kiện phù hợp. Write Lock: Dành cho thao tác ghi độc quyền.
Ví dụ với ReentrantReadWriteLock:
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ProductCache {
private final Map<String, String> cache =
new HashMap<>();
private final ReadWriteLock lock =
new ReentrantReadWriteLock();
public String get(String key) {
lock.readLock().lock();
try {
return cache.get(key);
} finally {
lock.readLock().unlock();
}
}
public void put(String key, String value) {
lock.writeLock().lock();
try {
cache.put(key, value);
} finally {
lock.writeLock().unlock();
}
}
}Nhiều Thread có thể đọc cache đồng thời. Khi một Thread ghi, thao tác ghi cần khóa độc quyền để tránh xung đột với các thao tác đọc hoặc ghi khác.
Tuy nhiên, ReadWriteLock không phải lúc nào cũng nhanh hơn synchronized. Nếu critical section rất ngắn hoặc số lần ghi cao, chi phí quản lý hai loại khóa có thể làm giảm hiệu năng.
2.6.1. StampedLock
StampedLock cung cấp các chế độ khóa đọc, khóa ghi và optimistic read. Optimistic read cho phép đọc dữ liệu trước mà chưa lấy read lock thực sự, sau đó kiểm tra xem có thao tác ghi nào làm mất hiệu lực của kết quả đọc hay không.
import java.util.concurrent.locks.StampedLock;
public class Point {
private double x;
private double y;
private final StampedLock lock =
new StampedLock();
public double distanceFromOrigin() {
long stamp = lock.tryOptimisticRead();
double currentX = x;
double currentY = y;
if (!lock.validate(stamp)) {
stamp = lock.readLock();
try {
currentX = x;
currentY = y;
} finally {
lock.unlockRead(stamp);
}
}
return Math.hypot(currentX, currentY);
}
}Nếu có thao tác ghi xảy ra trong khoảng đọc, chương trình lấy read lock để đọc lại dữ liệu theo cơ chế an toàn.
Không nên coi optimistic read là cách đọc nhất quán mọi trạng thái object trong mọi tình huống. Với dữ liệu phức tạp, cần hiểu rõ điều kiện validate và cách bảo vệ các field liên quan.
Ngoài ra, StampedLock không reentrant. Không nên sử dụng nó như một bản thay thế trực tiếp cho ReentrantLock khi code phụ thuộc vào tính tái nhập.
2.6.2. So sánh các cơ chế khóa
Nhu cầu
│
├── Bảo vệ critical section thông thường
│ └── synchronized
│
├── Cần timeout / interrupt / tryLock
│ └── ReentrantLock
│
├── Nhiều Reader, ít Writer
│ └── ReadWriteLock
│
└── Optimistic Read và kiểm tra stamp
└── StampedLock
Đây là hướng lựa chọn ban đầu, không phải quy tắc tuyệt đối. Cần đo đạc hiệu năng và kiểm tra đặc điểm của workload trước khi thay đổi cơ chế đồng bộ hóa.
2.7. Atomic Variables và thuật toán CAS
Java cung cấp nhiều lớp nguyên tử trong package java.util.concurrent.atomic, chẳng hạn:
AtomicInteger
AtomicLong
AtomicBoolean
AtomicReference<T>
AtomicIntegerArray
LongAdder
Các lớp này cung cấp những thao tác nguyên tử phù hợp với từng kiểu dữ liệu. Nhiều thao tác được triển khai bằng cơ chế CAS hoặc những primitive nguyên tử tương đương tùy nền tảng.
2.7.1. Compare-And-Swap (CAS)
CAS là một primitive nguyên tử có thể mô tả bằng ba giá trị:
Expected: Giá trị mà Thread mong đợi. Update: Giá trị mới cần ghi. Current: Giá trị hiện tại trong vùng nhớ.
Logic khái niệm:
CAS(expected, update)
│
▼
Đọc current atomically
│
┌─────┴──────┐
│ │
current == current !=
expected expected
│ │
▼ ▼
Ghi update Không ghi
│ │
▼ ▼
Success Failure
CAS giúp Thread thử cập nhật trạng thái mà không nhất thiết phải khóa một critical section theo kiểu mutex. Tuy nhiên, việc CAS thất bại có thể khiến chương trình phải thử lại.
2.7.2. AtomicInteger
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicCounter {
private final AtomicInteger count =
new AtomicInteger(0);
public void increment() {
count.incrementAndGet();
}
public int getCount() {
return count.get();
}
}incrementAndGet() là thao tác tăng nguyên tử. Khi nhiều Thread cùng gọi method này, mỗi lần tăng được tính đến trong giá trị cuối cùng.
AtomicInteger phù hợp với bộ đếm đơn giản. Nhưng nếu cần cập nhật nhiều field theo một bất biến chung, chỉ sử dụng một AtomicInteger không giải quyết được toàn bộ bài toán.
2.7.3. compareAndSet()
AtomicInteger value = new AtomicInteger(10);
boolean updated = value.compareAndSet(10, 20);
System.out.println(updated); // true nếu CAS thành công
System.out.println(value.get()); // 20 nếu thành côngNếu giá trị hiện tại không còn bằng 10 tại thời điểm CAS thực hiện, thao tác sẽ thất bại và không ghi giá trị 20.
2.7.4. CAS Loop
Một số thao tác cập nhật có điều kiện có thể được triển khai theo mẫu thử lại:
AtomicInteger value = new AtomicInteger(10);
int updateValue(int delta) {
while (true) {
int current = value.get();
int next = current + delta;
if (value.compareAndSet(current, next)) {
return next;
}
}
}Nếu Thread khác cập nhật giá trị trước khi CAS hoàn thành, vòng lặp sẽ đọc lại giá trị mới và thử lại.
Trong code thực tế, method này cần nằm trong một class và sử dụng field tương ứng. Ví dụ được viết dưới dạng minh họa thuật toán.
2.7.5. Lock-Free không đồng nghĩa với Wait-Free
Lock-Free là thuộc tính tiến triển của thuật toán: trong điều kiện thực thi liên tục, toàn hệ thống vẫn có Thread hoàn thành thao tác, ngay cả khi một số Thread có thể phải thử lại nhiều lần.
Wait-Free là bảo đảm mạnh hơn: mỗi thao tác hoàn thành sau một số bước hữu hạn được giới hạn độc lập với tiến triển của các Thread khác.
Một CAS loop không tự động trở thành Wait-Free. Nó có thể liên tục thất bại khi nhiều Thread cạnh tranh cùng một biến.
2.7.6. LongAdder
Với bộ đếm có nhiều Thread cập nhật đồng thời, LongAdder có thể giảm tranh chấp bằng cách phân tán cập nhật trên nhiều ô đếm nội bộ.
import java.util.concurrent.atomic.LongAdder;
public class RequestMetrics {
private final LongAdder totalRequests =
new LongAdder();
public void recordRequest() {
totalRequests.increment();
}
public long getTotalRequests() {
return totalRequests.sum();
}
}LongAdder đặc biệt hữu ích cho metrics và thống kê có nhiều thao tác cập nhật. Tuy nhiên, sum() không phải snapshot nguyên tử của toàn bộ quá trình cập nhật đồng thời. Nếu cần một giá trị đơn lẻ có ngữ nghĩa nguyên tử nghiêm ngặt, cần chọn công cụ phù hợp hơn.
3. Concurrency Utilities trong java.util.concurrent
Java cung cấp một hệ sinh thái lớn để xây dựng các chương trình đồng thời. Thay vì tự triển khai mọi cơ chế từ đầu, mình nên ưu tiên các API đã được thiết kế và kiểm chứng kỹ lưỡng.
3.1. Thread và ExecutorService
Cách truyền thống là tạo Thread trực tiếp:
Thread thread = new Thread(() -> {
System.out.println("Processing task");
});
thread.start();Cách này phù hợp với ví dụ đơn giản hoặc những tình huống cần quản lý Thread cụ thể. Nhưng nếu ứng dụng phải xử lý hàng nghìn task, việc tạo một Thread mới cho từng task có thể gây ra chi phí đáng kể.
Những chi phí này có thể bao gồm bộ nhớ cho stack, chi phí khởi tạo Thread, lập lịch của hệ điều hành và chi phí chuyển đổi ngữ cảnh.
Để giải quyết bài toán đó, Java cung cấp Executor và ExecutorService.
Executor tách việc gửi task khỏi cách thực thi task. ExecutorService mở rộng mô hình này với các khả năng quản lý vòng đời, trả về Future và shutdown.
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ExecutorExample {
public static void main(String[] args) {
ExecutorService executor =
Executors.newFixedThreadPool(4);
try {
for (int i = 0; i < 10; i++) {
int taskId = i;
executor.submit(() -> {
System.out.println(
"Task " + taskId
+ " executed by "
+ Thread.currentThread().getName()
);
});
}
} finally {
executor.shutdown();
}
}
}Trong ví dụ này, pool có bốn worker Thread. Các task được gửi vào Executor và được thực thi bởi những worker đó khi có thể.
Thứ tự gửi task không bảo đảm rằng output sẽ xuất hiện theo thứ tự tăng dần của taskId.
3.1.1. submit() và execute()
execute(Runnable) gửi một task không có kết quả trả về thông qua interface Executor.
submit(...) của ExecutorService trả về một Future để mình theo dõi kết quả hoặc trạng thái hoàn thành của task.
Ví dụ:
import java.util.concurrent.*;
public class FutureExample {
public static void main(String[] args)
throws Exception {
ExecutorService executor =
Executors.newSingleThreadExecutor();
try {
Future<Integer> future =
executor.submit(() -> 10 + 20);
Integer result = future.get();
System.out.println(result);
} finally {
executor.shutdown();
}
}
}future.get() chờ kết quả nếu task chưa hoàn thành. Vì vậy, gọi get() ngay sau mỗi lần submit trong một vòng lặp có thể vô tình khiến chương trình hoạt động gần như tuần tự.
3.2. Thread Pool và các thông số cốt lõi
Thread Pool là cơ chế tái sử dụng các Thread để thực hiện nhiều task. Một Thread sau khi hoàn thành task có thể tiếp tục lấy task khác từ hàng đợi.
Đối với ThreadPoolExecutor, các tham số quan trọng gồm:
corePoolSize: Số lượng worker cốt lõi mà pool duy trì theo chính sách vòng đời của nó. maximumPoolSize: Số lượng worker tối đa có thể được tạo khi cần. keepAliveTime: Thời gian một số worker nhàn rỗi có thể tồn tại trước khi bị loại bỏ. WorkQueue: Hàng đợi chứa task chờ thực thi. ThreadFactory: Cách tạo worker Thread. RejectedExecutionHandler: Chính sách xử lý khi Executor không thể nhận task mới.
3.2.1. Cách ThreadPoolExecutor xử lý task
Với một cấu hình ThreadPoolExecutor thông thường, trình tự tiếp nhận task có thể mô tả như sau:
Task submitted
│
▼
Worker count < corePoolSize?
│
┌───┴───┐
Yes No
│ │
▼ ▼
Create Queue has capacity?
worker │
┌────┴────┐
Yes No
│ │
▼ ▼
Enqueue Worker count
< maximumPoolSize?
│
┌────┴────┐
Yes No
│ │
▼ ▼
Create more Reject task
Sơ đồ này mô tả luồng tiếp nhận task cơ bản. Một số chi tiết như trạng thái shutdown, việc tạo worker ban đầu và các tình huống race khi thêm task được xử lý thêm bên trong implementation.
3.2.2. Ví dụ cấu hình ThreadPoolExecutor
import java.util.concurrent.*;
public class CustomThreadPool {
public static void main(String[] args) {
int corePoolSize = 4;
int maximumPoolSize = 8;
long keepAliveTime = 30;
BlockingQueue<Runnable> queue =
new ArrayBlockingQueue<>(100);
ThreadPoolExecutor executor =
new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
TimeUnit.SECONDS,
queue,
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
try {
executor.execute(() ->
System.out.println("Processing request")
);
} finally {
executor.shutdown();
}
}
}Ở ví dụ này, pool có thể sử dụng tối đa tám worker và hàng đợi có sức chứa một trăm task.
Khi các worker cốt lõi đã tồn tại, pool thường ưu tiên đưa task vào queue trước. Nếu queue đầy, pool mới xem xét tạo worker vượt quá corePoolSize, miễn là chưa đạt maximumPoolSize.
Khi pool không thể tiếp nhận task mới, AbortPolicy ném ra RejectedExecutionException.
3.2.3. Vì sao hàng đợi không giới hạn có thể nguy hiểm?
Trong nhiều cấu hình, Executors.newFixedThreadPool(n) sử dụng một hàng đợi không giới hạn. Khi tốc độ tiếp nhận task lớn hơn tốc độ xử lý, số lượng task chờ có thể tăng liên tục.
Điều này dẫn đến những vấn đề như:
- Bộ nhớ tăng do nhiều task tồn tại trong queue. - Độ trễ tăng vì task phải chờ lâu hơn. - Các request hết thời gian chờ trước khi được xử lý. - Hệ thống bị quá tải dây chuyền khi nhiều request cùng phụ thuộc vào một pool.
Hàng đợi hữu hạn giúp tạo một giới hạn tài nguyên rõ ràng hơn, nhưng cũng cần xác định chính sách xử lý khi queue đầy.
3.2.4. Rejection Policy
Java cung cấp các chính sách phổ biến:
AbortPolicy: Ném RejectedExecutionException.
CallerRunsPolicy: Nếu executor chưa shutdown, Thread gọi gửi task có thể trực tiếp thực thi task đó. Điều này có thể tạo hiệu ứng giảm tốc độ gửi task.
DiscardPolicy: Bỏ task bị từ chối mà không ném ngoại lệ.
DiscardOldestPolicy: Loại bỏ task cũ nhất trong queue rồi thử gửi task mới, theo điều kiện của chính sách.
Không có chính sách nào phù hợp với tất cả hệ thống. Ví dụ, bỏ task có thể chấp nhận được với một số metrics tạm thời nhưng hoàn toàn không phù hợp với giao dịch tài chính.
Không chỉ giới hạn số lượng worker. Hãy xem xét cả dung lượng queue, thời gian timeout, khả năng retry, giới hạn kết nối database và chính sách backpressure. Nếu hệ thống chỉ giới hạn Thread nhưng vẫn nhận task vô hạn, vấn đề cạn kiệt tài nguyên vẫn có thể xảy ra.
3.2.5. Shutdown đúng cách
shutdown() ngừng nhận task mới nhưng cho phép những task đã được chấp nhận tiếp tục hoàn thành.
shutdownNow() cố gắng dừng các task đang thực thi bằng cách interrupt worker và trả về những task chưa bắt đầu theo khả năng của implementation. Nó không bảo đảm mọi task lập tức kết thúc.
import java.util.concurrent.*;
public class ExecutorShutdown {
public static void shutdownGracefully(
ExecutorService executor) {
executor.shutdown();
try {
if (!executor.awaitTermination(
30, TimeUnit.SECONDS)) {
executor.shutdownNow();
if (!executor.awaitTermination(
10, TimeUnit.SECONDS)) {
System.err.println(
"Executor did not terminate"
);
}
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
}Đoạn code trên minh họa quy trình shutdown có thời hạn. Hiệu quả của việc dừng vẫn phụ thuộc vào việc task có phản ứng đúng với interruption hay không.
3.3. Future và CompletableFuture
Future biểu diễn kết quả của một computation có thể hoàn thành trong tương lai.
Những thao tác phổ biến gồm:
get(): Lấy kết quả, có thể chờ.
get(timeout, unit): Chờ có giới hạn thời gian.
isDone(): Kiểm tra trạng thái hoàn thành.
cancel(...): Yêu cầu hủy task.
Future truyền thống khá hữu ích, nhưng thường cần thêm code điều phối khi một task phụ thuộc vào kết quả của task khác.
CompletableFuture hỗ trợ xây dựng pipeline bất đồng bộ, kết hợp nhiều kết quả và xử lý ngoại lệ theo chuỗi.
3.3.1. supplyAsync() và runAsync()
supplyAsync() sử dụng một Supplier và tạo Future chứa kết quả.
runAsync() thực hiện một Runnable không trả về kết quả.
import java.util.concurrent.CompletableFuture;
public class CompletableFutureExample {
public static void main(String[] args) {
CompletableFuture<Integer> future =
CompletableFuture.supplyAsync(() -> 10 + 20);
future.thenAccept(result ->
System.out.println("Result: " + result)
);
}
}Nếu không truyền Executor riêng, các phương thức async thường sử dụng common ForkJoinPool. Vì vậy, cần chú ý đến loại công việc được đưa vào pool này.
3.3.2. thenApply() và thenCompose()
thenApply() biến đổi kết quả của một stage thành một giá trị mới.
CompletableFuture<String> future =
CompletableFuture.supplyAsync(() -> "java")
.thenApply(String::toUpperCase)
.thenApply(value -> "Language: " + value);Kết quả cuối cùng là Language: JAVA.
thenCompose() dùng khi bước tiếp theo trả về một CompletionStage khác và mình muốn nối chúng thành một pipeline phẳng.
CompletableFuture<User> findUser(String id) {
return CompletableFuture.supplyAsync(
() -> new User(id)
);
}
CompletableFuture<Order> findOrder(User user) {
return CompletableFuture.supplyAsync(
() -> new Order(user.getId())
);
}
CompletableFuture<Order> future =
findUser("U001")
.thenCompose(this::findOrder);Ví dụ minh họa này cần đặt trong một class có các method và kiểu dữ liệu tương ứng. Ý tưởng chính là kết quả của bước sau phụ thuộc vào kết quả bước trước.
3.3.3. thenCombine() và allOf()
Nếu hai tác vụ độc lập, mình có thể cho chúng chạy đồng thời và kết hợp kết quả.
CompletableFuture<Integer> priceFuture =
CompletableFuture.supplyAsync(() -> 100);
CompletableFuture<Integer> shippingFuture =
CompletableFuture.supplyAsync(() -> 20);
CompletableFuture<Integer> totalFuture =
priceFuture.thenCombine(
shippingFuture,
Integer::sum
);thenCombine() kết hợp hai kết quả thành một giá trị mới.
CompletableFuture.allOf(...) cho phép chờ nhiều Future hoàn thành, nhưng bản thân nó không tự thu thập tất cả kết quả thành một danh sách có kiểu dữ liệu cụ thể.
CompletableFuture<String> a =
CompletableFuture.supplyAsync(() -> "A");
CompletableFuture<String> b =
CompletableFuture.supplyAsync(() -> "B");
CompletableFuture<Void> all =
CompletableFuture.allOf(a, b);
CompletableFuture<String> combined =
all.thenApply(ignored ->
a.join() + b.join()
);Ở đây, join() nằm sau stage allOf(), khi hai Future đã hoàn thành. Nếu một Future thất bại, stage tổng hợp có thể hoàn thành ngoại lệ.
3.3.4. thenApply() và thenApplyAsync()
Các method không có hậu tố Async có thể thực thi callback ngay trên Thread hoàn thành stage trước hoặc Thread gọi, tùy trạng thái và cách triển khai.
Những method có hậu tố Async thường lên lịch thực thi callback qua một Executor. Nếu không truyền Executor, chúng thường sử dụng common ForkJoinPool.
Vì vậy, mình không nên giả định rằng mọi callback của CompletableFuture đều chạy trên một Thread mới hoặc một Thread riêng biệt.
3.3.5. Xử lý ngoại lệ
Ba method thường gặp là exceptionally(), handle() và whenComplete().
exceptionally(): Tạo giá trị thay thế khi stage trước thất bại.
handle(): Xử lý cả kết quả thành công và ngoại lệ, sau đó tạo kết quả mới.
whenComplete(): Thực hiện hành động quan sát kết quả hoặc ngoại lệ mà không chủ đích biến đổi kết quả thành công.
CompletableFuture<Integer> future =
CompletableFuture.supplyAsync(() -> {
if (Math.random() > 0.5) {
throw new RuntimeException("Failed");
}
return 100;
})
.exceptionally(ex -> {
System.err.println(ex.getMessage());
return 0;
});Ví dụ trên trả về 0 khi stage thất bại và xử lý ngoại lệ bằng một giá trị thay thế.
Với handle(), mình có thể viết:
CompletableFuture<String> result =
CompletableFuture.supplyAsync(() -> 100)
.handle((value, exception) -> {
if (exception != null) {
return "ERROR";
}
return "SUCCESS: " + value;
});3.3.6. Timeout và Cancellation
Java cung cấp các API như orTimeout() và completeOnTimeout() để xử lý những tình huống chờ quá lâu.
CompletableFuture<String> future =
CompletableFuture.supplyAsync(() -> {
return callRemoteService();
});
future.orTimeout(2, TimeUnit.SECONDS);Trong ví dụ này, Future có thể hoàn thành ngoại lệ nếu vượt quá thời hạn. Tuy nhiên, timeout của CompletableFuture không tự động bảo đảm rằng tác vụ I/O bên dưới đã bị hủy hoặc tài nguyên bên ngoài đã được giải phóng.
Đối với HTTP client hoặc database client, cần sử dụng timeout và cơ chế cancellation tương ứng của thư viện đó.
3.3.7. Ví dụ pipeline bất đồng bộ
import java.util.concurrent.*;
public class OrderPipeline {
public static void main(String[] args) {
ExecutorService ioExecutor =
Executors.newFixedThreadPool(4);
try {
CompletableFuture<String> future =
CompletableFuture
.supplyAsync(
() -> loadOrder("ORD-001"),
ioExecutor
)
.thenApplyAsync(
OrderPipeline::enrichOrder,
ioExecutor
)
.thenApply(
value -> "Result: " + value
)
.exceptionally(
ex -> "Fallback: " + ex.getMessage()
);
System.out.println(future.join());
} finally {
ioExecutor.shutdown();
}
}
static String loadOrder(String id) {
return "Order " + id;
}
static String enrichOrder(String order) {
return order + " [enriched]";
}
}Ví dụ minh họa cách chỉ định Executor cho các bước bất đồng bộ. Trong ứng dụng thật, cần thiết kế thời gian chờ, chính sách retry, logging và việc shutdown theo vòng đời của hệ thống.
Asynchronous không đồng nghĩa với Parallel. Một pipeline có thể chạy bất đồng bộ nhưng vẫn có các bước phụ thuộc tuần tự. Chỉ những tác vụ độc lập mới có thể được thực thi song song một cách có ý nghĩa.
3.4. CountDownLatch, CyclicBarrier và Semaphore
Đây là ba công cụ đồng bộ hóa thường gặp trong các ứng dụng Java thực tế. Chúng giải quyết những bài toán khác nhau và không nên thay thế lẫn nhau một cách máy móc.
3.4.1. CountDownLatch
CountDownLatch cho phép một hoặc nhiều Thread chờ cho đến khi một bộ đếm về 0.
Ví dụ, một ứng dụng cần đợi ba bước khởi tạo hoàn thành trước khi bắt đầu nhận request.
import java.util.concurrent.CountDownLatch;
public class StartupExample {
public static void main(String[] args)
throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
new Thread(() -> {
try {
initializeComponent();
} finally {
latch.countDown();
}
}).start();
}
latch.await();
System.out.println("All components initialized");
}
static void initializeComponent() {
// Khởi tạo một component
}
}Phương thức countDown() giảm bộ đếm. Khi bộ đếm về 0, các Thread đang chờ bằng await() được phép tiếp tục.
CountDownLatch không thể reset về giá trị ban đầu. Nếu cần cơ chế chờ có thể tái sử dụng cho nhiều vòng, hãy xem xét CyclicBarrier hoặc những công cụ phù hợp khác.
3.4.2. CyclicBarrier
CyclicBarrier cho phép một nhóm Thread chờ nhau tại một điểm đồng bộ hóa. Khi đủ số Thread đến barrier, chúng có thể tiếp tục sang giai đoạn tiếp theo.
import java.util.concurrent.CyclicBarrier;
public class BarrierExample {
public static void main(String[] args) {
int workers = 3;
CyclicBarrier barrier =
new CyclicBarrier(workers, () ->
System.out.println(
"All workers reached the barrier"
)
);
for (int i = 0; i < workers; i++) {
int workerId = i;
new Thread(() -> {
try {
System.out.println(
"Worker " + workerId + " preparing"
);
barrier.await();
System.out.println(
"Worker " + workerId + " continuing"
);
} catch (Exception e) {
Thread.currentThread().interrupt();
}
}).start();
}
}
}Barrier có thể được sử dụng qua nhiều vòng đồng bộ hóa. Tuy nhiên, nếu một Thread rời barrier do interruption, timeout hoặc một lỗi liên quan, barrier có thể chuyển sang trạng thái broken và các Thread đang chờ cũng phải xử lý tình huống đó.
3.4.3. Semaphore
Semaphore quản lý một số lượng permit hữu hạn. Một Thread cần lấy permit trước khi sử dụng tài nguyên được giới hạn.
Ví dụ, hệ thống chỉ cho phép tối đa năm công việc sử dụng đồng thời một tài nguyên đặc biệt.
import java.util.concurrent.Semaphore;
public class ResourceLimiter {
private final Semaphore semaphore =
new Semaphore(5);
public void execute() throws InterruptedException {
semaphore.acquire();
try {
useResource();
} finally {
semaphore.release();
}
}
private void useResource() {
// Sử dụng tài nguyên
}
}Trong ví dụ này, nhiều nhất năm Thread có thể giữ permit cùng lúc.
Semaphore thường phù hợp để giới hạn số lượng request đồng thời, truy cập vào tài nguyên có chi phí cao hoặc giới hạn mức độ concurrency đối với một dependency.
Khác với mutex thông thường, Semaphore không ràng buộc permit với một Thread sở hữu cụ thể. Vì vậy, cần bảo đảm rằng mỗi permit được trả lại đúng số lần.
3.4.4. So sánh ba công cụ
Bài toán
│
├── Chờ N sự kiện hoàn thành
│ └── CountDownLatch
│
├── N Thread gặp nhau tại cùng một điểm
│ └── CyclicBarrier
│
└── Giới hạn N thao tác đồng thời
└── Semaphore
3.5. BlockingQueue và mô hình Producer-Consumer
Producer-Consumer là một mô hình kinh điển trong lập trình đồng thời.
Producer tạo task hoặc dữ liệu. Consumer lấy dữ liệu từ hàng đợi và xử lý. Hàng đợi là điểm giao tiếp giữa hai phía.
Producer 1 ──┐
│
Producer 2 ──┼──► BlockingQueue ───► Consumer 1
│ └──► Consumer 2
Producer 3 ──┘
Producer: tạo và đưa task vào queue
Consumer: lấy task và xử lý
Queue có giới hạn:
giúp kiểm soát lượng công việc chờ xử lý.
BlockingQueue cung cấp những thao tác an toàn khi nhiều Thread cùng sử dụng. Với các method blocking, Thread có thể chờ khi queue rỗng hoặc đã đầy.
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class ProducerConsumerExample {
private static final BlockingQueue<String> QUEUE =
new ArrayBlockingQueue<>(100);
public static void main(String[] args) {
Thread producer = new Thread(() -> {
try {
for (int i = 0; i < 1000; i++) {
QUEUE.put("Task-" + i);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread consumer = new Thread(() -> {
try {
while (!Thread.currentThread().isInterrupted()) {
String task = QUEUE.take();
System.out.println("Processing " + task);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
consumer.start();
producer.start();
}
}put() chờ khi queue đầy. take() chờ khi queue rỗng.
Ví dụ này minh họa cơ chế trao đổi dữ liệu; trong một ứng dụng cần shutdown có kiểm soát, mình phải thiết kế thêm giao thức kết thúc consumer, chẳng hạn poison pill hoặc cơ chế cancellation thích hợp.
3.6. Concurrent Collections
Khi nhiều Thread cùng truy cập collection, việc sử dụng HashMap, ArrayList hoặc HashSet mà không có cơ chế bảo vệ có thể gây ra lỗi.
Java cung cấp nhiều cấu trúc dữ liệu chuyên dụng:
ConcurrentHashMap: Map hỗ trợ truy cập đồng thời.
CopyOnWriteArrayList: Phù hợp với tình huống đọc nhiều, ghi ít.
BlockingQueue: Hàng đợi hỗ trợ phối hợp giữa Producer và Consumer.
ConcurrentLinkedQueue: Queue không blocking được thiết kế cho truy cập đồng thời.
ConcurrentSkipListMap: Map có thứ tự hỗ trợ thao tác đồng thời.
3.6.1. ConcurrentHashMap
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class CacheExample {
private final ConcurrentMap<String, String> cache =
new ConcurrentHashMap<>();
public void put(String key, String value) {
cache.put(key, value);
}
public String get(String key) {
return cache.get(key);
}
public String getOrLoad(String key) {
return cache.computeIfAbsent(
key,
this::loadValue
);
}
private String loadValue(String key) {
return "Value for " + key;
}
}ConcurrentHashMap hỗ trợ các thao tác nguyên tử như putIfAbsent(), computeIfAbsent() và replace() theo hợp đồng của API.
Tuy nhiên, việc sử dụng ConcurrentHashMap không tự động làm cho mọi thao tác trên nhiều key hoặc nhiều object khác nhau trở thành một giao dịch nguyên tử duy nhất.
Nếu cần duy trì một bất biến liên quan đến nhiều thao tác, mình vẫn cần thiết kế cơ chế đồng bộ hóa hoặc giao dịch phù hợp.
3.6.2. CopyOnWriteArrayList
CopyOnWriteArrayList tạo bản sao của mảng nội bộ khi thực hiện những thay đổi. Cách tiếp cận này có thể hữu ích khi số lần đọc vượt xa số lần ghi và việc lặp qua collection cần hạn chế sự can thiệp của những lần ghi đồng thời.
Đổi lại, các thao tác ghi có thể tốn bộ nhớ và thời gian đáng kể khi collection lớn.
3.7. Fork/Join Framework và Parallel Streams
Fork/Join Framework phù hợp với những bài toán có thể chia nhỏ thành các công việc con, sau đó tổng hợp kết quả.
Mô hình cơ bản:
Main Task
│
┌─────┴─────┐
▼ ▼
Task A Task B
/ \ / \
▼ ▼ ▼ ▼
A1 A2 B1 B2
\ / \ /
Combine Combine
\ /
Result
Một cơ chế quan trọng là work-stealing: worker có thể lấy công việc từ hàng đợi của worker khác khi bản thân không còn đủ công việc để thực hiện.
Ví dụ sử dụng RecursiveTask để tính tổng mảng:
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class SumTask extends RecursiveTask<Long> {
private final int[] numbers;
private final int start;
private final int end;
private static final int THRESHOLD = 1000;
public SumTask(int[] numbers, int start, int end) {
this.numbers = numbers;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int length = end - start;
if (length <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += numbers[i];
}
return sum;
}
int middle = start + length / 2;
SumTask left =
new SumTask(numbers, start, middle);
SumTask right =
new SumTask(numbers, middle, end);
left.fork();
long rightResult = right.compute();
long leftResult = left.join();
return leftResult + rightResult;
}
public static void main(String[] args) {
int[] numbers = new int[1_000_000];
ForkJoinPool pool = new ForkJoinPool();
try {
long result = pool.invoke(
new SumTask(numbers, 0, numbers.length)
);
System.out.println(result);
} finally {
pool.shutdown();
}
}
}Trong ví dụ, mỗi task chia mảng thành các phần nhỏ cho đến khi kích thước đủ nhỏ để xử lý tuần tự.
Fork/Join thường hữu ích cho CPU-bound tasks có khả năng chia nhỏ. Nó không tự động là lựa chọn tốt nhất cho mọi công việc I/O-bound hoặc những tác vụ thường xuyên blocking.
3.7.1. Parallel Streams
Java Streams hỗ trợ xử lý song song thông qua parallelStream() hoặc parallel().
import java.util.List;
List<Integer> numbers =
List.of(1, 2, 3, 4, 5, 6);
int sum = numbers.parallelStream()
.mapToInt(Integer::intValue)
.sum();
System.out.println(sum);Parallel Streams có thể hữu ích với những phép biến đổi dữ liệu độc lập, nhưng không bảo đảm rằng chương trình sẽ nhanh hơn trong mọi trường hợp.
Không nên đưa side effect lên shared mutable state một cách tùy tiện trong các pipeline song song.
List<Integer> result = new ArrayList<>();
numbers.parallelStream()
.forEach(result::add);Ở ví dụ trên, nhiều Thread có thể đồng thời gọi add() trên cùng một ArrayList. Điều đó tạo ra nguy cơ Race Condition.
Nên ưu tiên các thao tác reduction hoặc collector phù hợp:
List<Integer> result =
numbers.parallelStream()
.filter(n -> n % 2 == 0)
.toList();Parallel Streams có thể sử dụng common ForkJoinPool. Vì vậy, cần xem xét sự tương tác với các công việc khác đang sử dụng pool này, đặc biệt trong ứng dụng server.
4. Virtual Threads trong Java 21+
Virtual Threads là một thay đổi quan trọng trong mô hình xử lý đồng thời của Java. Chúng được thiết kế để giúp ứng dụng có thể duy trì số lượng lớn tác vụ đồng thời, đặc biệt khi các tác vụ dành phần lớn thời gian chờ I/O.
Virtual Threads được phát triển trong Project Loom và trở thành tính năng chính thức từ Java 21.
Điểm cần nhớ là Virtual Threads không thay thế hoàn toàn Platform Threads. Chúng thay đổi cách biểu diễn và lập lịch những đơn vị thực thi trong Java, giúp mô hình lập trình blocking truyền thống có thể mở rộng tốt hơn trong nhiều tình huống.
4.1. Platform Thread và Virtual Thread
Platform Thread là loại Thread Java truyền thống được ánh xạ với Thread của hệ điều hành theo mô hình thông dụng của JVM.
Virtual Thread là Thread được quản lý và lập lịch bởi Java runtime. Nhiều Virtual Thread có thể lần lượt thực thi trên một tập hợp Platform Thread đóng vai trò carrier.
Java Task A ──► Platform Thread A ──► OS Thread A Java Task B ──► Platform Thread B ──► OS Thread B Java Task C ──► Platform Thread C ──► OS Thread C Mỗi Platform Thread thường cần tài nguyên để duy trì một OS Thread tương ứng.
Virtual Thread A ──┐
Virtual Thread B ──┤
Virtual Thread C ──┼──► Carrier Threads
Virtual Thread D ──┤ │
Virtual Thread E ──┘ ▼
OS Threads
Nhiều Virtual Thread có thể được lập lịch
trên một số lượng Carrier Thread nhỏ hơn.
Platform Thread vẫn rất cần thiết. Chúng phù hợp với những tác vụ CPU-bound cần khai thác các CPU core, cũng như những trường hợp cần điều khiển tài nguyên thực thi theo cách truyền thống.
Virtual Threads giúp giảm đáng kể chi phí duy trì một đơn vị thực thi cho mỗi tác vụ chờ đợi, nhưng không làm cho CPU có thêm năng lực tính toán vật lý.
4.2. Carrier Thread, Parking và Scheduling
Để hiểu Virtual Threads, mình cần nắm ba khái niệm: Virtual Thread, Carrier Thread và cơ chế parking.
Virtual Thread: Thread Java đại diện cho một đơn vị công việc.
Carrier Thread: Platform Thread được sử dụng để thực thi Virtual Thread trong một khoảng thời gian.
Parking: Trạng thái Virtual Thread tạm ngừng thực thi để chờ một điều kiện hoặc sự kiện. Khi có thể tiếp tục, nó được unpark và lập lịch lại.
Virtual Thread starts
│
▼
Scheduled on Carrier Thread
│
▼
Execute Java code
│
▼
Reach supported blocking operation
│
▼
Virtual Thread parks
│
├──► Carrier Thread can execute other work
│
▼
Blocking condition becomes ready
│
▼
Virtual Thread is unparked
│
▼
Scheduled again on a Carrier Thread
│
▼
Continue execution
Với các thao tác blocking được runtime hỗ trợ, Virtual Thread có thể tạm ngừng mà không bắt buộc giữ một OS Thread chỉ để chờ. Carrier Thread sau đó có thể thực thi công việc khác.
Đây là một trong những lý do quan trọng giúp Virtual Threads cải thiện khả năng mở rộng cho những ứng dụng có nhiều request đồng thời.
4.2.1. Tạo Virtual Thread
Java 21 hỗ trợ những cách tạo Virtual Thread như sau:
Thread thread = Thread.ofVirtual()
.name("virtual-worker")
.start(() -> {
System.out.println(
Thread.currentThread()
.isVirtual()
);
});
try {
thread.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}Java cũng cung cấp Executors.newVirtualThreadPerTaskExecutor() để tạo một Executor khởi chạy một Virtual Thread cho mỗi task được chấp nhận.
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class VirtualThreadExample {
public static void main(String[] args) {
try (ExecutorService executor =
Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < 1000; i++) {
int taskId = i;
executor.submit(() -> {
process(taskId);
});
}
}
}
private static void process(int taskId) {
// Xử lý công việc của task
}
}Trong Java 21, ExecutorService triển khai AutoCloseable, nên có thể sử dụng try-with-resources để quản lý vòng đời. Khi khối try kết thúc, executor được shutdown và chờ những task đã được gửi hoàn thành theo hợp đồng của API.
4.3. Tại sao Virtual Threads thay đổi cuộc chơi cho I/O-bound?
Hãy xem xét một HTTP server nhận hàng nghìn request đồng thời. Mỗi request cần truy vấn database, gọi API bên ngoài và tổng hợp kết quả.
Nếu sử dụng mô hình Platform Thread truyền thống với một Thread cho mỗi request, số lượng Thread lớn có thể tiêu tốn nhiều tài nguyên ngay cả khi phần lớn thời gian chúng đang chờ I/O.
Với Virtual Threads, mình có thể giữ cách viết code blocking dễ đọc mà vẫn hỗ trợ số lượng lớn tác vụ chờ đợi đồng thời, trong những điều kiện mà các thao tác blocking được tích hợp tốt với runtime.
Request 1: Read request ↓ Call database ↓ Wait for database ↓ Build response Request 2: Read request ↓ Call external API ↓ Wait for API ↓ Build response Virtual Threads allow many request tasks to wait without reserving a carrier thread for every waiting task.
Điều này có thể cải thiện throughput và giảm nhu cầu duy trì một lượng lớn Platform Thread trong ứng dụng I/O-bound.
Tuy nhiên, số lượng request đồng thời vẫn cần được giới hạn theo tài nguyên thực tế như database connection pool, API rate limit, bộ nhớ và dung lượng queue.
4.3.1. Ví dụ xử lý I/O bằng Virtual Thread
import java.util.concurrent.*;
public class IoService {
public static void main(String[] args) {
try (ExecutorService executor =
Executors.newVirtualThreadPerTaskExecutor()) {
Future<String> userFuture =
executor.submit(
() -> fetchUser("U001")
);
Future<String> orderFuture =
executor.submit(
() -> fetchOrders("U001")
);
try {
String user = userFuture.get();
String orders = orderFuture.get();
System.out.println(user);
System.out.println(orders);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
throw new RuntimeException(
"Request processing failed",
e.getCause()
);
}
}
}
static String fetchUser(String id) {
return "User " + id;
}
static String fetchOrders(String id) {
return "Orders for " + id;
}
}Hai tác vụ được gửi độc lập và có thể tiến triển đồng thời. Ví dụ minh họa cơ chế thực thi; trong hệ thống thật, các method cần thực hiện I/O thực sự và được cấu hình timeout thích hợp.
4.4. Những giới hạn cần lưu ý khi dùng Virtual Threads
Virtual Threads không phải là một giải pháp tự động làm mọi chương trình nhanh hơn.
4.4.1. Không tạo thêm CPU core
Nếu công việc dành phần lớn thời gian để tính toán liên tục và không có thao tác blocking, Virtual Threads không tự động cải thiện đáng kể hiệu năng so với một số lượng Platform Thread phù hợp.
Đối với CPU-bound workloads, cần tập trung vào việc sử dụng CPU hiệu quả, chia nhỏ task phù hợp và tránh tạo ra quá nhiều công việc tranh chấp cùng tài nguyên.
4.4.2. Không nên pool Virtual Threads chỉ để tiết kiệm Thread
Virtual Threads được thiết kế để có chi phí thấp hơn đáng kể so với Platform Threads. Mô hình thông dụng là tạo một Virtual Thread cho mỗi task thay vì giữ một pool Virtual Thread nhỏ chỉ để tái sử dụng Thread.
Nếu cần giới hạn số lượng truy cập database hoặc số request gọi đến một API, hãy giới hạn tài nguyên hoặc thao tác tương ứng thay vì dùng pool Thread làm cơ chế giới hạn duy nhất.
4.4.3. Pinning
Trong một số trường hợp, Virtual Thread có thể bị pin vào Carrier Thread khi thực hiện một thao tác blocking. Tình trạng này làm giảm khả năng giải phóng carrier để xử lý công việc khác.
Với Java 21, một trường hợp cần lưu ý là blocking operation bên trong vùng synchronized trong những tình huống có thể dẫn đến pinning.
Java 24 đã thay đổi đáng kể hành vi pinning liên quan đến synchronized monitor thông qua các cải tiến của Project Loom. Vì vậy, khi tối ưu Virtual Threads, cần kiểm tra phiên bản JDK thực tế thay vì áp dụng máy móc các khuyến nghị dành cho Java 21.
Các hoạt động native hoặc những trường hợp phụ thuộc vào implementation vẫn cần được đánh giá riêng.
4.4.4. Giới hạn tài nguyên vẫn còn nguyên
Giả sử ứng dụng có thể tạo 50.000 Virtual Threads. Điều đó không có nghĩa database có thể xử lý 50.000 truy vấn đồng thời hoặc API bên ngoài chấp nhận 50.000 request mỗi giây.
Mình vẫn cần giới hạn các tài nguyên có dung lượng hữu hạn.
import java.util.concurrent.Semaphore;
public class DatabaseAccess {
private final Semaphore permits =
new Semaphore(20);
public String query() throws InterruptedException {
permits.acquire();
try {
return executeQuery();
} finally {
permits.release();
}
}
private String executeQuery() {
return "Database result";
}
}Semaphore trong ví dụ giới hạn tối đa 20 thao tác đi vào vùng sử dụng tài nguyên. Đây chỉ là ví dụ minh họa; ứng dụng thật nên đồng bộ giới hạn này với connection pool và chính sách của database client.
4.4.5. ThreadLocal và InheritableThreadLocal
ThreadLocal gắn dữ liệu với Thread. Khi chuyển từ Platform Thread sang Virtual Thread, cần xem xét cách sử dụng ThreadLocal vì số lượng Virtual Thread có thể rất lớn.
Không nên lưu trữ những object nặng trong ThreadLocal nếu chúng không cần thiết cho toàn bộ vòng đời Thread. Cần dọn dẹp dữ liệu khi phù hợp, đặc biệt khi có các cơ chế tái sử dụng Thread.
Ngoài ra, Virtual Thread không phải lúc nào cũng là lựa chọn đúng khi thư viện phụ thuộc mạnh vào ThreadLocal, native code hoặc những giả định đặc thù về Thread truyền thống. Cần kiểm tra khả năng tương thích của dependency.
Virtual Threads giúp giảm chi phí của việc duy trì nhiều tác vụ đồng thời, đặc biệt với I/O-bound. Chúng không loại bỏ nhu cầu Thread-Safety, timeout, backpressure, giới hạn tài nguyên hoặc xử lý lỗi.
5. Race Condition, Deadlock và Debugging thực tế
Trong các dự án Java thực tế, những lỗi Concurrency thường không xuất hiện rõ ràng như lỗi cú pháp. Chúng có thể chỉ xảy ra khi tải tăng cao, trong những thời điểm hiếm gặp hoặc khi nhiều sự kiện trùng nhau.
Vì vậy, lập trình đồng thời không chỉ là lựa chọn API. Nó còn đòi hỏi kỹ năng thiết kế, đo đạc, đọc Thread Dump và xây dựng quy trình debug có hệ thống.
5.1. Phân tích Race Condition
Race Condition có thể xảy ra khi nhiều Thread cùng sửa đổi trạng thái chia sẻ mà không có một cơ chế phối hợp thích hợp.
Ví dụ trong hệ thống đặt hàng, một sản phẩm chỉ còn một đơn vị hàng tồn kho.
public class Inventory {
private int stock = 1;
public boolean purchase() {
if (stock > 0) {
stock--;
return true;
}
return false;
}
}Nếu hai Thread cùng gọi purchase(), cả hai có thể cùng đọc thấy stock > 0 trước khi một Thread cập nhật kho.
Kết quả là hệ thống có thể chấp nhận hai giao dịch dù chỉ còn một sản phẩm.
5.1.1. Giải pháp với synchronized
public class Inventory {
private int stock = 1;
public synchronized boolean purchase() {
if (stock > 0) {
stock--;
return true;
}
return false;
}
}Toàn bộ thao tác kiểm tra và giảm tồn kho nằm trong cùng một critical section. Những Thread sử dụng method này trên cùng instance sẽ được phối hợp qua cùng monitor.
Tuy nhiên, nếu tồn kho được lưu trong database và có nhiều application instance, synchronized trên một object Java không đủ để bảo vệ giao dịch giữa các instance. Khi đó cần dùng cơ chế đồng bộ hóa phù hợp ở tầng database, chẳng hạn transaction và cập nhật có điều kiện.
5.2. Deadlock: Nguyên nhân và cách khắc phục
Deadlock xảy ra khi các Thread chờ tài nguyên của nhau theo một vòng phụ thuộc khiến không Thread nào trong vòng đó có thể tiếp tục.
Một ví dụ kinh điển là hai Thread lấy hai khóa theo thứ tự ngược nhau.
public class DeadlockExample {
private static final Object LOCK_A = new Object();
private static final Object LOCK_B = new Object();
public static void main(String[] args) {
Thread thread1 = new Thread(() -> {
synchronized (LOCK_A) {
sleep();
synchronized (LOCK_B) {
System.out.println("Thread 1 completed");
}
}
});
Thread thread2 = new Thread(() -> {
synchronized (LOCK_B) {
sleep();
synchronized (LOCK_A) {
System.out.println("Thread 2 completed");
}
}
});
thread1.start();
thread2.start();
}
private static void sleep() {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}Thread 1 lấy LOCK_A rồi chờ LOCK_B. Thread 2 lấy LOCK_B rồi chờ LOCK_A.
Thread 1
│
├── Holds LOCK_A
│
└── Waits for LOCK_B
▲
│
│ Held by
│
Thread 2
│
├── Holds LOCK_B
│
└── Waits for LOCK_A
Circular Wait → Deadlock
5.2.1. Bốn điều kiện kinh điển của Deadlock
Mutual Exclusion: Có tài nguyên chỉ cho phép một Thread sử dụng tại một thời điểm.
Hold and Wait: Thread đang giữ tài nguyên này trong khi chờ tài nguyên khác.
No Preemption: Tài nguyên không bị lấy khỏi Thread một cách tùy ý theo cơ chế thông thường của khóa.
Circular Wait: Có một vòng phụ thuộc, trong đó mỗi Thread chờ tài nguyên do Thread tiếp theo trong vòng đang giữ.
Để phòng tránh Deadlock, mình có thể thiết kế hệ thống sao cho không đồng thời tồn tại các điều kiện cần thiết này.
5.2.2. Giải pháp 1: Thống nhất thứ tự lấy khóa
Đây là một trong những cách đơn giản và hiệu quả nhất khi nhiều thao tác cần lấy nhiều khóa.
public class SafeLocking {
private static final Object LOCK_A = new Object();
private static final Object LOCK_B = new Object();
public static void operation1() {
synchronized (LOCK_A) {
synchronized (LOCK_B) {
// Thực hiện công việc
}
}
}
public static void operation2() {
synchronized (LOCK_A) {
synchronized (LOCK_B) {
// Thực hiện công việc khác
}
}
}
}Cả hai thao tác đều lấy LOCK_A trước LOCK_B. Điều này loại bỏ vòng phụ thuộc do hai thao tác lấy khóa theo thứ tự ngược nhau trong ví dụ.
Trong ứng dụng lớn, cần xây dựng quy ước thứ tự khóa nhất quán trên toàn bộ codebase, thay vì chỉ sửa riêng một method.
5.2.3. Giải pháp 2: tryLock() với timeout
Với các khóa tường minh, có thể sử dụng timeout để tránh chờ vô hạn trong một số tình huống.
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
public class TransferService {
private final ReentrantLock lockA =
new ReentrantLock();
private final ReentrantLock lockB =
new ReentrantLock();
public boolean transfer() throws InterruptedException {
if (!lockA.tryLock(1, TimeUnit.SECONDS)) {
return false;
}
try {
if (!lockB.tryLock(1, TimeUnit.SECONDS)) {
return false;
}
try {
// Thực hiện chuyển dữ liệu
return true;
} finally {
lockB.unlock();
}
} finally {
lockA.unlock();
}
}
}Timeout có thể giúp hệ thống thoát khỏi việc chờ khóa quá lâu, nhưng không tự động bảo đảm mọi thao tác nghiệp vụ đều thành công. Nếu một giao dịch phải thực hiện nhiều bước, cần thiết kế cách rollback hoặc retry sao cho không làm sai dữ liệu.
5.3. Thread Dump và công cụ chẩn đoán
Thread Dump là ảnh chụp trạng thái các Thread tại một thời điểm, bao gồm tên Thread, trạng thái, stack trace và một số thông tin liên quan đến monitor hoặc lock.
Đây là công cụ rất quan trọng khi ứng dụng có biểu hiện:
- Request bị treo hoặc phản hồi chậm bất thường. - CPU sử dụng cao nhưng throughput thấp. - Nhiều Thread ở trạng thái BLOCKED. - Thread Pool không còn worker rảnh. - Ứng dụng không thể shutdown.
5.3.1. Lấy Thread Dump
Với ứng dụng Java đang chạy trên máy có JDK, mình có thể sử dụng những công cụ như jcmd hoặc jstack.
jcmd <pid> Thread.print
Hoặc:
jstack <pid>
Trong đó, pid là Process ID của ứng dụng Java cần kiểm tra.
Trên Linux, mình có thể kết hợp với những công cụ như top, ps hoặc top -H để quan sát mức sử dụng CPU của các Thread hệ điều hành.
5.3.2. Dấu hiệu Deadlock trong Thread Dump
Một Thread Dump có thể chứa thông tin tương tự như sau:
"Thread-1":
BLOCKED
waiting to lock LOCK_B
locked LOCK_A
"Thread-2":
BLOCKED
waiting to lock LOCK_A
locked LOCK_B
Found one Java-level deadlock
Đây là nội dung minh họa, không phải output thực tế của một lần chạy cụ thể.
Khi đọc Thread Dump, mình cần xác định:
1. Thread nào đang chờ khóa. 2. Khóa đó đang được Thread nào giữ. 3. Có vòng phụ thuộc giữa các Thread hay không. 4. Các Thread có cùng xuất phát từ một nhóm request hoặc một đoạn code không. 5. Có thể giảm phạm vi critical section hoặc thống nhất thứ tự khóa hay không.
5.3.3. BLOCKED không phải lúc nào cũng là Deadlock
Một Thread ở trạng thái BLOCKED có thể chỉ đang chờ một Thread khác hoàn thành critical section ngắn. Đây không nhất thiết là lỗi.
Điều cần quan tâm là mức độ tranh chấp, thời gian chờ, số lượng Thread bị ảnh hưởng và liệu các Thread có còn tiến triển hay không.
Tương tự, một Thread ở trạng thái WAITING hoặc TIMED_WAITING cũng không mặc định là có vấn đề. Thread có thể đang chờ một công việc hợp lệ hoàn thành.
5.3.4. Dùng JFR và Java Mission Control
Java Flight Recorder (JFR) có thể ghi lại các sự kiện runtime với overhead tương đối thấp trong nhiều cấu hình sử dụng thông thường. Java Mission Control (JMC) hỗ trợ phân tích dữ liệu ghi được.
Những thông tin hữu ích khi điều tra Concurrency bao gồm:
- Thread scheduling và thời gian hoạt động. - Monitor contention. - Thread park và các sự kiện liên quan. - CPU usage và allocation. - Các sự kiện I/O và độ trễ liên quan nếu được ghi nhận.
Đối với Virtual Threads, những công cụ tương thích với phiên bản JDK đang sử dụng sẽ giúp quan sát số lượng Thread, thời gian chờ và các vấn đề về pinning hoặc scheduling khi có thể.
5.4. Starvation, Livelock và Thread Pool Exhaustion
5.4.1. Starvation
Starvation xảy ra khi một Thread liên tục không có cơ hội nhận tài nguyên cần thiết để tiến triển, dù các Thread khác vẫn tiếp tục chạy.
Nguyên nhân có thể là ưu tiên tài nguyên không hợp lý, lock contention, công việc dài giữ tài nguyên hoặc thiết kế pool khiến một nhóm task bị bỏ đói.
Để hạn chế, mình cần kiểm tra fairness khi phù hợp, giảm thời gian giữ khóa, tránh thiết kế phụ thuộc vòng và theo dõi thời gian chờ thực tế.
5.4.2. Livelock
Livelock xảy ra khi các Thread vẫn hoạt động nhưng liên tục phản ứng với nhau khiến công việc không tiến triển.
Ví dụ, hai Thread cùng phát hiện xung đột rồi cùng nhường tài nguyên, sau đó cùng thử lại vào một thời điểm tương tự. Nếu hành vi này lặp lại, cả hai đều hoạt động nhưng không hoàn thành công việc.
Một cách giảm nguy cơ là sử dụng chiến lược retry có giới hạn và backoff phù hợp, chẳng hạn thêm độ trễ ngẫu nhiên khi thực hiện retry các request cạnh tranh.
5.4.3. Thread Pool Exhaustion
Thread Pool Exhaustion xảy ra khi pool không còn worker có thể xử lý những task cần thực thi theo thời gian mong muốn.
Một ví dụ điển hình là tất cả worker của pool cha đều chờ các task con được gửi vào chính pool đó, trong khi không còn worker nào rảnh để thực thi task con.
Pool has 2 workers
│
├── Worker 1: waits for Child Task A
│
└── Worker 2: waits for Child Task B
Child Task A and B are queued
│
▼
No worker is available
│
▼
Tasks cannot make progress
Đây là một dạng starvation có thể xuất hiện trong những thiết kế nested task không phù hợp.
Giải pháp có thể bao gồm không chặn worker chờ task con trong cùng một pool, tách pool theo đặc điểm công việc, sử dụng cấu trúc bất đồng bộ hoặc thiết kế lại dependency giữa các task.
5.4.4. Blocking I/O trong CPU Thread Pool
Nếu một pool được thiết kế cho CPU-bound task nhưng lại chứa nhiều thao tác I/O blocking, tất cả worker có thể bị giữ trong lúc chờ I/O. Các task tính toán khác vì thế không còn cơ hội chạy.
Cần phân biệt pool phục vụ công việc CPU với cơ chế thực thi tác vụ I/O. Trong Java 21+, Virtual Threads là một lựa chọn đáng cân nhắc cho nhiều tác vụ I/O blocking, trong khi CPU-bound tasks vẫn cần được giới hạn theo năng lực xử lý.
6. Best Practices và tư duy thiết kế Concurrency
Sau khi tìm hiểu từ Thread Lifecycle đến JMM, Locks, Atomic Variables, ExecutorService và Virtual Threads, mình cần tổng hợp lại những nguyên tắc quan trọng khi xây dựng ứng dụng Java thực tế.
6.1. Ưu tiên giảm chia sẻ trạng thái có thể thay đổi
Phòng tránh Race Condition từ thiết kế thường đơn giản hơn việc sửa lỗi sau này.
Nếu hai task không cần chia sẻ trạng thái có thể thay đổi, hãy để mỗi task làm việc với dữ liệu riêng và chỉ kết hợp kết quả khi cần.
Những chiến lược hữu ích bao gồm:
- Sử dụng object bất biến. - Truyền dữ liệu qua message hoặc queue. - Giữ state trong phạm vi một task khi có thể. - Chỉ sử dụng khóa tại những nơi thực sự cần bảo vệ bất biến.
6.2. Không sử dụng Thread-Safe Collection thay cho Transaction
Một collection có khả năng truy cập đồng thời không đồng nghĩa với việc nhiều thao tác liên tiếp trên collection đó trở thành một transaction nguyên tử.
Ví dụ:
if (!map.containsKey(key)) {
map.put(key, value);
}Nếu nhiều Thread chạy đoạn code trên, chúng có thể cùng kiểm tra thấy key chưa tồn tại rồi cùng thực hiện cập nhật.
Nếu mục tiêu là cập nhật theo điều kiện một cách nguyên tử, hãy sử dụng API phù hợp:
map.putIfAbsent(key, value);Hoặc sử dụng computeIfAbsent() khi cần tính toán giá trị nếu key chưa có, đồng thời tuân thủ các yêu cầu và giới hạn của API callback.
6.3. Không giữ Lock trong thời gian quá dài
Critical section càng dài thì khả năng Thread khác phải chờ càng cao.
Nên tránh giữ khóa trong khi:
- Gọi HTTP API không kiểm soát được thời gian phản hồi. - Chờ thao tác database kéo dài. - Thực hiện công việc tính toán nặng không cần bảo vệ. - Gọi callback từ code bên ngoài có thể thực hiện hành vi không dự đoán được.
Điều này không có nghĩa là tuyệt đối không được thực hiện I/O trong vùng khóa. Một số bài toán yêu cầu đồng bộ hóa xuyên suốt thao tác. Tuy nhiên, quyết định đó cần dựa trên tính đúng đắn và được đánh giá kỹ về độ trễ cũng như nguy cơ deadlock.
6.4. Luôn thiết kế cơ chế Cancellation
Một task không nên giả định rằng nó sẽ được chạy đến cùng trong mọi tình huống.
Ứng dụng có thể cần dừng task do timeout, shutdown, người dùng hủy request hoặc một thành phần khác gặp lỗi.
Trong Java, interruption là một cơ chế quan trọng để thông báo yêu cầu dừng.
public void process() {
while (!Thread.currentThread().isInterrupted()) {
doOneUnitOfWork();
}
}Với những method ném InterruptedException, nếu không thể truyền tiếp ngoại lệ theo chữ ký method, mình thường khôi phục trạng thái interrupt sau khi xử lý phù hợp.
try {
blockingOperation();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}Không nên bắt InterruptedException rồi bỏ qua một cách tùy tiện, vì điều đó có thể khiến hệ thống không thể dừng task theo yêu cầu.
6.5. Đặt Timeout ở các ranh giới hệ thống
Timeout giúp ngăn một request chờ vô hạn khi dependency gặp sự cố.
Trong hệ thống thực tế, cần xem xét timeout ở nhiều tầng:
- HTTP client connection timeout. - HTTP response hoặc request timeout. - Database query timeout. - Future waiting timeout. - Hạn xử lý tổng thể của request.
Timeout ở một tầng không nhất thiết tự động hủy công việc ở những tầng còn lại. Cần xác định rõ việc timeout chỉ ngừng chờ kết quả hay thực sự yêu cầu dependency hủy thao tác.
6.6. Không retry vô hạn
Retry có thể giúp xử lý lỗi tạm thời, nhưng retry không giới hạn dễ làm hệ thống quá tải hơn khi dependency đang gặp sự cố.
Một chính sách retry tốt thường có:
- Số lần retry tối đa. - Backoff giữa các lần thử. - Jitter khi phù hợp để giảm việc nhiều client retry cùng lúc. - Phân biệt lỗi có thể retry và lỗi không nên retry. - Kiểm soát tính idempotent của thao tác.
Đối với những thao tác ghi dữ liệu hoặc thanh toán, retry cần được thiết kế cùng cơ chế chống xử lý trùng và tính nhất quán của giao dịch.
6.7. Phân biệt giới hạn Thread và giới hạn tài nguyên
Trong kiến trúc server, những tài nguyên sau có thể là nút thắt thực sự:
- Database connection pool. - Số kết nối HTTP đồng thời. - CPU và Heap memory. - API rate limit. - Hàng đợi và dung lượng buffer. - Khả năng ghi của storage hoặc message broker.
Tăng số Thread không giải quyết được giới hạn ở những tài nguyên này. Thậm chí, việc tăng concurrency quá mức còn làm độ trễ và tỷ lệ lỗi tăng lên.
6.8. Không tối ưu Concurrency bằng cảm tính
Khi ứng dụng chậm, cần đo đạc trước khi quyết định thay đổi kiến trúc.
Một số chỉ số hữu ích:
Throughput: Số lượng công việc hoàn thành trong một đơn vị thời gian. Latency: Thời gian phản hồi của một request. Queue depth: Số lượng task đang chờ xử lý. Pool utilization: Mức độ sử dụng worker của pool. Lock contention: Mức độ tranh chấp khóa. CPU usage: Mức độ sử dụng CPU. Allocation và GC: Lượng cấp phát bộ nhớ và chi phí Garbage Collection.
Không nên chỉ nhìn vào CPU usage. Một ứng dụng có CPU usage thấp vẫn có thể bị chậm do tất cả request đang chờ database hoặc bị kẹt trong một queue.
6.9. Lựa chọn công cụ theo bài toán
Cần bảo vệ một trạng thái chia sẻ?
│
├── Critical section thông thường
│ └── synchronized
│
├── Cần khóa với timeout / interrupt
│ └── ReentrantLock
│
├── Cần cập nhật một biến nguyên tử
│ └── AtomicInteger / AtomicReference
│
├── Cần giới hạn tài nguyên đồng thời
│ └── Semaphore
│
├── Cần phối hợp các giai đoạn
│ └── CountDownLatch / CyclicBarrier
│
├── Cần xử lý task và quản lý vòng đời
│ └── ExecutorService
│
├── Cần pipeline bất đồng bộ
│ └── CompletableFuture
│
├── Cần xử lý dữ liệu chia nhỏ CPU-bound
│ └── ForkJoin / Parallel Streams
│
└── Nhiều tác vụ I/O blocking
└── Virtual Threads (Java 21+)
Sơ đồ trên là hướng lựa chọn ban đầu. Một hệ thống có thể cần phối hợp nhiều công cụ với nhau, và không có một API duy nhất giải quyết mọi vấn đề Concurrency.
7. Bài tập thực hành và tổng kết
Để thực sự nắm vững Concurrency, mình nên tự viết các chương trình nhỏ, quan sát kết quả và kiểm chứng giả thuyết thay vì chỉ đọc lý thuyết.
7.1. Bài tập 1: Race Condition với Counter
Yêu cầu: Tạo mười Thread, mỗi Thread tăng bộ đếm 100.000 lần.
Mục tiêu: Quan sát sự khác biệt giữa kết quả mong đợi và kết quả thực tế khi sử dụng int, synchronized và AtomicInteger.
Hãy thực hiện ba phiên bản:
1. Bộ đếm int không đồng bộ hóa.
2. Bộ đếm được bảo vệ bằng synchronized.
3. Bộ đếm sử dụng AtomicInteger.
Ghi lại kết quả nhiều lần chạy. Sau đó giải thích vì sao phiên bản thứ nhất không có bảo đảm tính đúng đắn, còn hai phiên bản còn lại có cơ chế đảm bảo phép tăng được phối hợp đúng.
7.2. Bài tập 2: Producer-Consumer
Yêu cầu: Tạo hai Producer và ba Consumer, sử dụng một BlockingQueue có dung lượng giới hạn.
Mục tiêu: Quan sát cách queue điều tiết tốc độ giữa hai phía.
Hãy thử những tình huống:
- Producer nhanh hơn Consumer. - Consumer nhanh hơn Producer. - Queue có dung lượng nhỏ. - Một Consumer bị interrupt trong khi chờ. - Ứng dụng cần shutdown trong khi queue còn task.
Giải thích vì sao queue hữu hạn giúp kiểm soát lượng công việc chờ xử lý và vì sao cần một giao thức dừng Consumer rõ ràng.
7.3. Bài tập 3: Tái hiện và khắc phục Deadlock
Yêu cầu: Viết chương trình có hai khóa và hai Thread lấy khóa theo thứ tự ngược nhau.
Mục tiêu: Lấy Thread Dump và xác định vòng phụ thuộc.
Sau đó, sửa chương trình bằng cách thống nhất thứ tự khóa. So sánh Thread Dump trước và sau khi sửa.
7.4. Bài tập 4: CompletableFuture Pipeline
Yêu cầu: Xây dựng một pipeline gồm ba bước độc lập hoặc phụ thuộc:
1. Tải thông tin người dùng. 2. Tải đơn hàng. 3. Tổng hợp kết quả và xử lý ngoại lệ.
Mục tiêu: Thực hành thenApply(), thenCompose(), thenCombine(), allOf() và các API xử lý lỗi.
Hãy phân biệt rõ hai tình huống: bước sau phụ thuộc kết quả bước trước và hai bước có thể chạy độc lập.
7.5. Bài tập 5: So sánh Platform Threads và Virtual Threads
Yêu cầu: Viết hai chương trình thực hiện cùng một workload I/O mô phỏng, một chương trình sử dụng Platform Threads và một chương trình sử dụng Virtual Threads.
Mục tiêu: So sánh throughput, latency, mức sử dụng CPU và lượng tài nguyên khi tăng số lượng task đồng thời.
Để kết quả có ý nghĩa, cần bảo đảm hai chương trình thực hiện cùng một công việc, có giới hạn tài nguyên tương đương và sử dụng cách đo đạc phù hợp.
Không nên kết luận Virtual Threads luôn nhanh hơn chỉ dựa trên một benchmark đơn giản. Kết quả phụ thuộc vào loại workload, thư viện I/O, giới hạn dependency và môi trường thực thi.
7.6. Checklist kiến thức cần nắm vững
Đồng bộ hóa - [ ] Sử dụng synchronized đúng cách. - [ ] Phân biệt synchronized và volatile. - [ ] Sử dụng ReentrantLock, Condition và tryLock. - [ ] Hiểu CAS, AtomicInteger và LongAdder.
Concurrency Utilities - [ ] Thiết kế Thread Pool với queue và rejection policy phù hợp. - [ ] Quản lý vòng đời ExecutorService. - [ ] Xây dựng pipeline với CompletableFuture. - [ ] Phân biệt CountDownLatch, CyclicBarrier và Semaphore. - [ ] Sử dụng Concurrent Collections đúng ngữ nghĩa.
Kiến trúc và vận hành - [ ] Phân biệt CPU-bound và I/O-bound. - [ ] Hiểu Carrier Thread và Virtual Thread. - [ ] Phát hiện Race Condition và Deadlock. - [ ] Đọc Thread Dump và sử dụng công cụ profiling. - [ ] Thiết kế timeout, cancellation, backpressure và shutdown.
7.7. Tổng kết
Java Concurrency không chỉ là việc tạo nhiều Thread để tăng tốc chương trình. Đây là lĩnh vực nghiên cứu cách nhiều đơn vị thực thi phối hợp với nhau, cùng truy cập dữ liệu và sử dụng tài nguyên mà vẫn duy trì tính đúng đắn của ứng dụng.
Để làm chủ lĩnh vực này, mình cần hiểu rõ mối quan hệ giữa ba tầng:
APPLICATION DESIGN
│
Task / Dependency / State
│
▼
JAVA CONCURRENCY API
│
Executor / Lock / Atomic / Future
│
▼
JAVA MEMORY MODEL (JMM)
│
Visibility / Ordering / Atomicity
│
▼
JVM + OPERATING SYSTEM
│
Scheduling / Synchronization / I/O
│
▼
HARDWARE
CPU / Memory / I/O
Ở tầng thiết kế ứng dụng, mình xác định task nào độc lập, dữ liệu nào được chia sẻ và tài nguyên nào cần giới hạn.
Ở tầng API, mình lựa chọn cơ chế thực thi và đồng bộ hóa phù hợp như ExecutorService, synchronized, Atomic Variables, CompletableFuture hoặc Virtual Threads.
Ở tầng bộ nhớ và runtime, mình cần hiểu JMM, Happens-Before, cách JVM lập lịch và những giới hạn thực tế của phần cứng.
Khi đã nắm vững những nền tảng này, mình có thể tiến tới các chủ đề nâng cao hơn như Lock-Free Algorithms, Memory Ordering, Structured Concurrency, Reactive Programming, Backpressure, Distributed Concurrency Control và các chiến lược thiết kế hệ thống có khả năng chịu tải cao.
Đích đến không phải là sử dụng càng nhiều Thread càng tốt, mà là xây dựng một hệ thống có thể xử lý nhiều công việc đồng thời một cách đúng đắn, có kiểm soát, dễ quan sát và có khả năng mở rộng.