Structured Concurrency & ScopedValue: vòng đời nhóm task

TL;DR: ExecutorService + Future để hở hai lỗ: subtask có thể sống lâu hơn method đã sinh ra nó (task leak), và lỗi một nhánh không tự lan sang các nhánh anh em. StructuredTaskScope (JEP 505, preview thứ năm ở Java 25) ràng vòng đời cả nhóm subtask vào một khối try: fork tỏa nhánh, join hợp lưu, close bảo đảm không nhánh nào sống sót ra ngoài block. Mặc định open() là fail-fast (Joiner.awaitAllSuccessfulOrThrow()): một nhánh ném exception thì scope cancel ngay, các nhánh còn lại bị interrupt, join() ném FailedException. Đi cùng là ScopedValue (final Java 25, JEP 506) — context bất biến, tự thu hồi theo phạm vi, kế thừa gần như miễn phí xuống subtask — thay ThreadLocal thời virtual thread.

1. Vì sao concurrency không cấu trúc lại rò rỉ?

Một handler cần dữ liệu user và tồn kho để dựng response — hai nguồn độc lập, nên ta chạy song song bằng ExecutorService:

ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
Future<User>      userF  = executor.submit(() -> findUser(userId));
Future<Inventory> invF   = executor.submit(() -> checkInventory(eventId));
User user      = userF.get();   // (1)
Inventory inv  = invF.get();    // (2)
return new Response(user, inv);

Đoạn này gọn nhưng rò rỉ ở chỗ khó thấy. Nếu findUser ném exception ở dòng (1), ta thoát method — nhưng checkInventory vẫn chạy tiếp: không ai gọi invF.cancel(...), nên nó còn chiếm một virtual thread, có thể giữ một database connection, rồi kết quả bị vứt đi. Đó là task leak. Chiều ngược lại cũng tệ: checkInventory treo vì service dưới chết thì dòng (2) block mãi cùng cả thread xử lý request — caller không áp được deadline lên cả nhóm, chỉ thấy hai Future rời rạc.

Gốc rễ ở chữ "rời rạc". Mỗi submit ném một task vào pool như thả thư vào hòm thư công cộng: quan hệ cha–con giữa code đang gọi và task sinh ra không được mã hóa ở đâu cả, nên task con sống lâu hơn method sinh ra nó, lỗi không tự lan về cha, việc hủy phải làm thủ công từng cái. Code tuần tự thì trái ngược: hàm con luôn trả về trước hàm cha, exception lan theo call stack, block kết thúc thì mọi thứ nó mở ra đã đóng. Structured concurrency mang chính cấu trúc khối ấy trở lại cho code song song: ràng vòng đời subtask vào một block cú pháp, mở scope đầu block thì chắc chắn mọi subtask kết thúc trước khi ra khỏi block.

2. StructuredTaskScope: fan-out, rồi fan-in

Đến Java 25, StructuredTaskScope (trong java.util.concurrent) đang ở preview lần thứ năm (JEP 505): API ổn định hình dạng nhưng vẫn cần --enable-preview để biên dịch. Hình dạng dưới đây bám JDK 25; các bản preview trước có API khác đáng kể.

Ý tưởng cốt lõi gói trong ba động tác: mở scope, fork các subtask, rồi join chờ cả nhóm. Viết lại ví dụ ở phần 1:

Response handle(String userId, String eventId) throws InterruptedException {
    try (var scope = StructuredTaskScope.open()) {          // mo pham vi
        Subtask<User>      user = scope.fork(() -> findUser(userId));
        Subtask<Inventory> inv  = scope.fork(() -> checkInventory(eventId));
        scope.join();                                       // diem hop luu duy nhat
        return new Response(user.get(), inv.get());
    }
}

Khác biệt so với ExecutorService không nằm ở số dòng — gần như bằng nhau — mà ở những đảm bảo cấu trúc khối áp đặt. scope.fork(...) khởi chạy mỗi subtask trên một virtual thread riêng và trả về một Subtask — handle nhẹ cho kết quả tương lai. scope.join() chặn tới khi mọi subtask kết thúc, hoặc tới khi đủ điều kiện dừng sớm theo chính sách. Mấu chốt ở try-with-resources: khi luồng điều khiển rời khối try — dù return bình thường hay vì exception — scope.close() được gọi tự động và bảo đảm không subtask nào của scope còn sống, không có đường nào để một nhánh vượt ra ngoài block đã sinh ra nó.

Subtask.get() chỉ được gọi sau join() — khác cố ý với Future.get(). Future.get() tự nó là lệnh block, mời ta chờ từng task một theo thứ tự xen kẽ khó lường; Subtask.get() không block: tại lúc gọi nó, join() đã bảo đảm subtask kết thúc, nên nó chỉ đọc kết quả có sẵn. Việc chờ dồn về đúng một chỗ là join() — tỏa nhiều nhánh rồi gom tại một điểm hợp lưu duy nhất, đúng tinh thần "fan-out rồi fan-in".

Hãy hình dung scope như quản đốc giao việc cho tổ thợ: phát phiếu (fork), rồi đứng cửa xưởng đợi (join). Quy tắc bất di bất dịch: quản đốc không rời xưởng chừng nào còn một thợ chưa về; buộc phải đi (hết giờ, cháy xưởng) thì gọi tất cả về trước. Cái xưởng có tường có cửa ấy chính là khối try.

3. Khi nào scope dừng cả nhóm task?

Vậy scope.open() không tham số áp chính sách gì? Mặc định (theo JEP 505) là Joiner.awaitAllSuccessfulOrThrow() — fail-fast, không phải "chờ tất cả rồi tính sau". Mọi nhánh thành công thì nó chờ đủ cả nhóm rồi cho join() trả về null (kiểu Void), đọc kết quả qua từng Subtask.get(). Nhưng chỉ cần một subtask ném exception, scope bị cancel ngay: mọi nhánh chưa xong bị interrupt, join() ném StructuredTaskScope.FailedException bọc nguyên nhân gốc, kích hoạt close() dọn sạch (đoạn code phần 2 đã mang sẵn hành vi này).

Vẽ chuỗi sự kiện đó thành sơ đồ — một subtask fail kéo cả cây xuống có trật tự:

flowchart TD
    P["Scope cha: open() roi join()"] --> A["Subtask A: findUser"]
    P --> B["Subtask B: checkInventory"]
    P --> C["Subtask C: fetchPromo"]
    A -- "nem exception" --> X["Scope bi cancel"]
    X -- "interrupt" --> B
    X -- "interrupt" --> C
    X --> J["join() nem FailedException<br/>close() bao dam moi nhanh da ket thuc"]
    style X fill:#FCA5A5
    style B fill:#FCD34D
    style C fill:#FCD34D

Chính sách kết thúc do Joiner truyền vào open(...) quyết định. (Bản preview cũ dùng ShutdownOnFailure/ShutdownOnSuccess cho vai trò này — đã thay bằng các Joiner tương ứng.)

3.1 Ba Joiner cho kết quả "tất cả phải thành công"

Họ Joiner có ba thành viên hay gặp cho bài toán chờ nhiều nhánh, khác nhau ở hai câu hỏi: có fail-fast không và join() trả về gì.

Ví dụ user + inventory ở phần 2 rơi vào ô thứ nhất. Khi các nhánh đồng kiểu, muốn gom kết quả như dòng chảy, allSuccessfulOrThrow() gọn hơn vì join() trả thẳng stream:

List<Quote> fetchQuotes(List<String> symbols) throws InterruptedException {
    try (var scope = StructuredTaskScope.open(Joiner.<Quote>allSuccessfulOrThrow())) {
        symbols.forEach(s -> scope.fork(() -> fetchQuote(s)));
        return scope.join()                  // Stream cac Subtask da thanh cong
                    .map(Subtask::get)
                    .toList();
    }
}

Còn awaitAll() (ô thứ ba) cho khi fail-fast không phải điều ta muốn — chạy mười phép kiểm tra rồi cần báo cáo đầy đủ cái nào đậu cái nào rớt: scope chờ hết mọi nhánh, ta tự duyệt Subtask.state() từng cái.

Việc hủy dựa trên interrupt của Java, nên chỉ cắt được subtask tôn trọng interrupt: phần lớn blocking I/O trên virtual thread đều tôn trọng, nhưng vòng lặp thuần CPU không kiểm tra Thread.interrupted() thì không bị cắt. Đây là cancellation hợp tác — cơ chế đã học ở Thread API và vòng đời, gặp lại ở Executor (hủy qua Future.cancel).

3.2 Lấy kết quả thành công đầu tiên

Mặt đối xứng của fail-fast là success-fast. Query cùng thông tin từ ba replica, chỉ cần một câu trả lời nhanh nhất, các cái còn lại đều dư. Joiner.anySuccessfulResultOrThrow() phục vụ đúng kiểu này:

String fetchFastest(List<String> replicas) throws InterruptedException {
    try (var scope = StructuredTaskScope.open(
             Joiner.<String>anySuccessfulResultOrThrow())) {
        for (String r : replicas) {
            scope.fork(() -> queryReplica(r));
        }
        return scope.join();                // tra ket qua thanh cong DAU TIEN
    }
}

Subtask thành công đầu tiên kích hoạt cancel scope; các nhánh còn lại bị interrupt. scope.join() trả thẳng giá trị đó nên không cần giữ Subtask ref; nếu mọi nhánh đều thất bại, join() ném exception gom các nguyên nhân. Đây là pattern hedged request kinh điển — đánh đổi ít tài nguyên dư để cắt đuôi độ trễ — gói trong vài dòng lifecycle đảm bảo, thay vì CompletableFuture.anyOf (bài Future & CompletableFuture) cộng logic hủy thủ công.

3.3 Deadline cho cả nhóm

Vì cả nhóm subtask sống trong một scope, ta áp được deadline lên toàn bộ nhóm tại điểm join — thứ ba Future rời rạc không cho làm gọn. Timeout là tham số thứ hai của open(...): open(joiner, cf -> cf.withTimeout(Duration.ofMillis(500))) (phải viết joiner tường minh vì cần config phía sau). Quá hạn 500ms, scope bị cancel như khi một subtask fail: mọi nhánh chưa xong bị interrupt, join() ném StructuredTaskScope.TimeoutException (unchecked, lồng trong API mới — không phải java.util.concurrent.TimeoutException cũ). Deadline thành thuộc tính của cả phép tính concurrent, không phải gánh thủ công lên từng nhánh; pattern đầy đủ ở capstone phần 6.

4. So với ExecutorService: vì sao có kỷ luật hơn

Đặt hai mô hình cạnh nhau, khác biệt là hình dạng vòng đời. ExecutorService tách rời ba việc — submit, chờ kết quả, dọn dẹp — nên quan hệ cha–con chỉ tồn tại trong đầu lập trình viên; quên hủy hay quên chờ thì trình biên dịch không cản. StructuredTaskScope buộc cả ba về cùng một block try, khiến ba thuộc tính khó-đảm-bảo thành mặc định miễn phí: không task leak (close() luôn chờ hết), lỗi không bị nuốt (lan lên cha), cancellation không bị quên (tự động theo chính sách).

Điều này không có nghĩa ExecutorService bị khai tử. Một pool dài hạn nhận việc từ nhiều nguồn không liên quan — ví dụ background worker xử lý hàng đợi sự kiện — vẫn là việc của nó: ở đó vốn dĩ không có quan hệ cha–con để cấu trúc hóa. StructuredTaskScope dành cho tình huống ngược lại: một đơn vị công việc tỏa nhiều nhánh rồi gom lại, sống và chết trong phạm vi một lời gọi — phần lớn fan-out khi xử lý một request thuộc loại này.

5. ScopedValue: mang context bất biến qua phạm vi

Còn một mảnh ghép. Khi request fan-out thành nhiều subtask, thường có dữ liệu ngữ cảnh — user đăng nhập, traceId gắn log, locale — cần tới tay mọi nhánh. Cách cũ là ThreadLocal (confinement theo thread từ bài Confinement), và dưới thời virtual thread nó lộ rõ các vết nứt.

ThreadLocal có ba vấn đề cố hữu. Một, nó mutable không giới hạn: bất kỳ code nào cầm được đều có thể set đè, nên khó suy luận giá trị thực tại một điểm. Hai, vòng đời không có biên rõ: quên remove() thì giá trị bám lại trên thread và rò sang request sau khi thread tái dùng từ pool. Ba — và virtual thread làm nặng thêm như bài trước đã phân tích: mỗi ThreadLocal giam một bản sao riêng theo từng thread, mà với cả triệu virtual thread, chi phí bộ nhớ phình theo cách mô hình vài trăm platform thread không phải lo.

ScopedValue (final ở Java 25, JEP 506) là câu trả lời. Thay vì "gắn giá trị lên thread rồi nhớ gỡ", nó "ràng giá trị vào một phạm vi động, tự biến mất khi ra khỏi phạm vi". ScopedValue nằm trong java.lang nên không cần import:

static final ScopedValue<RequestContext> CONTEXT = ScopedValue.newInstance();

void handleRequest(RequestContext ctx) {
    ScopedValue.where(CONTEXT, ctx).run(() -> {
        // Trong pham vi nay - va moi loi goi no dan toi - CONTEXT.get() tra ve ctx
        processOrder();
    });
    // Ra khoi run(): CONTEXT khong con bound nua
}

void processOrder() {
    RequestContext ctx = CONTEXT.get();   // doc o sau trong call tree, khong can truyen tay
    log.info("processing user={}, trace={}", ctx.userId(), ctx.traceId());
}

Ba tính chất ấy đối ngược đúng ba vấn đề trên: giá trị bất biến (không set để đè), vòng đời tự đóng (hết run là tan, không remove() để quên nên không rò sang request sau), và kế thừa rẻ (bất biến nên chia cho hàng triệu virtual thread không tốn bản sao).

Mảnh ghép cuối: ScopedValue bắt tay StructuredTaskScope. Khi fork một subtask bên trong phạm vi ScopedValue, subtask đó — chạy trên virtual thread riêng — vẫn kế thừa các ràng buộc đang hiệu lực. Đó là lý do hai cơ chế ra đời cùng thế hệ và thiết kế ăn khớp:

ScopedValue.where(CONTEXT, ctx).run(() -> {
    try (var scope = StructuredTaskScope.open()) {   // default: fail-fast
        scope.fork(() -> { /* CONTEXT.get() o day van la ctx */ return checkInventory(); });
        scope.fork(() -> { /* va o day cung vay */              return chargePayment();  });
        scope.join();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

6. Capstone: TicketFlow đặt vé có cấu trúc

Hãy ráp tất cả vào TicketFlow. Ở các phiên bản trước, BookingService.book đã thread-safe nhờ ConcurrentHashMap.compute và tầng request đã chuyển sang virtual thread. Giờ dựng phần còn thiếu: một request đặt vé cần song song hai việc độc lập — kiểm tồn kho và xác thực thanh toán — với fail-fast, một deadline chung, và mang context request qua cả hai nhánh.

record RequestContext(String userId, String traceId) {}
static final ScopedValue<RequestContext> CONTEXT = ScopedValue.newInstance();

BookingResult handleBooking(BookingRequest req) {
    var ctx = new RequestContext(req.userId(), newTraceId());
    return ScopedValue.where(CONTEXT, ctx).call(() -> doBooking(req));
}

BookingResult doBooking(BookingRequest req) {
    try (var scope = StructuredTaskScope.open(
             Joiner.<Object>awaitAllSuccessfulOrThrow(),   // = default, viet ro vi can config
             cf -> cf.withTimeout(Duration.ofSeconds(2)))) {
        Subtask<Inventory>   inv = scope.fork(() -> checkInventory(req.eventId()));
        Subtask<PaymentAuth> pay = scope.fork(() -> authorizePayment(req.userId()));
        scope.join();   // fail-fast: het ton kho HOAC the tu choi -> huy nhanh kia ngay
        // Ca hai da thanh cong; giu cho la buoc cuoi, nhanh va nguyen tu.
        Booking booking = service.book(req.eventId(), req.userId());
        return BookingResult.ok(booking, inv.get(), pay.get());
    } catch (Exception e) {
        return BookingResult.fail(rootCause(e).getMessage());
    }
}

Hai chi tiết đáng dừng. Một, doBooking không khai throws: join() có thể ném InterruptedException, FailedException hay TimeoutException, nhưng catch (Exception e) đón hết và đổi thành BookingResult.fail, nên handleBooking gọi qua ScopedValue.where(...).call(...) mà không phải khai checked exception. Joiner default ở đây trả null nên kết quả đọc qua hai ref inv và pay.

So với Future rời rạc ở phần 1, đoạn này đóng kín mọi lỗ rò: authorizePayment ném CardDeclinedException thì checkInventory bị interrupt ngay thay vì chạy phí; một service treo thì deadline 2 giây cắt cả nhóm; dù ra bằng đường nào, close() cũng đảm bảo không subtask nào sống quá khối try; và context userId/traceId chảy xuống cả hai nhánh không cần nhét vào chữ ký từng hàm. Việc giữ chỗ thực (service.book) chỉ chạy sau khi cả hai điều kiện đã thỏa, nên không bao giờ trừ tồn kho cho một thanh toán hỏng. Đây là code minh họa, chưa có module riêng; lõi đặt vé thread-safe vẫn là service.book từ v3.

7. 📚 Deep Dive Oracle

8. Liên hệ các bài khác

• Thread API và vòng đời — interrupt và cooperative cancellation là cơ chế bên dưới việc scope hủy subtask; không hiểu interrupt thì không giải thích được vì sao vòng lặp CPU thuần "không chịu chết".
• Confinement — ThreadLocal mà ScopedValue thay thế chính là công cụ confinement theo thread; so hai cách ràng dữ liệu vào "phạm vi" rất đáng để ngẫm.
• Executor và thread pool — đối trọng trực tiếp: hủy thủ công qua Future.cancel so với cancel tự động theo chính sách Joiner.
• Future & CompletableFuture — anySuccessfulResultOrThrow() thay cho pattern CompletableFuture.anyOf + hủy tay; nhìn lại để thấy structured concurrency rút gọn được gì.
• Virtual Threads — tiền đề vật chất của bài này: fork tạo một virtual thread cho mỗi subtask, chỉ khả thi vì thread đã rẻ; và ScopedValue giải đúng bài toán ThreadLocal phình heap nêu ở đó.

9. Tóm tắt — khép lại module

Mười bảy bài concept đã đi một vòng dài: từ Process và Thread — nơi một OS thread là tài nguyên đắt, ánh xạ one-to-one xuống kernel — tới thế giới virtual thread rẻ đến mức tạo cả triệu, và structured concurrency cho bộ khung kỷ luật để cái rẻ ấy không thành hỗn loạn. Java đã đi từ mô phỏng thread của hệ điều hành đến tự dựng mô hình thực thi nhẹ của riêng JVM.

Nhưng một bài học không đổi suốt chặng: dù thread là platform hay virtual, dù gom bằng pool hay scope, công thức gây bug vẫn nguyên — shared memory + mutable data + unsynchronized access. Mọi công cụ trong module — confinement, immutability, lock, atomic, concurrent collection, executor, future, virtual thread, structured concurrency — đều chỉ là cách triệt tiêu một trong ba yếu tố đó, hoặc canh gác khi buộc phải giữ cả ba. Câu hỏi senior phải hỏi trước mỗi dòng concurrent thì không đổi: dữ liệu có chia sẻ không, có thay đổi không, và nếu cả hai thì ai canh gác? Bài kế tiếp gom cả chặng vào một trang cheat sheet để bookmark và tự kiểm.

10. Tự kiểm tra

Bài tiếp theo: Tổng kết module — cheat sheet & self-assessment