1. Giới thiệu
Khi các luồng mỗi bên chơi phần của riêng mình
Lập trình đa luồng thông thường thường giống như buổi tập không có nhạc trưởng. Mỗi luồng — như một nhạc công chơi giai điệu của mình mà không nghe người khác. Có người xong sớm và đi hút thuốc, có người kẹt ở một hợp âm, có người thậm chí đánh nhầm nốt và gây lỗi. Kết quả không phải là bản giao hưởng mà là sự hỗn tạp: gần như không thể hiểu ai sai ở đâu, và dừng tất cả cùng lúc — đúng là một nhiệm vụ khó.
Structured Concurrency giải quyết vấn đề này. Nó biến các luồng rời rạc thành một dàn nhạc thật sự: mọi tác vụ đều nằm dưới một “nhạc trưởng”. Nếu nhạc trưởng ra lệnh dừng — cả dàn nhạc im lặng. Nếu một nhạc công mắc lỗi — những người còn lại dừng lại một cách gọn gàng, không phá vỡ tổng thể. Tất cả kết quả và lỗi được thu thập tập trung, không bị rải rác khắp nơi trong mã.
Hãy tưởng tượng: Bạn không thả các nhạc công chơi mỗi người một kiểu, mà tập hợp họ vào một phòng. Có nhạc trưởng, có bản tổng phổ, và ngay cả khi kèn trumpet lạc tông — dàn nhạc không vỡ trận mà kết thúc buổi diễn một cách đẹp đẽ.
Structured Concurrency mang lại gì
- Phạm vi tác vụ thống nhất: mọi tác vụ con sống trong một khối mã duy nhất, vòng đời của chúng bị giới hạn bởi khối này.
- Kết thúc có thể dự đoán: luồng cha sẽ không kết thúc cho đến khi tất cả tác vụ con kết thúc.
- Hủy bỏ tập trung: nếu một tác vụ thất bại hoặc luồng cha quyết định kết thúc — tất cả tác vụ con được hủy một cách đúng đắn.
- Xử lý lỗi nhất quán: lỗi của các tác vụ con được gom lại, có thể nhận được “cây nguyên nhân” (tree of causes).
- Mã sạch và dễ đọc: không có luồng “treo”, không có tác vụ bị quên, không tranh chấp khi hủy.
Structured Concurrency — không chỉ là API mới mà còn là một cách tư duy mới: các tác vụ cần được cấu trúc giống như các khối mã thông thường (ví dụ try-with-resources).
2. Tình trạng Structured Concurrency trong Java
Tại thời điểm viết khóa học, Structured Concurrency ở trạng thái Preview (Java 21–23), nhưng dự kiến sẽ tiến tới GA (General Availability) trong Java 24/25. API nằm trong package jdk.incubator.concurrent. Trước khi dùng trong production, hãy nhớ kiểm tra release notes hiện hành của phiên bản JDK của bạn!
Các lớp chính:
- StructuredTaskScope — lớp cơ sở để quản lý một nhóm tác vụ.
- Biến thể: StructuredTaskScope.ShutdownOnFailure, StructuredTaskScope.ShutdownOnSuccess — các chính sách kết thúc tác vụ.
Các khái niệm chính của StructuredTaskScope
Mô hình: fork, join và tổng duyệt kết quả
Khi nhạc trưởng (tức là tác vụ cha) ra hiệu — các tác vụ con tỏa ra chơi phần của mình. Khoảnh khắc này gọi là fork — như thể bạn cho các nhạc công chơi các đoạn ở những phòng khác nhau.
Sau đó đến lúc join — nhạc trưởng giơ đũa lên, và mọi người quay lại để chơi hợp âm cuối cùng cùng nhau.
Rồi bạn có thể hỏi mỗi thành viên xem mọi thứ diễn ra thế nào:
- thông qua resultNow() để lấy kết quả ngay nếu mọi thứ chạy xong không lỗi;
- thông qua throwIfFailed() — đảm bảo không ai lạc nhịp. Nếu ai đó đã rối nốt, sẽ ném ra một ngoại lệ hợp nhất — như thể nhạc trưởng nói: “Chúng ta có sự cố trong dàn nhạc, bắt đầu lại”.
Chính sách kết thúc
Mỗi nhạc trưởng đều có quy tắc riêng về thời điểm dừng nhạc. Trong Structured Concurrency điều này được đặt bằng chính sách kết thúc:
- ShutdownOnFailure — nếu một nhạc công lạc nhịp, nhạc trưởng phất tay: “Dừng! Làm lại từ đầu”. Những người còn lại dừng ngay.
- ShutdownOnSuccess — ngược lại, khi có người chơi trọn vẹn hoàn hảo, nhạc trưởng hài lòng: “Thế là đủ, không cần tiếp tục, chúng ta đã có người chiến thắng”. Những người khác im lặng — chính sách “ai thành công đầu tiên”.
Làm việc với các luồng ảo
Mỗi tác vụ con StructuredTaskScope được chạy trong luồng ảo. Giống như bạn có một dàn nhạc mà mỗi nhạc công đều hiểu chuyện và nhanh nhạy, không đòi hỏi sân khấu. Bạn có thể tự tin tạo hàng trăm, hàng nghìn “nhạc công” như thế — chúng không phải thread nặng, mà gần như những nốt nhạc nhẹ, vang lên đúng lúc cần thiết.
3. Ví dụ: bộ tổng hợp các yêu cầu HTTP
Xem một bài toán thực tế: chúng ta có ba nguồn dữ liệu (ví dụ ba máy chủ khác nhau), và muốn nhận phản hồi hoặc từ tất cả (để tổng hợp), hoặc từ nguồn đầu tiên phản hồi thành công.
Phương án 1: “Tất cả phải thành công” (ShutdownOnFailure)
import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;
public class AggregatorAllSuccess {
public static void main(String[] args) throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<String> f1 = scope.fork(() -> fetchFromSource1());
Future<String> f2 = scope.fork(() -> fetchFromSource2());
Future<String> f3 = scope.fork(() -> fetchFromSource3());
scope.join(); // chờ tất cả tác vụ hoàn thành
scope.throwIfFailed(); // nếu có ít nhất một cái thất bại — ném ngoại lệ
// Tất cả tác vụ đều thành công — có thể tổng hợp kết quả
String result = f1.resultNow() + f2.resultNow() + f3.resultNow();
System.out.println("Kết quả tổng hợp: " + result);
}
}
static String fetchFromSource1() { /* ... */ return "A"; }
static String fetchFromSource2() { /* ... */ return "B"; }
static String fetchFromSource3() { /* ... */ return "C"; }
}
Điều gì xảy ra:
- Cả ba tác vụ được chạy song song (trong các luồng ảo).
- Nếu ít nhất một cái thất bại — những cái còn lại bị hủy, và ném ngoại lệ.
- Nếu tất cả đều thành công — có thể tổng hợp kết quả một cách an toàn.
Phương án 2: “Thành công theo kết quả hợp lệ đầu tiên” (ShutdownOnSuccess)
import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;
public class AggregatorFirstSuccess {
public static void main(String[] args) throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
Future<String> f1 = scope.fork(() -> fetchFromSource1());
Future<String> f2 = scope.fork(() -> fetchFromSource2());
Future<String> f3 = scope.fork(() -> fetchFromSource3());
scope.join(); // chờ tác vụ đầu tiên thành công
scope.throwIfFailed(); // nếu tất cả đều thất bại — ném ngoại lệ
String result = scope.result(); // kết quả của tác vụ thành công đầu tiên
System.out.println("Kết quả thành công đầu tiên: " + result);
}
}
static String fetchFromSource1() { /* ... */ return "A"; }
static String fetchFromSource2() { /* ... */ return "B"; }
static String fetchFromSource3() { /* ... */ return "C"; }
}
Điều gì xảy ra:
- Ngay khi một tác vụ hoàn thành thành công — các tác vụ còn lại bị hủy.
- Nếu tất cả đều thất bại — ném ngoại lệ.
4. Hủy tự động và suy giảm chức năng
StructuredTaskScope tự xử lý việc hủy các tác vụ còn lại nếu chính sách yêu cầu. Ví dụ, nếu một tác vụ thất bại (ShutdownOnFailure) hoặc một tác vụ hoàn thành thành công (ShutdownOnSuccess), các tác vụ còn lại nhận tín hiệu hủy (interrupt).
Ví dụ: kết thúc đúng đắn với timeout
import jdk.incubator.concurrent.StructuredTaskScope;
import java.time.Instant;
import java.util.concurrent.Future;
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<String> f1 = scope.fork(() -> fetchWithTimeout());
Future<String> f2 = scope.fork(() -> fetchWithTimeout());
scope.joinUntil(Instant.now().plusSeconds(2)); // chờ tối đa 2 giây
scope.throwIfFailed();
String result = f1.resultNow() + f2.resultNow();
System.out.println(result);
}
Nếu các tác vụ không hoàn thành trong 2 giây — sẽ ném ngoại lệ và tất cả tác vụ bị hủy.
5. Lỗi và xử lý ngoại lệ
Cách ngoại lệ của các tác vụ con được định tuyến vào scope
Đôi khi trong buổi diễn vẫn có người đánh sai nốt — StructuredTaskScope không giả vờ như không có gì xảy ra. Nó ghi lại cẩn thận ai đã chơi lệch, rồi gửi cho nhạc trưởng một báo cáo đầy đủ. Khi bạn gọi throwIfFailed(), nó ném ra một ngoại lệ hợp nhất — kiểu như báo cáo tổng hợp: “Đây là danh sách những ai hôm nay đánh sai nốt”. Nếu cần, bạn có thể mở “cây nguyên nhân” đó ra để xem ai làm hỏng cụ thể. Còn nếu muốn biết về một người chơi cụ thể — Future.exceptionNow() sẽ cho biết phần của người đó kết thúc thế nào.
Khi việc hủy không phải là thất bại
Điều quan trọng cần nhớ: hủy một tác vụ không phải lúc nào cũng là lỗi. Nếu nhạc trưởng nói “xong, buổi diễn kết thúc”, thì các nhạc công chỉ cất nhạc cụ — đó là cancelled, chứ không phải failed. Lỗi chỉ được tính khi ai đó thực sự chơi sai, và ngoại lệ đó sẽ xuất hiện trong bản tổng hợp chung.
Ví dụ: cây nguyên nhân
import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<String> f1 = scope.fork(() -> { throw new RuntimeException("Lỗi 1"); });
Future<String> f2 = scope.fork(() -> { throw new RuntimeException("Lỗi 2"); });
scope.join();
scope.throwIfFailed(); // sẽ ném ngoại lệ kèm cả hai nguyên nhân
} catch (Exception e) {
e.printStackTrace();
// Có thể lấy các suppressed exceptions qua e.getSuppressed()
}
6. So sánh với CompletableFuture
StructuredTaskScope và CompletableFuture — cả hai đều cho phép chạy các tác vụ song song, nhưng:
- StructuredTaskScope phù hợp khi các tác vụ có liên hệ logic và cần kết thúc/hủy cùng nhau (hệ phân cấp tác vụ).
- CompletableFuture phù hợp để phối hợp các tác vụ không phân cấp (ví dụ chuỗi biến đổi, kịch bản phản ứng).
Khi nào StructuredTaskScope giúp mã đơn giản hơn:
- Khi cần đảm bảo mọi tác vụ con đã kết thúc trước khi thoát khỏi khối.
- Khi cần hủy bỏ và xử lý lỗi tập trung.
- Khi quan trọng là không còn tác vụ “treo”.
Khi nào CompletableFuture tiện hơn:
- Khi các tác vụ không liên quan và có thể sống độc lập.
- Khi cần phối hợp phức tạp (thenCombine, thenCompose, v.v.).
7. Thực hành: bộ tổng hợp các yêu cầu HTTP
Bài toán: gửi yêu cầu tới 3 nguồn, nhận phản hồi thành công đầu tiên
import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;
public class HttpAggregator {
public static void main(String[] args) throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
Future<String> f1 = scope.fork(() -> httpRequest("https://api1.example.com"));
Future<String> f2 = scope.fork(() -> httpRequest("https://api2.example.com"));
Future<String> f3 = scope.fork(() -> httpRequest("https://api3.example.com"));
scope.join();
scope.throwIfFailed();
String result = scope.result();
System.out.println("Phản hồi thành công đầu tiên: " + result);
}
}
static String httpRequest(String url) throws Exception {
// Mô phỏng yêu cầu (có thể dùng HttpClient)
Thread.sleep((long) (Math.random() * 1000));
if (Math.random() < 0.3) throw new RuntimeException("Lỗi yêu cầu: " + url);
return "Phản hồi từ " + url;
}
}
Bài toán: nếu một tác vụ con thất bại — hủy đúng đắn phần còn lại
import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<String> f1 = scope.fork(() -> httpRequest("https://api1.example.com"));
Future<String> f2 = scope.fork(() -> httpRequest("https://api2.example.com"));
scope.join();
scope.throwIfFailed();
String result = f1.resultNow() + f2.resultNow();
System.out.println("Cả hai phản hồi: " + result);
} catch (Exception e) {
System.err.println("Lỗi ở một trong các tác vụ: " + e.getMessage());
}
8. Lỗi thường gặp khi làm việc với StructuredTaskScope
Lỗi số 1: quên gọi join() hoặc throwIfFailed().
Nếu không gọi join(), các tác vụ có thể chưa kịp kết thúc trước khi thoát khỏi khối. Nếu không gọi throwIfFailed(), lỗi của các tác vụ con sẽ bị bỏ qua.
Lỗi số 2: cố lấy kết quả trước khi tác vụ hoàn thành.
Gọi resultNow() trước khi tác vụ hoàn thành sẽ ném IllegalStateException. Hãy đợi tác vụ kết thúc qua join() trước.
Lỗi số 3: bỏ qua việc hủy.
Nếu một tác vụ đã bị hủy (ví dụ do chính sách của scope), đừng cố lấy kết quả của nó — sẽ có ngoại lệ.
Lỗi số 4: trộn lẫn các chính sách kết thúc khác nhau.
Không nên tự hủy các tác vụ thủ công bên trong scope — hãy dùng các chính sách ShutdownOnFailure hoặc ShutdownOnSuccess.
Lỗi số 5: chạy các tác vụ CPU-bound dài trong luồng ảo.
StructuredTaskScope mặc định dùng các luồng ảo — chúng lý tưởng cho tác vụ I/O-bound, nhưng không làm nhanh các phép tính nặng.
Lỗi số 6: quên đóng scope (không dùng try-with-resources).
StructuredTaskScope hiện thực AutoCloseable — luôn dùng try-with-resources để đảm bảo tất cả tác vụ được kết thúc.
GO TO FULL VERSION