CompletableFuture: dựng pipeline async compose được

TL;DR: CompletableFuture thay Future blocking bằng pipeline khai báo: thenApply cho bước đồng bộ, thenCompose cho bước async nối tiếp (chính là flatMap của thế giới async), thenCombine hợp lưu hai nhánh song song, allOf/anyOf gom nhiều future, exceptionally/handle đón lỗi, orTimeout chặn treo vĩnh viễn. Ba luật sống còn: luôn truyền executor riêng cho các method *Async (đừng mượn commonPool cho I/O), luôn đóng chuỗi bằng một chỗ tiêu thụ lỗi, và nhớ rằng ThreadLocal/MDC không tự theo task qua các stage. Kết quả là một luồng xử lý không thread nào phải ngồi chờ ở giữa — thread chỉ làm khi có việc.

Bài trước kết ở hai giới hạn của Future thuần: get chặn thread, và chuỗi phụ thuộc phải cắt vụn thành các lần get nối tiếp vì Future không có chỗ gắn "khi xong thì làm tiếp việc này". Nhớ lại yêu cầu TicketFlow: lấy user → tính giá dựa trên user → gửi xác nhận. Bài này lấp đúng khoảng trống đó — dựng cả chuỗi thành một pipeline khai báo, và không một lời gọi get chặn nào trong đường đi chính.

1. CompletableFuture: pipeline async compose được

CompletableFuture<T> (Java 8) vừa là Future (cho get/cancel quen thuộc) vừa là CompletionStage — vế sau mới là phần đổi cuộc chơi: nó cho phép khai báo các bước sẽ chạy khi kết quả sẵn sàng, mà không thread nào phải ngồi chờ ở giữa.

1.1 Khởi tạo: supplyAsync và runAsync

Cách thông dụng nhất mở đầu pipeline là supplyAsync (nhận Supplier, có trả về) hoặc runAsync (nhận Runnable, không trả về).

CompletableFuture<User> cf =
    CompletableFuture.supplyAsync(() -> loadUser(id), pool);

Nếu không truyền executor, các method *Async chạy trên ForkJoinPool.commonPool() — bẫy mà phần Cạm bẫy sẽ mổ. Quy tắc thực dụng: production gần như luôn truyền executor riêng.

1.2 Ba phép biến đổi: thenApply, thenCompose, thenCombine

Ba phép ghép nối này phủ gần hết nhu cầu hằng ngày; phân biệt chúng là chìa khóa dùng đúng.

thenApply biến đổi kết quả bằng hàm thường T -> U — dùng khi bước tiếp theo là phép tính đồng bộ, nhanh.

CompletableFuture<String> name =
    CompletableFuture.supplyAsync(() -> loadUser(id), pool)
                     .thenApply(User::displayName);          // User -> String

thenCompose dùng khi bước tiếp theo bản thân nó trả về một CompletableFuture: T -> CompletableFuture<U>. Nó chính là flatMap của thế giới async — làm phẳng cái lồng, đúng như flatMap ở Optional/Stream.

Nếu lỡ dùng thenApply cho hàm trả future, bạn nhận một CompletableFuture<CompletableFuture<U>> lồng hai lớp — code vẫn compile, nhưng để lấy giá trị thật phải bóc hai lần (và rất dễ quên chờ future bên trong hoàn tất, mất luôn cả lỗi của nó). thenCompose làm phẳng để tránh đúng cái lồng đó.

// loadUser va priceForAsync deu la thao tac async tra CompletableFuture
CompletableFuture<Price> price =
    CompletableFuture.supplyAsync(() -> loadUser(id), pool)
                     .thenCompose(user -> priceForAsync(user, pool));  // KHONG long hai lop

thenCombine dùng khi bạn có hai pipeline độc lập và muốn gộp kết quả khi cả hai cùng xong: (T, U) -> V. Hai nhánh chạy song song, không nhánh nào phụ thuộc nhánh kia.

CompletableFuture<Seat>    seatCf = CompletableFuture.supplyAsync(() -> reserveSeat(id), pool);
CompletableFuture<Invoice> invCf  = CompletableFuture.supplyAsync(() -> draftInvoice(id), pool);
CompletableFuture<Confirmation> conf =
    seatCf.thenCombine(invCf, (seat, invoice) -> new Confirmation(seat, invoice));

2. Một stage chạy trên thread nào?

Mỗi phép ghép nối có ba dạng, và khác nhau giữa chúng quyết định task chạy trên thread nào.

thenApply(fn) — không hậu tố — chạy fn trên thread vừa hoàn tất stage trước, hoặc trên thread gọi nếu stage trước đã xong sẵn; bạn không kiểm soát được đó là thread nào, và fn nặng hoặc blocking sẽ ghim thread của stage trước. thenApplyAsync(fn) đẩy sang commonPool; thenApplyAsync(fn, executor) đẩy sang đúng executor bạn chỉ định.

Quy tắc: hàm nhỏ, thuần CPU, nhanh thì biến thể không hậu tố tối ưu vì tránh chi phí chuyển thread; hàm blocking hoặc nặng thì dùng *Async với executor riêng để chủ động chọn nơi nó chạy — đúng kỹ thuật capstone dùng.

Ghép ba phép biến đổi và quy tắc chọn thread, pipeline điển hình cho trang xác nhận TicketFlow trông như sau:

flowchart LR
    A["supplyAsync(fetchUser, ioPool)"] --> B["thenCompose(u -> fetchOrders(u))"]
    P["supplyAsync(fetchPrices, ioPool)"] --> C["thenCombine(pricesCf, gop 2 nhanh)"]
    B --> C
    C --> D["exceptionally(fallback)"]
    style C fill:#6EE7B7
    style D fill:#FCD34D

Đọc sơ đồ theo con mắt "thread nào chạy stage nào": hai supplyAsync chạy trên ioPool (quên executor thì rơi xuống commonPool — bẫy phần Cạm bẫy); thenCompose/thenCombine không hậu tố chạy trên thread vừa hoàn tất stage trước; exceptionally cuối hứng lỗi từ bất kỳ stage nào phía trên.

3. Kết hợp nhiều future và xử lý lỗi

3.1 allOf và anyOf

allOf cho bạn một CompletableFuture<Void> hoàn tất khi mọi future trong một mảng đều xong.

CompletableFuture<?>[] all = requests.stream()
        .map(r -> gateway.submit(r))
        .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(all)
        .thenRun(() -> log.info("Đã xử lý {} request", all.length));

allOf trả Void, nên muốn lấy từng kết quả vẫn phải join từng future sau khi allOf xong — nhưng lúc đó join không còn chặn vì mọi thứ đã hoàn tất. anyOf ngược lại: hoàn tất ngay khi bất kỳ future nào xong, hợp pattern "hỏi nhiều nguồn, lấy câu trả lời sớm nhất".

3.2 exceptionally, handle, whenComplete

Lỗi trong pipeline async lan theo chuỗi: một stage ném exception thì các stage biến đổi phía sau bị bỏ qua, lỗi trôi xuống tới khi gặp stage biết xử lý. Có ba cách bắt.

exceptionally(fn) chỉ kích hoạt khi có lỗi, cho bạn trả về một giá trị thay thế: Throwable -> T. Lưới an toàn cuối chuỗi.

CompletableFuture.supplyAsync(() -> loadUser(id), pool)
    .thenApply(User::displayName)
    .exceptionally(ex -> "khách ẩn danh");   // loi o bat ky dau phia tren deu roi vao day

handle(fn) luôn chạy, dù thành công hay lỗi: (T, Throwable) -> U. Vừa xử lý lỗi vừa biến đổi kết quả, hợp khi muốn quy cả hai nhánh về một kiểu chung.

.handle((value, ex) -> ex != null
        ? BookingResult.fail(ex.getMessage())
        : BookingResult.ok(value));

whenComplete(fn) cũng luôn chạy nhưng không đổi được kết quả: (T, Throwable) -> void. Nó dành cho side-effect như logging hay dọn tài nguyên, để nguyên kết quả (hoặc lỗi) chảy tiếp.

Một chi tiết hay vấp: exception bên trong một stage bị gói trong CompletionException, nên khi xử lý ở exceptionally/handle, nguyên nhân thật thường nằm ở ex.getCause() — đúng như ExecutionException của Future; luôn bóc lớp đó trước khi phân loại lỗi.

3.3 orTimeout và completeOnTimeout

Một pipeline không giới hạn thời gian có thể treo vĩnh viễn. Từ Java 9, CompletableFuture có hai van an toàn dựng sẵn: orTimeout(t, unit) làm future hỏng với TimeoutException nếu chưa xong sau t; completeOnTimeout(value, t, unit) hoàn tất future với một giá trị mặc định thay vì ném lỗi.

gateway.submit(req)
       .orTimeout(2, TimeUnit.SECONDS)               // qua 2s thi hong voi TimeoutException
       .exceptionally(ex -> BookingResult.fail("hệ thống bận, thử lại sau"));

Cần phân biệt van này với việc thực sự hủy công việc. orTimeout chỉ làm future hoàn tất theo nhánh lỗi, không dừng task phía dưới: task đang gọi mạng vẫn tiếp diễn tới khi tự xong — đúng giới hạn cancellation đã bàn ở bài Executor & thread pool. Timeout là "tôi thôi không chờ nữa", không phải "công việc dừng lại".

4. Capstone: luồng xác nhận → thông báo viết bằng CompletableFuture

TicketFlow v3 ráp những mảnh trên thành luồng đặt vé thật. BookingGateway nhận request và trả về CompletableFuture<BookingResult> — caller không bao giờ phải chặn. Bên trong là chuỗi compose validate → book → notify, mỗi bước trên pool phù hợp.

public CompletableFuture<BookingResult> submit(BookingRequest req) {
    Objects.requireNonNull(req);
    return CompletableFuture
            .supplyAsync(() -> validate(req), bookingPool)        // mo chuoi tren bookingPool
            .thenApplyAsync(this::performBooking, bookingPool)    // book: T -> BookingResult
            .thenComposeAsync(this::notifyAsync, notifyPool)      // notify async: noi stage khac
            .orTimeout(overallTimeout.toMillis(), MILLISECONDS)   // van an toan ca chuoi
            .exceptionally(this::handleException);                // gom moi nhanh loi ve mot cho
}

Mỗi lựa chọn nhắc lại một ý của bài. thenComposeAsync thay thenApplyAsync ở bước notify vì gửi thông báo là thao tác async trả CompletableFuture (dùng thenApply sẽ ôm future lồng future); tách bookingPool/notifyPool theo quy tắc phần 2 để notify gọi mạng không ăn mòn pool đặt vé; và handleException bóc CompletionException để phân loại nguyên nhân thật:

private BookingResult handleException(Throwable ex) {
    Throwable cause = (ex instanceof CompletionException) ? ex.getCause() : ex;
    if (cause instanceof SoldOutException so) return BookingResult.fail("Sold out: " + so.getMessage());
    if (cause instanceof TimeoutException)    return BookingResult.fail("Booking timed out");
    return BookingResult.fail("Internal error: " + cause.getMessage());
}

Bước notifyAsync còn cho thấy một quyết định nghiệp vụ gọn: nếu kết quả đã là thất bại (sold out), nó trả thẳng future đã hoàn tất bằng CompletableFuture.completedFuture(r) thay vì gửi thông báo; nếu thành công, nó runAsync việc gửi trên notifyPool rồi thenApply(v -> r) giữ nguyên kết quả book.

Toàn bộ pipeline không một lời gọi get chặn nào trong đường đi chính: thread chỉ làm khi có việc, lúc chờ I/O thì nhường pool cho request khác. Và vì BookingService đã thread-safe từ các bài trước, chuỗi này không bao giờ bán vượt capacity kể cả khi cả lô request đổ vào cùng lúc (gom bằng allOf).

5. Cạm bẫy

Ba cái bẫy dưới đây gây phần lớn sự cố CompletableFuture production, và cả ba đều âm thầm — không bẫy nào báo lỗi lúc biên dịch.

Cái bẫy nguy hiểm nhất là nuốt exception. Pipeline có nhánh cuối không exceptionally/handle/whenComplete và không ai gọi get/join sẽ nuốt exception trong im lặng — task hỏng mà không một dòng log. Đây là phiên bản async của catch rỗng. Quy tắc: mọi pipeline phải kết thúc ở một chỗ tiêu thụ lỗi — stage xử lý lỗi, hoặc get/join có bắt exception đàng hoàng.

Cái bẫy thứ hai là dùng commonPool cho task blocking. Method *Async không truyền executor chạy trên ForkJoinPool.commonPool() — một pool dùng chung toàn JVM, sizing nhỏ (theo số CPU core) cho task ngắn thuần CPU. Đổ blocking I/O vào đó — gọi DB, gọi HTTP — chỉ vài task chờ là cả pool nghẹt, và mọi thứ khác trong JVM dùng chung nó (kể cả parallel stream) bị vạ lây. Cơ chế bên trong commonPool — work-stealing, cách nó được sizing, ManagedBlocker khi buộc phải chặn — thuộc bài Fork/Join; ở đây chỉ cần luật: đừng mượn commonPool cho I/O, luôn truyền executor riêng (đây là lý do capstone tách bookingPool/notifyPool).

Cái bẫy thứ ba tinh vi hơn: mất thread context. Vì một stage chạy trên bất kỳ thread nào trong pool — biến thể không *Async thậm chí chạy trên thread của stage trước — mọi thứ gắn vào thread không theo task sang stage sau: ThreadLocal, security context, MDC của logging (Mapped Diagnostic Context — map key-value gắn theo thread mà logging framework chèn vào mỗi dòng log, thường chứa trace id), transaction context. Đó là nguồn của lỗi "log mất trace id" hay "mất user context giữa chừng" khi một stage *Async chạy trên thread khác. ScopedValue (final ở Java 25) được thiết kế chính để truyền context qua các ranh giới như thế — xem bài ScopedValue.

6. 📚 Deep Dive Oracle

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

• Future, FutureTask & CompletionService — bài trước, đặt ra hai giới hạn (blocking + không compose) mà pipeline này lấp; CompletionException ở đây là họ hàng của ExecutionException ở đó.
• Executor & thread pool — nguồn của các executor bạn truyền cho *Async, và interruption policy mà orTimeout không vượt qua được.
• Fork/Join: chia để trị song song — mổ ForkJoinPool.commonPool(): work-stealing, cách sizing, ManagedBlocker; là pool mặc định của mọi *Async không truyền executor.
• ScopedValue — truyền context ở quy mô virtual thread — ScopedValue truyền context qua ranh giới stage, chữa gốc bẫy "mất thread context".

7. Tóm tắt

• CompletableFuture = Future + CompletionStage: khai báo các bước sẽ chạy khi kết quả sẵn sàng, không thread nào chờ ở giữa.
• Ba phép biến đổi: thenApply (T -> U đồng bộ), thenCompose (T -> CompletableFuture<U>, làm phẳng như flatMap), thenCombine ((T, U) -> V, hợp lưu hai nhánh song song).
• Ba biến thể thread: không hậu tố (thread stage trước), *Async (commonPool), *Async có executor (pool chỉ định) — hàm blocking phải dùng biến thể cuối.
• Gom + xử lý lỗi: allOf/anyOf gom nhiều future; exceptionally/handle/whenComplete bắt lỗi (bóc CompletionException lấy nguyên nhân thật); orTimeout/completeOnTimeout chặn treo.
• Ba luật production: truyền executor riêng cho *Async; đóng chuỗi bằng một chỗ tiêu thụ lỗi; nhớ ThreadLocal/MDC không tự theo task qua stage.

8. Tự kiểm tra

Bài tiếp theo: Fork/Join: Chia để trị song song với work-stealing