Future & CompletableFuture: từ blocking tới pipeline async

TL;DR: submit một Callable trả về Future — tay cầm tới kết quả async, nhưng là tay cầm cứng: get chặn thread và không ghép nối được các bước phụ thuộc. CompletableFuture thay nó 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 khi dùng: 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.

1. Future gánh được bao xa khi các bước phụ thuộc nhau?

Bài Executor & thread pool khép lại bằng một câu hỏi treo: submit một Callable xong, ta nhận về một Future — dùng thế nào cho đúng, và nó gánh được bao xa khi các bước bắt đầu phụ thuộc nhau?

Lấy trang xác nhận đặt vé của TicketFlow: để dựng nó, hệ thống gọi ba service — load hồ sơ user, tính giá cuối cho ghế đã chọn, rồi gửi email xác nhận sau khi hai bước trên xong. Mỗi lời gọi chừng 200ms. Gọi tuần tự là 600ms; gọi khéo — hai bước đầu song song, bước ba nối sau — chỉ còn hơn 400ms, và thread phục vụ request không đứng chờ giây nào.

2. Callable và Future

2.1 Runnable không trả về gì, Callable thì có

Runnable.run() trả void và không được ném checked exception — hợp task "làm rồi thôi" nhưng vô dụng khi cần một giá trị. Từ Java 5, java.util.concurrent thêm Callable<V> lấp chỗ đó: call() trả về V và được phép ném checked exception.

Callable<Integer> countSeats = () -> {
    Thread.sleep(50);                 // gia lap truy van DB
    return 128;
};

submit một Callable cho executor, ta không nhận kết quả ngay — task còn chưa chắc đã chạy. Thứ nhận lại là một Future<V>: lời hứa rằng kết quả kiểu V sẽ có ở một thời điểm trong tương lai.

try (var pool = Executors.newFixedThreadPool(2)) {
    Future<Integer> seats = pool.submit(countSeats);
    // ... lam viec khac trong luc task chay ...
    int n = seats.get();              // chan o day toi khi task xong
    System.out.println("Còn " + n + " chỗ");
}

Future mô hình hóa vòng đời task async qua bốn method: get() lấy kết quả (chặn nếu chưa xong), get(timeout, unit) chờ tối đa một khoảng, cancel(mayInterruptIfRunning) yêu cầu hủy, và isDone()/isCancelled() hỏi trạng thái — một vòng đời chỉ đi một chiều, từ "chưa xong" sang "xong" (thành công, lỗi, hoặc bị hủy) và không bao giờ quay lại.

2.2 get là một biên giới: nó dồn mọi thứ về thread gọi

Điểm tinh tế nhất của Future là get làm gì với lỗi. Task chạy trên thread khác; nếu call() ném exception, nó không bay ngược về thread gọi theo cách thường được. Executor bắt lấy, gói vào ExecutionException, ném ra đúng lúc bạn gọi get — nguyên nhân thật nằm trong getCause().

Future<Integer> f = pool.submit(() -> { throw new IllegalStateException("kho vé tạm khóa"); });
try {
    f.get();
} catch (ExecutionException e) {
    Throwable cause = e.getCause();   // IllegalStateException goc nam o day
    log.warn("Task hỏng: {}", cause.getMessage());
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();   // khoi phuc co interrupt roi moi xu ly
}

get như cửa hải quan giữa hai thread: mọi thứ task sản sinh bị giữ lại bên kia cho tới khi ai đó qua cửa mang về — lỗi không bị nuốt, nhưng bạn buộc phải đứng chờ.

Hai checked exception get ném ra nói hai chuyện khác hẳn. ExecutionException nghĩa là task chạy và hỏng. InterruptedException nghĩa là chính thread đang chờ bị interrupt — task có thể vẫn chạy bình thường. Cách xử lý đúng cho InterruptedException gần như luôn là khôi phục cờ interrupt bằng Thread.currentThread().interrupt() rồi mới quyết định dừng, đúng interruption policy đã bàn ở bài Executor & thread pool.

2.3 FutureTask: bản thân Future cũng là một Runnable

Một mảnh ghép giải thích vì sao mô hình này ráp được vào executor: FutureTask. Nó vừa là Runnable (executor chạy được) vừa là Future (bạn lấy kết quả được). Khi bạn submit(callable), executor gói callable vào một FutureTask, chạy nó như Runnable, rồi trả về dưới mặt nạ Future.

FutureTask<Integer> task = new FutureTask<>(countSeats);
new Thread(task).start();             // chay nhu Runnable
int n = task.get();                   // lay ket qua nhu Future

FutureTask cũng dùng được làm "kết quả tính một lần, nhiều thread cùng chờ": gọi get nhiều lần từ nhiều thread đều an toàn, lần đầu chặn tới khi tính xong, các lần sau trả ngay kết quả đã cache.

3. CompletionService: gom kết quả theo thứ tự hoàn tất

Một tình huống thường gặp: bạn submit một loạt task và muốn xử lý kết quả ngay khi từng cái xong, chứ không đợi cả lô. Giữ một List<Future> rồi lặp get theo thứ tự submit sẽ ép bạn chờ theo đúng thứ tự đó: task đầu chậm nhất chặn bạn, không cho chạm tới những task đã xong từ lâu phía sau.

Hãy tưởng tượng bạn gửi quần áo tới năm tiệm giặt. Bạn muốn lấy đồ ngay khi tiệm nào xong, chứ không đứng lì trước tiệm số 1 trong khi đồ ở tiệm số 4 đã giặt xong. CompletionService chính là cái quầy "ai xong trước trả trước" đó.

ExecutorService pool = Executors.newFixedThreadPool(4);
var ecs = new ExecutorCompletionService<Integer>(pool);
for (String region : regions) {
    ecs.submit(() -> querySeatCount(region));   // 5 truy van song song
}
int total = 0;
for (int i = 0; i < regions.size(); i++) {
    Future<Integer> done = ecs.take();          // lay task DA xong som nhat
    total += done.get();                        // get() o day khong con chan lau
}

ExecutorCompletionService bọc quanh executor và đẩy mỗi task vừa hoàn tất vào một hàng đợi nội bộ; take() chặn tới khi có kết quả bất kỳ rồi trả về đúng Future đó. Bạn tiêu thụ kết quả theo thứ tự hoàn tất thay vì submit, nên latency tổng gần với task chậm nhất chứ không phải tổng các lần chờ xếp chồng. Khi cần "lấy cái xong đầu tiên rồi bỏ phần còn lại", poll() không chặn cộng cancel các future còn lại cho bạn hành vi đó.

4. Vì sao Future thuần là chưa đủ?

Future giải quyết gọn bài toán "chạy một task rồi lấy kết quả". Nhưng đặt nó vào một hệ thống thật, hai giới hạn lộ ra ngay.

Thứ nhất, get là blocking. Mỗi lần gọi, một thread — thường đắt đỏ — bị ghim chỉ để ngồi chờ. Trong service xử lý nghìn request, mỗi request một thread chờ get là đốt đúng thứ tài nguyên mà thread pool sinh ra để tiết kiệm.

Thứ hai, và sâu hơn, Future không compose được. Thử diễn đạt một yêu cầu đời thường: "lấy thông tin user, rồi dựa trên đó gọi service tính giá, rồi dựa trên đó gửi xác nhận." Với Future thuần, bạn phải get kết quả bước một (chặn) mới có dữ liệu submit bước hai, rồi lại get (chặn), rồi mới submit bước ba. Chuỗi phụ thuộc lẽ ra chảy mượt như dây chuyền lại bị cắt vụn thành những lần chặn nối tiếp.

// Future thuan: moi buoc phu thuoc buoc truoc deu phai get() — chan roi moi di tiep
Future<User>    fu = pool.submit(() -> loadUser(id));
User user        = fu.get();                                 // chan (1)
Future<Price>   fp = pool.submit(() -> priceFor(user));
Price price      = fp.get();                                 // chan (2)
Future<Receipt> fr = pool.submit(() -> sendReceipt(user, price));
Receipt receipt  = fr.get();                                 // chan (3)

Future không có chỗ gắn callback "khi xong thì làm tiếp việc này" — chỉ cho hỏi "xong chưa?" rồi đứng chờ "cho tôi kết quả". Đúng khoảng trống này là lý do CompletableFuture ra đời.

5. 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.

5.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 8 sẽ mổ. Quy tắc thực dụng: production gần như luôn truyền executor riêng.

5.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ếu lỡ dùng thenApply, bạn nhận một CompletableFuture<CompletableFuture<U>> lồng hai lớp. thenCompose chính là flatMap của thế giới async: nó làm phẳng cái lồng đó — đúng như flatMap ở Optional/Stream.

// 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));

5.3 Stage chạy trên thread nào: ba biến thể và *Async

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 8); 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.

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

6.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".

6.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.

6.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".

7. 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 5.3 để 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.

Khi cần xử lý cả lô — 500 request đổ vào cùng một sự kiện — allOf gom lại để chờ tất cả mà không chặn từng cái, 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. 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.

8. 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(), vốn sizing theo số CPU core (mặc định khoảng số core - 1) 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 commonPool (kể cả parallel stream) bị vạ lây. Đây là lý do capstone luôn truyền bookingPool/notifyPool riêng.

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 Structured Concurrency & Scoped Values.

9. 📚 Deep Dive Oracle

10. Tổng kết

CompletableFuture thay Future blocking bằng pipeline khai báo — thenApply/thenCompose/thenCombine, allOf/anyOf, exceptionally/handle/orTimeout — gói trong ba luật: chọn executor cho *Async, đóng chuỗi bằng một chỗ tiêu thụ lỗi, và nhớ thread context không tự theo task qua các stage.

Cả Future lẫn CompletableFuture đều giả định task là những đơn vị công việc tương đối độc lập mà executor đem chạy. Nhưng có một lớp bài toán khác — divide-and-conquer trên dữ liệu lớn, thuần CPU — nơi một task tự chẻ mình thành task con đệ quy rồi gộp kết quả, cần cách lập lịch khác hẳn: thread nhàn rỗi "ăn cắp" việc của thread bận để cân tải giữa các core. Đó là Fork/Join, chủ đề bài kế tiếp.

11. Tự kiểm tra

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