Khi một request tách thành nhiều tác vụ chạy song song, ví dụ lấy thông tin khách hàng và tình trạng đơn hàng cùng lúc, ta thường dùng ExecutorService và Future. Cách này chạy được nhưng có một lỗ hổng khó thấy, và đó là lý do structured concurrency ra đời: nếu một tác vụ lỗi, những tác vụ còn lại vẫn chạy tiếp mà không ai cần kết quả, và khi hàm kết thúc, các thread đó vẫn còn sống. Structured concurrency (đồng thời có cấu trúc) sửa gốc vấn đề bằng một quy tắc đơn giản: mọi tác vụ con phải kết thúc trước khi khối code đã tạo ra chúng kết thúc. Bài này giải thích ý tưởng, chạy thật StructuredTaskScope và ScopedValue trên JDK 21, rồi đối chiếu với API đã đổi ở JDK 25 và 26.
Bài nối tiếp Virtual threads trong Java 21: khi nào thay thread pool: virtual threads làm thread trở nên rẻ, còn structured concurrency trả lời câu hỏi làm sao quản lý đúng vòng đời của vô số thread đó. Về nền tảng thread pool và CompletableFuture, xem Concurrency patterns: Producer-Consumer, thread pool và CompletableFuture.
Vấn đề: tác vụ con "mồ côi" khi dùng ExecutorService
Hãy bắt đầu bằng việc chứng minh lỗ hổng mà structured concurrency nhắm tới. Chương trình Demo1FutureLeak.java gọi hai tác vụ song song: findUser() mất 1 giây, fetchOrder() ném lỗi sau 100 ms. Mã chính chờ user.get() trước rồi mới order.get():
Future<String> user = ex.submit(Demo1FutureLeak::findUser);
Future<Integer> order = ex.submit(Demo1FutureLeak::fetchOrder);
String u = user.get(); // chờ user trước: 1 giây, dù order đã lỗi từ 100 ms
int o = order.get();Kết quả chạy thật trên JDK 21 (Apple M4, 16 GB):
[ 4 ms] fetchOrder bắt đầu
[ 4 ms] findUser bắt đầu
[ 115 ms] fetchOrder ném lỗi
[1015 ms] findUser xong (vẫn chạy, không ai cần kết quả)
[1016 ms] main handle() thất bại: java.lang.IllegalStateException: kho hàng không phản hồiLỗi xảy ra ở 115 ms nhưng handle() chỉ biết ở 1.016 ms, và suốt thời gian đó findUser tốn tài nguyên cho kết quả bị vứt đi. Đây chính là các điểm yếu mà JEP 453 liệt kê: vòng đời tác vụ con không gắn với tác vụ cha nên lỗi làm rò rỉ thread, việc ngắt (interrupt) không lan sang tác vụ con, một tác vụ chậm có thể trì hoãn việc xử lý lỗi của tác vụ khác, và thread dump không cho thấy quan hệ cha con. Muốn tránh mà không có structured concurrency, phải tự viết thêm cancel(true) trong từng finally, một việc rất dễ sót.
Structured concurrency: ý tưởng "go statement considered harmful"
Ý tưởng structured concurrency không riêng của Java.
Năm 2018, Nathaniel J. Smith viết bài Notes on structured concurrency, or: Go statement considered harmful, so sánh việc "bắn rồi quên" một tác vụ nền (câu lệnh go, tạo thread, đăng ký callback) với lệnh goto mà Dijkstra từng phê phán: luồng điều khiển rẽ nhánh ra ngoài, và một lời gọi hàm có thể ngầm sinh ra việc chạy nền sống lâu hơn chính hàm đó. Giải pháp của ông là "nursery": một khối mà tác vụ cha phải mở trước khi sinh con, và không thoát khỏi khối cho tới khi mọi con xong. Smith nêu bốn hệ quả: hàm trả về nghĩa là việc đã xong, lỗi của con được ném lại ở cha sau khi huỷ các con khác, huỷ bỏ được lan truyền có trật tự, và tài nguyên (file, kết nối) không bị đóng khi còn tác vụ nền dùng chúng.
Project Loom đưa structured concurrency vào Java: ở JDK 19 và 20 dưới dạng incubator (JEP 428, 437), từ JDK 21 là API preview (JEP 453). Điểm nối với virtual threads rất tự nhiên: vì virtual thread rẻ, mỗi tác vụ con có thể có thread riêng, và StructuredTaskScope mặc định fork mỗi subtask vào một virtual thread mới.
Structured concurrency trong Java 21: StructuredTaskScope và ShutdownOnFailure
Cùng kịch bản trên, viết lại theo structured concurrency với StructuredTaskScope.ShutdownOnFailure (code/s6-32/Demo2Scope.java):
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Subtask<String> user = scope.fork(Demo2Scope::findUser);
Subtask<Integer> order = scope.fork(Demo2Scope::fetchOrder);
scope.join().throwIfFailed();
log("kết quả " + user.get() + order.get());
} catch (ExecutionException e) {
log("handle() thất bại: " + e.getCause());
}Vì đây là API preview, phải bật cờ khi biên dịch và chạy: java --enable-preview --source 21 Demo2Scope.java. Không có cờ, trình biên dịch từ chối:
Demo2Scope.java:2: error: StructuredTaskScope is a preview API and is disabled by default.
(use --enable-preview to enable preview APIs)Với cờ, kết quả khác hẳn lần trước:
[ 2 ms] findUser bắt đầu
[ 3 ms] fetchOrder bắt đầu
[ 112 ms] fetchOrder ném lỗi
[ 114 ms] findUser bị ngắt
[ 114 ms] main handle() thất bại: java.lang.IllegalStateException: kho hàng không phản hồi
[ 114 ms] main rời handle(): mọi subtask đã kết thúcHai ms sau khi fetchOrder lỗi, findUser bị ngắt và cả handle() kết thúc ở 114 ms thay vì 1.016 ms. Cơ chế gồm các thành phần sau, theo JEP 453:
- Owner: thread tạo scope. Chỉ owner được gọi
join. Vớifork, owner và các thread nằm trong chính scope đó (subtask được fork từ scope) đều được phép; thread ngoài scope gọiforksẽ nhậnWrongThreadException. fork(Callable)tạo subtask trong thread mới (mặc định là virtual thread) và trả vềSubtask.join()chờ mọi subtask xong hoặc bị huỷ. Bắt buộc phải gọi.shutdown()huỷ các subtask chưa xong bằng cách ngắt thread của chúng và chặn fork thêm.ShutdownOnFailuretự gọi nó khi có subtask lỗi đầu tiên.close(), thường gọi ngầm bởi try-with-resources, chờ mọi subtask thực sự kết thúc. Nhờ đó khi thoát khỏi khốitry, không còn thread nào sót.
Hệ quả quan trọng của structured concurrency: nếu thread owner bị ngắt trong lúc join(), mọi subtask cũng bị huỷ. Việc huỷ bỏ lan theo cấu trúc code chứ không phụ thuộc kỷ luật của lập trình viên. Lưu ý rằng ngắt chỉ có tác dụng với code hưởng ứng ngắt, như Thread.sleep, I/O của JDK; vòng lặp CPU thuần không kiểm tra cờ ngắt sẽ không dừng.
Subtask: chỉ đọc sau join
fork trả về Subtask với state() là SUCCESS, FAILED hoặc UNAVAILABLE. Gọi get() trước join(), hoặc quên join(), là lỗi lập trình mà API bắt ngay (Demo4Violation.java):
get() trước join -> IllegalStateException: Owner did not join after forking subtasks
sau join: state=SUCCESS, get()=1
đóng scope khi chưa join -> IllegalStateException: Owner did not join after forking subtasksĐây là điểm hay của thiết kế structured concurrency: Future.get() có thể gọi bất cứ lúc nào và chặn, còn Subtask.get() không bao giờ chặn và chỉ hợp lệ sau khi join, nên không thể vô tình tạo ra điểm chờ ẩn.
ShutdownOnSuccess và hạn chót
Chính sách thứ hai của structured concurrency là ShutdownOnSuccess: lấy kết quả của subtask thành công đầu tiên rồi huỷ phần còn lại. Ví dụ hỏi báo giá ba đơn vị vận chuyển song song (Demo3Success.java, thời gian giả lập bằng Thread.sleep):
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
scope.fork(() -> quote("GHN", 300));
scope.fork(() -> quote("GHTK", 120));
scope.fork(() -> quote("VTP", 500));
System.out.println("Báo giá nhanh nhất: " + scope.join().result());
} VTP bị huỷ sau khi có kết quả khác
GHN bị huỷ sau khi có kết quả khác
Báo giá nhanh nhất: GHTK:21200đ sau 131 msHai subtask chậm bị ngắt ngay khi GHTK trả lời, không phải chờ hết 500 ms. Để đặt hạn chót, dùng joinUntil(Instant); hết hạn thì ném TimeoutException và close() vẫn huỷ các subtask còn chạy:
--- hạn chót 200 ms, cả hai đều chậm hơn ---
A bị huỷ sau khi có kết quả khác
B bị huỷ sau khi có kết quả khác
TimeoutException sau 206 ms(Thông điệp "bị huỷ sau khi có kết quả khác" là chuỗi cố định trong code ví dụ; ở trường hợp thứ hai nó có nghĩa là bị huỷ khi hết hạn.) Cần chính sách khác, chẳng hạn thu thập mọi kết quả thành công và bỏ qua lỗi, thì ta kế thừa StructuredTaskScope và ghi đè handleComplete.
Quan sát structured concurrency bằng thread dump
Lợi ích thứ ba của structured concurrency là khả năng quan sát. Thread dump dạng văn bản (jstack) không có virtual thread, nhưng jcmd <pid> Thread.dump_to_file -format=json <file> ghi cả virtual thread và nhóm chúng theo "container". Ta chạy Demo7Dump.java (scope với hai subtask ngủ 8 giây) rồi dump. Output của jcmd là một file JSON dài (cả thread hệ thống, ForkJoinPool...), nên bài chỉ trích phần container của scope bằng script dump_view.py; lệnh đầy đủ nằm ở run_dump.sh và kết quả ở run_Demo7Dump.log:
java.util.concurrent.StructuredTaskScope$ShutdownOnFailure@232204a1 | threads: 2
thread 20 ''
thread 22 ''Hai virtual thread (tên rỗng) nằm gọn dưới container của scope ShutdownOnFailure, thay vì lẫn trong hàng nghìn thread của một pool chung. Khi một request treo, ta nhìn dump là biết những subtask nào thuộc cùng một scope.
Scoped values trong structured concurrency: thay ThreadLocal
Subtask của structured concurrency chạy ở thread khác, vậy làm sao truyền ngữ cảnh như trace id hay người dùng hiện tại? ThreadLocal không làm được điều này một cách rẻ: JEP 506 chỉ ra ba nhược điểm. Một là biến có thể bị ghi tuỳ ý (mutable). Hai là vòng đời không giới hạn: giá trị tồn tại đến khi gọi remove, và quên gọi thì rò rỉ hoặc nhiễm giữa các tác vụ trong thread pool. Ba là kế thừa tốn kém: thread con sao chép toàn bộ InheritableThreadLocal, rất nặng khi có hàng triệu thread.
ScopedValue ra đời để thay thế: bất biến, một chiều, và có vòng đời gắn với một khối code. Ở JDK 21 nó là preview (JEP 446). Demo5ScopedValue.java kết hợp cả hai API:
static final ScopedValue<RequestContext> CTX = ScopedValue.newInstance();
ScopedValue.where(CTX, new RequestContext("trace-7f3a", "khach-01")).run(() -> {
System.out.println(handle()); // handle() fork hai subtask, mỗi subtask đọc CTX.get()
ScopedValue.where(CTX, new RequestContext("trace-7f3a", "admin")).run(() ->
System.out.println("rebind trong: " + describe()));
System.out.println("sau rebind: " + describe());
});
System.out.println("sau run: " + describe());ngoài scope: (không có ngữ cảnh)
kho thấy trace-7f3a/khach-01
ví thấy trace-7f3a/khach-01
rebind trong: trace-7f3a/admin
sau rebind: trace-7f3a/khach-01
sau run: (không có ngữ cảnh)
--- ThreadLocal: subtask KHÔNG thấy giá trị của owner nếu không dùng Inheritable ---
subtask thấy ThreadLocal = nullCó bốn điều cần đọc từ kết quả. Thứ nhất, hai subtask chạy ở thread khác vẫn thấy ngữ cảnh của owner: StructuredTaskScope cho thread con kế thừa binding mà không sao chép gì. Thứ hai, rebinding chỉ có hiệu lực trong khối lồng bên trong; ra khỏi đó giá trị cũ trở lại. Thứ ba, ra khỏi run thì binding biến mất, nên không có chuyện quên dọn. Thứ tư, ThreadLocal thường không tự sang thread con (dòng cuối là null). Một lưu ý khác theo JEP: ForkJoinPool không hỗ trợ kế thừa scoped value, vì nó không bảo đảm thread con kết thúc trước scope của cha.
Scoped values không thay thế mọi ca dùng ThreadLocal. Giao tiếp hai chiều (callee đặt giá trị cho caller ở xa) hoặc bộ đệm theo thread cho đối tượng đắt khởi tạo vẫn thuộc về ThreadLocal; JEP không có ý định loại bỏ nó.
Đường đi của API structured concurrency qua các phiên bản
Structured concurrency là API đang tiến hoá, nên ghi rõ phiên bản là bắt buộc. Theo các JEP:
| JDK | StructuredTaskScope | ScopedValue |
|---|---|---|
| 19, 20 | Incubator (JEP 428, 437) | JDK 20: incubator (JEP 429) |
| 21 (LTS) | Preview (JEP 453) | Preview (JEP 446) |
| 22, 23, 24 | Preview lại (JEP 462, 480, 499) | Tinh chỉnh (JEP 464, 481, 487) |
| 25 (LTS) | Preview lần 5 (JEP 505): đổi sang open() và Joiner | Chính thức (JEP 506) |
| 26 | Preview lần 6 (JEP 525) | Chính thức |
Thay đổi lớn nhất của structured concurrency là ở JDK 25: không còn new StructuredTaskScope.ShutdownOnFailure(), mà dùng phương thức tĩnh StructuredTaskScope.open(), và chính sách hoàn thành được tách thành đối tượng Joiner. Ta chạy Demo6Jdk25.java trên JDK 25 (LTS) với --enable-preview --source 25:
try (var scope = StructuredTaskScope.open()) { // chờ tất cả thành công
Subtask<String> user = scope.fork(() -> call("user", 100));
Subtask<String> order = scope.fork(() -> call("order", 150));
scope.join();
}
try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulResultOrThrow())) { ... }
try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(),
cf -> cf.withTimeout(Duration.ofMillis(200)))) { ... }open(): user@trace-9b1c | order@trace-9b1c
anySuccessful: GHTK@trace-9b1c
timeout: TimeoutExceptionTương ứng: ShutdownOnFailure thành open() mặc định, ShutdownOnSuccess thành Joiner.anySuccessfulResultOrThrow(), và joinUntil thành cấu hình withTimeout khi mở scope. Giá trị TRACE kiểu ScopedValue trong chương trình này vẫn được các subtask thấy. Sang JDK 26 (JEP 525, 03/2026) API thay đổi tiếp: anySuccessfulResultOrThrow đổi tên thành anySuccessfulOrThrow, allSuccessfulOrThrow trả về danh sách kết quả thay vì stream subtask, và có thêm Joiner.onTimeout(). Ta kiểm chứng đúng điều đó: chạy cùng file trên JDK 26 thì biên dịch báo thiếu phương thức cũ, còn bản đã đổi tên (Demo6Jdk26.java) chạy cho cùng kết quả:
Dòng đầu của mỗi log (phiên bản JDK đầy đủ) được lược bớt ở đây; xem run_Demo6_jdk25.log và run_Demo6_jdk26.log. Thông báo lỗi biên dịch trên JDK 26 được lưu ở compile_error_jdk26.txt; lưu ý JDK 26 không còn trên máy khi viết lại bài này, nên đoạn lỗi biên dịch được lưu từ lần chạy trước và chưa chạy lại.
Demo6Jdk25.java:27: error: cannot find symbol
symbol: method <String>anySuccessfulResultOrThrow()
---
open(): user@trace-9b1c | order@trace-9b1c
anySuccessful: GHTK@trace-9b1c
timeout: TimeoutExceptionBài học thực tế: vì còn là preview, code structured concurrency dùng StructuredTaskScope có thể không biên dịch lại được khi nâng JDK. Quy tắc tốt là bọc API này trong một lớp mỏng của dự án (ví dụ phương thức parallel(task1, task2)), để khi API đổi chỉ sửa một chỗ.
Khi nào dùng structured concurrency và khi nào không
- Nên dùng structured concurrency: fan-out trong một request (gọi nhiều dịch vụ rồi ghép kết quả), "race" lấy kết quả đầu tiên, tác vụ cần huỷ đồng loạt khi một phần lỗi.
- Không phù hợp: tác vụ nền sống lâu hơn request (gửi email sau khi trả lời) vì chúng cố ý vượt khỏi phạm vi; pipeline sự kiện liên tục; xử lý dữ liệu song song kiểu stream, nơi
parallelStreamvẫn đúng chỗ. - Cẩn trọng: môi trường production chỉ chấp nhận API chính thức. Hiện tại
ScopedValueđã final từ JDK 25, cònStructuredTaskScopevẫn là preview và đòi--enable-previewở mọi nơi build, chạy, test. - Thay thế hiện tại: với JDK 21 không bật preview, cách an toàn nhất là
CompletableFuture.allOfkèm huỷ thủ công, hoặcExecutorService.invokeAnycho trường hợp "lấy kết quả đầu".
Lỗi thường gặp
- Quên
--enable-previewở một trong các bước (javac, java, test, IDE): lỗi "preview API disabled". - Quên
join(): némIllegalStateExceptionnhư đã thấy. - Gọi fork từ thread ngoài scope: bị từ chối bằng
WrongThreadException; owner và các thread trong scope được fork, cònjoinchỉ dành cho owner. - Subtask không hưởng ứng ngắt: scope không thể huỷ vòng lặp CPU thuần;
close()sẽ chờ nó xong. - Dùng ThreadLocal để truyền ngữ cảnh vào subtask: subtask không thấy giá trị (xem kết quả ở trên); dùng
ScopedValue. - Coi structured concurrency preview là ổn định: tên phương thức đã đổi hai lần từ JDK 21 tới 26.
Checklist structured concurrency
- Mọi tác vụ con tạo trong
try-with-resources của scope,join()trước khi đọc kết quả. - Chọn chính sách đúng: chờ tất cả (
ShutdownOnFailurehoặcopen()), lấy đầu tiên (ShutdownOnSuccesshoặcanySuccessful...). - Đặt hạn chót cho mọi fan-out gọi dịch vụ ngoài.
- Truyền ngữ cảnh bằng
ScopedValue, không bằng tham số rải rác hayThreadLocal. - Dùng
jcmd Thread.dump_to_file -format=jsonđể xem cây tác vụ khi chẩn đoán. - Ghi rõ JDK và bọc API preview trong một lớp mỏng.
Tóm lại, structured concurrency biến việc "sinh thread rồi nhớ dọn" thành một quy tắc của cấu trúc code: tác vụ con không thể sống lâu hơn khối đã sinh ra chúng. Tài liệu chính: JEP 453: Structured Concurrency (Preview), JEP 505: Structured Concurrency (Fifth Preview), JEP 525 (Sixth Preview) và JEP 506: Scoped Values.
Bài tiếp theo
Sau structured concurrency và phần concurrency hiện đại, series chuyển sang góc nhìn ngược lại: những thiết kế trông hợp lý nhưng làm hỏng code theo thời gian. Mời bạn đọc tiếp Anti-pattern trong Java: God object, anemic model, service locator.
