[Advanced Java] CompletableFuture trong Java: lập trình bất đồng bộ, chain và xử lý lỗi

Future cho bạn biết rằng sẽ có một kết quả. Nó không cho bạn nói mình muốn làm gì với kết quả đó. get() block chính thread đang gọi, không có method nào để gắn thêm một bước tiếp theo, không có cách nào gộp hai future thành một, và cũng không có cách nào tạo ra một Future rồi tự tay complete nó sau. Bốn lỗ hổng riêng biệt, và cái nào cũng đẩy bạn quay về kiểu code block.

CompletableFuture có từ Java 8 và bịt kín cả bốn. Đây là một future để chain tiếp chứ không phải để ngồi đợi: bạn mô tả bước kế tiếp, bước sau nữa, và phải làm gì nếu có bước nào fail, rồi runtime chạy đúng chuỗi đó khi giá trị về. Bài này đi qua cách tạo một future, stage nào thực sự chạy trên thread nào, cách kết hợp nhiều future, cách một lỗi đi xuống chain, và những cái bẫy âm thầm làm mất kết quả.

Mọi dòng output và mọi message lỗi bên dưới đều lấy từ việc compile và chạy code thật trên OpenJDK 21.0.6 (arm64). Tên thread và thứ tự các stage vốn không xác định, nên mỗi output kiểu đó đều được ghi rõ và được lấy mẫu qua nhiều lần chạy. Bài viết không có bất kỳ con số thời gian hay throughput nào, vì số đo trên một máy đang chạy nhiều việc khác thì vô giá trị.

Vì sao Future là chưa đủ

Future cho bạn get(), isDone() và cancel(). Đó là một cái tay cầm để poll hoặc để block, và hết. Bạn không đăng ký được callback, nên đoạn code nào cần giá trị thì hoặc ngồi trên get(), hoặc quay vòng trên isDone(). Bạn không nói được "khi cả A và B xong thì gộp lại" mà không block một trong hai trước. Và vì một Future thường do chính executor chạy task sinh ra, bạn không tự tạo được một cái, trả cho caller, rồi complete nó từ chỗ mà câu trả lời thực sự xuất hiện — chẳng hạn một callback của thư viện network.

CompletableFuture implement Future, nên get() và cancel() vẫn dùng được, và bổ sung hai thứ ở trên đó: interface CompletionStage chứa toàn bộ từ vựng để nối các bước, và các method complete công khai (complete, completeExceptionally) cho phép bất kỳ ai kết thúc future từ bất kỳ đâu. Mọi thứ còn lại trong bài đều là hệ quả của hai bổ sung này.

Tạo một CompletableFuture

Bốn cách để có một future

Có bốn cách, và chúng phủ đúng bốn tình huống bạn gặp thật: chạy một task sinh giá trị trên thread khác, chạy một side effect trên thread khác, bọc một giá trị đã có sẵn, và tạo một future rỗng để người khác complete.

import java.util.concurrent.*;
 
public class Creating {
    static void p(String label) {
        System.out.println(label + " -> " + Thread.currentThread().getName());
    }
 
    public static void main(String[] args) throws Exception {
        System.out.println("availableProcessors = " + Runtime.getRuntime().availableProcessors());
        System.out.println("common pool parallelism = " + ForkJoinPool.commonPool().getParallelism());
        p("main");
 
        CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> {
            p("supplyAsync body");
            return "A";
        });
 
        ExecutorService io = Executors.newFixedThreadPool(2, r -> new Thread(r, "io-pool"));
        CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> {
            p("supplyAsync(io) body");
            return "B";
        }, io);
 
        CompletableFuture<Void> c = CompletableFuture.runAsync(() -> p("runAsync body"));
 
        CompletableFuture<String> d = CompletableFuture.completedFuture("D");
        System.out.println("completedFuture isDone=" + d.isDone());
 
        CompletableFuture<String> e = new CompletableFuture<>();
        System.out.println("manual isDone before complete = " + e.isDone());
        new Thread(() -> e.complete("E"), "completer").start();
 
        System.out.println("values: " + a.join() + b.join() + d.join() + e.join());
        c.join();
        io.shutdown();
    }
}

availableProcessors = 10
common pool parallelism = 9
main -> main
supplyAsync body -> ForkJoinPool.commonPool-worker-1
supplyAsync(io) body -> io-pool
runAsync body -> ForkJoinPool.commonPool-worker-1
completedFuture isDone=true
manual isDone before complete = false
values: ABDE

Số processor và do đó cả parallelism đều phụ thuộc máy, còn số hiệu worker thì không có gì bảo đảm — máy này báo mười processor và ra worker-1 cho cả hai body trong tám lần chạy được lấy mẫu, nhưng đó là quan sát chứ không phải luật. Phần còn lại của output thì ổn định.

Các cách tạo, và mỗi cách dùng cho việc gì:

Cái cuối cùng là cái nhiều người bỏ sót. Một CompletableFuture rỗng là một chỗ trống cho lời hứa: bạn trả nó về ngay lập tức, và đoạn code nào cuối cùng biết được câu trả lời — một handler của Netty, một JMS listener, một retry timer — sẽ gọi complete(value) lên nó. Mọi thư viện client bất đồng bộ trong hệ sinh thái đều trả kết quả cho bạn theo đúng kiểu này.

Body chạy trên thread nào

supplyAsync và runAsync khi không truyền executor sẽ dùng defaultExecutor(), và trên mọi máy có hơn một processor thì đó chính là common ForkJoinPool:

CompletableFuture<String> f = CompletableFuture.completedFuture("x");
System.out.println("is common pool? = " + (f.defaultExecutor() == ForkJoinPool.commonPool()));

is common pool? = true

Truyền một Executor vào là đổi được, đúng như dòng io-pool ở trên. Lựa chọn nhỏ đó lại là lựa chọn nặng ký nhất trong cả bài, và phần bẫy ở cuối sẽ quay lại giải thích vì sao.

Chain: thenApply, thenAccept và thenRun

Một chain được dựng từ ba dạng stage, khác nhau duy nhất ở hình dạng của function mà bạn truyền vào.

Ba dạng stage

Vẫn là ba dạng mà Optional và Stream dùng, chỉ khác là áp lên một giá trị chưa về. Mỗi method trả về một future mới đại diện cho stage bạn vừa thêm, điều này quan trọng hơn vẻ ngoài của nó và có hẳn một cái bẫy riêng ở phần cuối.

Stage nào chạy trên thread nào?

Đây là chỗ gần như mọi bài hướng dẫn đều phát biểu như một luật, mà nó không phải luật. Nếu không có hậu tố Async, JDK không hứa function của bạn chạy trên thread nào. Nó chỉ hứa function chạy sau khi source complete — và tùy vào việc source đã complete hay chưa lúc bạn đăng ký stage, thread đó có thể là thread hoàn tất source, hoặc chính thread gọi thenApply.

public class Chain {
    static void p(String stage) {
        System.out.println(stage + " -> " + Thread.currentThread().getName());
    }
 
    public static void main(String[] args) {
        p("main");
 
        System.out.println("-- case A: source still running when the stage is registered --");
        CompletableFuture.supplyAsync(() -> {
            sleep(200);
            p("  supplyAsync");
            return "raw";
        })
        .thenApply(v -> { p("  thenApply"); return v.toUpperCase(); })
        .thenAccept(v -> p("  thenAccept " + v))
        .thenRun(() -> p("  thenRun"))
        .join();
 
        System.out.println("-- case B: source already complete when the stage is registered --");
        CompletableFuture<String> done = CompletableFuture.completedFuture("raw");
        done.thenApply(v -> { p("  thenApply"); return v.toUpperCase(); })
            .thenAccept(v -> p("  thenAccept " + v))
            .join();
 
        System.out.println("-- case C: thenApplyAsync on the same already-complete future --");
        done.thenApplyAsync(v -> { p("  thenApplyAsync"); return v.toUpperCase(); })
            .thenAcceptAsync(v -> p("  thenAcceptAsync " + v))
            .join();
    }
 
    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

main -> main
-- case A: source still running when the stage is registered --
  supplyAsync -> ForkJoinPool.commonPool-worker-1
  thenApply -> ForkJoinPool.commonPool-worker-1
  thenAccept RAW -> ForkJoinPool.commonPool-worker-1
  thenRun -> ForkJoinPool.commonPool-worker-1
-- case B: source already complete when the stage is registered --
  thenApply -> main
  thenAccept RAW -> main
-- case C: thenApplyAsync on the same already-complete future --
  thenApplyAsync -> ForkJoinPool.commonPool-worker-1
  thenAcceptAsync RAW -> ForkJoinPool.commonPool-worker-1

Đây là output không xác định, nhưng giống hệt nhau trong ba lần chạy được lấy mẫu vì cái sleep 200 ms làm case A nghiêng hẳn về một phía. Ở case A source vẫn đang chạy lúc thenApply được đăng ký, nên thread của pool hoàn tất nó cũng chạy luôn mọi stage phía sau. Ở case B source đã complete, không còn gì để đợi, và main tự chạy các body ngay tại chỗ. Chỉ case C mới là một luật: thenApplyAsync luôn đẩy body sang một executor.

Bỏ cái sleep đi thì hai kết cục đua nhau ngay trong một lần chạy:

Map<String, Integer> tally = new TreeMap<>();
for (int i = 0; i < 20; i++) {
    CompletableFuture.supplyAsync(() -> "raw")
        .thenApply(v -> {
            String n = Thread.currentThread().getName();
            synchronized (tally) {
                tally.merge(n.startsWith("ForkJoin") ? "a pool worker" : n, 1, Integer::sum);
            }
            return v;
        })
        .join();
    spin(i * 20_000L);
}
System.out.println(tally);

{a pool worker=14, main=6}
{a pool worker=16, main=4}
{a pool worker=17, main=3}

Ba trong năm lần chạy được lấy mẫu; tỉ lệ thay đổi giữa các lần, từ 14/6 tới 17/3. Hai mươi lần đăng ký y hệt nhau, và thread nào chạy body được quyết định bởi việc task đã kịp được nhặt lên hay chưa. Qua năm lần chạy được lấy mẫu, có từ ba tới sáu trong hai mươi lần đăng ký rơi vào main. Viết code chỉ đúng khi stage chạy trên worker, hoặc chỉ đúng khi nó chạy trên caller, là bạn đã viết một bug xuất hiện ở một tỉ lệ không nhỏ các lời gọi và không bao giờ chịu hiện ra trong debugger.

Các biến thể Async

Method nào cũng có một bản sinh đôi Async, và bản đó có hai overload:

Dùng dạng thường cho các phép biến đổi rẻ và không block, khi bạn thật sự không quan tâm. Dùng Async kèm executor tường minh mỗi khi body có block, có đụng tài nguyên gắn với thread, hoặc phải tránh xa thread của caller. "Thường thì nó chạy trên worker" không phải là một thiết kế.

thenApply khác thenCompose thế nào

thenApply map một giá trị sang một giá trị. Khi chính function của bạn trả về một CompletableFuture — vì nó gọi tiếp một API bất đồng bộ khác — thenApply sẽ lồng chúng vào nhau, và compiler nói thẳng điều đó:

static CompletableFuture<String> loadUser(String id) {
    return CompletableFuture.supplyAsync(() -> "user:" + id);
}
 
CompletableFuture<String> id = CompletableFuture.supplyAsync(() -> "u-42");
CompletableFuture<String> nested = id.thenApply(v -> loadUser(v));

Compose.java:11: error: incompatible types: inference variable U has incompatible bounds
        CompletableFuture<String> nested = id.thenApply(v -> loadUser(v));
                                                       ^
    equality constraints: String
    lower bounds: CompletableFuture<String>
  where U,T are type-variables:
    U extends Object declared in method <U>thenApply(Function<? super T,? extends U>)
    T extends Object declared in class CompletableFuture
1 error

Khai báo đúng type mà nó thực sự có thì code compile được, và hình dạng của vấn đề lộ ra ngay:

CompletableFuture<CompletableFuture<String>> nested = id.thenApply(v -> loadUser(v));
System.out.println("thenApply  -> " + nested.join().getClass().getSimpleName()
                   + ", inner value " + nested.join().join());
 
CompletableFuture<String> flat = id.thenCompose(v -> loadUser(v));
System.out.println("thenCompose -> " + flat.join());

thenApply  -> CompletableFuture, inner value user:u-42
thenCompose -> user:u-42

thenCompose chính là flat map. Nó nhận Function<T, CompletionStage<U>> và trả về CompletableFuture<U>, phẳng một tầng, đúng như Optional.flatMap và Stream.flatMap làm với container của chúng. Quy tắc rất máy móc: function trả về giá trị thường thì dùng thenApply, function trả về future thì dùng thenCompose. Hai lời gọi join() lồng nhau là dấu hiệu cho thấy bạn đã chọn nhầm.

Kết hợp nhiều future độc lập

Chain nối những việc phụ thuộc vào bước trước. Composition ghép những việc không phụ thuộc nhau.

thenCombine cho hai future

thenCombine đợi cả hai bên rồi đưa cả hai giá trị vào một BiFunction. Cả hai future đều đã chạy sẵn, nên không có gì bị tuần tự hóa.

CompletableFuture<String> profile = fetch("profile", 150);
CompletableFuture<String> orders  = fetch("orders", 100);
 
String merged = profile.thenCombine(orders, (p, o) -> p + " + " + o).join();
System.out.println("thenCombine -> " + merged);

thenCombine -> profile + orders

Đây là cái duy nhất trong ba cái giữ nguyên type của bạn. thenAcceptBoth là bản Consumer và runAfterBoth là bản Runnable; applyToEither, acceptEither và runAfterEither là các biến thể "cái nào xong trước" cho đúng hai future.

allOf, anyOf và những return type khó dùng

Với nhiều hơn hai, các static factory nhận một mảng varargs — và cả hai đều trả về thứ bạn không dùng trực tiếp được.

List<CompletableFuture<String>> all = List.of(
    fetch("a", 120), fetch("b", 60), fetch("c", 30));
 
CompletableFuture<Void> gate = CompletableFuture.allOf(all.toArray(new CompletableFuture[0]));
System.out.println("allOf type    -> " + gate.getClass().getSimpleName()
                   + ", join() returns " + gate.join());
 
List<String> results = gate.thenApply(ignored ->
        all.stream().map(CompletableFuture::join).collect(Collectors.toList())).join();
System.out.println("collected     -> " + results);
 
CompletableFuture<Object> first = CompletableFuture.anyOf(
    fetch("slow", 200), fetch("fast", 20), fetch("medium", 100));
Object winner = first.join();
System.out.println("anyOf         -> " + winner + " (static type Object, runtime "
                   + winner.getClass().getName() + ")");

allOf type    -> CompletableFuture, join() returns null
collected     -> [a, b, c]
anyOf         -> fast (static type Object, runtime java.lang.String)

allOf trả về CompletableFuture<Void> vì các input của nó có thể khác type nhau, nên không có một kết quả chung nào để đưa cho bạn. Nó là một cái cổng chứ không phải một cái giỏ: nó báo mọi thứ đã xong và đưa bạn null. Cách lấy lại các giá trị là map trên chính list ban đầu ngay khi cổng mở — và các lời gọi join() trong đó không thể block, vì theo định nghĩa mọi source đều đã complete tại thời điểm đó.

anyOf trả về CompletableFuture<Object> cũng vì lý do đó nhưng theo chiều ngược lại: input nào cũng có thể thắng, nên type duy nhất chứa được tất cả là Object. Nếu các input cùng type thì bạn phải cast, và với compiler thì phép cast đó là unchecked dù thực tế nó an toàn.

⚠️ allOf đợi mọi input, kể cả những cái xong rất lâu sau lần fail đầu tiên. Nó không short-circuit.

Khi một input fail

Cả ba đều lan lỗi ra ngoài, nhưng không giống nhau.

try {
    ok("a", 50).thenCombine(bad("b", 20), (x, y) -> x + y).join();
} catch (CompletionException e) {
    System.out.println("thenCombine with a failing side -> " + e.getCause());
}
 
List<CompletableFuture<String>> all = List.of(ok("a", 30), bad("b", 10), ok("c", 20));
try {
    CompletableFuture.allOf(all.toArray(new CompletableFuture[0])).join();
} catch (CompletionException e) {
    System.out.println("allOf with one failure          -> " + e.getCause());
}
System.out.println("but the successful ones still hold values: a=" + all.get(0).join()
                   + " c=" + all.get(2).join());
 
try {
    CompletableFuture.anyOf(ok("slow", 300), bad("fast", 10)).join();
} catch (CompletionException e) {
    System.out.println("anyOf when the first to settle fails -> " + e.getCause());
}

thenCombine with a failing side -> java.lang.IllegalStateException: b failed
allOf with one failure          -> java.lang.IllegalStateException: b failed
but the successful ones still hold values: a=a c=c
anyOf when the first to settle fails -> java.lang.IllegalStateException: fast failed

Ổn định qua ba lần chạy được lấy mẫu, vì các sleep làm thứ tự nghiêng hẳn về một phía. Dòng đáng chú ý là dòng thứ ba: allOf fail không hề phá hủy những kết quả đã về. Các source future thành công vẫn complete và vẫn giữ giá trị của chúng, nên chiến lược lấy kết quả từng phần chỉ là join từng future kèm exceptionally của riêng nó, thay vì join cái cổng.

Lỗi trong một async chain

Exception ném ra bên trong một stage không nổi lên ngay tại chỗ nó bị ném. Stage bắt nó lại, lưu nó làm kết quả của future, và mọi stage phía dưới vốn chỉ biết xử lý giá trị đều bị bỏ qua.

Lỗi bỏ qua mọi stage cho tới khi gặp handler

CompletableFuture<String> f = CompletableFuture.<String>supplyAsync(() -> {
    System.out.println("stage 1 runs");
    throw new IllegalStateException("upstream broke");
})
.thenApply(v -> { System.out.println("stage 2 runs"); return v + "!"; })
.thenApply(v -> { System.out.println("stage 3 runs"); return v + "?"; })
.exceptionally(ex -> {
    System.out.println("exceptionally sees " + ex.getClass().getName()
                       + ": " + ex.getMessage());
    System.out.println("  cause is " + ex.getCause());
    return "fallback";
})
.thenApply(v -> { System.out.println("stage 4 runs"); return v.toUpperCase(); });
 
System.out.println("result = " + f.join());

stage 1 runs
exceptionally sees java.util.concurrent.CompletionException: java.lang.IllegalStateException: upstream broke
  cause is java.lang.IllegalStateException: upstream broke
stage 4 runs
result = FALLBACK

Stage 2 và 3 không in ra gì cả. Lỗi đi thẳng qua chúng, exceptionally biến nó trở lại thành một giá trị, rồi stage 4 — được đăng ký sau handler — chạy bình thường trên giá trị đó. Toàn bộ mô hình nằm ở đây: một chain mang theo hoặc một giá trị hoặc một lỗi, thenApply và họ hàng chỉ phản ứng với giá trị, còn các method recover chỉ phản ứng với lỗi.

exceptionally, handle và whenComplete

Ba method, ba việc, và cái thứ ba là cái hay bị dùng sai nhất.

System.out.println("handle on failure  -> " +
    broken().handle((v, ex) -> ex == null ? v : -1).join());
System.out.println("handle on success  -> " +
    fine().handle((v, ex) -> ex == null ? v : -1).join());
 
Integer observed = fine()
    .whenComplete((v, ex) -> System.out.println("whenComplete saw v=" + v + " ex=" + ex))
    .join();
System.out.println("whenComplete result unchanged -> " + observed);
 
CompletableFuture<Integer> stillBroken = broken()
    .whenComplete((v, ex) -> System.out.println("whenComplete saw v=" + v
                                                + " ex=" + (ex == null ? null : ex.getClass().getSimpleName())));
try {
    stillBroken.join();
} catch (CompletionException e) {
    System.out.println("whenComplete did NOT swallow it -> " + e.getCause());
}

handle on failure  -> -1
handle on success  -> 7
whenComplete saw v=7 ex=null
whenComplete result unchanged -> 7
whenComplete saw v=null ex=CompletionException
whenComplete did NOT swallow it -> java.lang.IllegalStateException: boom

whenComplete nhận một BiConsumer, nên nó không có giá trị trả về và không đổi được gì: future mà nó sinh ra mang đúng giá trị hoặc đúng lỗi của source. Nó dành cho log, metric và dọn dẹp, và nó là công cụ sai nếu ý bạn là recover — recover là việc của handle hoặc exceptionally.

Có đúng một ngoại lệ cho câu "quan sát mà không thay đổi". Nếu chính action trong whenComplete ném exception, exception đó trở thành kết quả của stage mới:

CompletableFuture<String> ok = CompletableFuture.completedFuture("fine");
try {
    ok.whenComplete((v, ex) -> { throw new RuntimeException("action failed"); }).join();
} catch (CompletionException e) {
    System.out.println("whenComplete action threw -> " + e.getCause());
}
System.out.println("source future untouched   -> " + ok.join());

whenComplete action threw -> java.lang.RuntimeException: action failed
source future untouched   -> fine

Source không hề bị đụng tới vì mọi stage đều trả về một future mới. Chỉ cái stage bạn dựng từ whenComplete mới mang lỗi mới.

completeExceptionally, và handler thực sự nhận được gì

completeExceptionally là bản đối ứng thất bại của complete: nó kết thúc một future chưa xong bằng một Throwable bạn đưa vào, và đó chính là thứ một thư viện client bất đồng bộ gọi khi callback của nó báo lỗi.

Con đường đó cũng tạo ra một khác biệt mà gần như mọi bài hướng dẫn đều nói sai. Khi exception được ném bên trong body của một stage, JDK bọc nó trong một CompletionException trước khi lưu. Khi nó được set trực tiếp bằng completeExceptionally, hoặc do orTimeout sinh ra, nó được lưu nguyên xi. Handler nhận đúng cái đang được lưu:

static void show(String label, CompletableFuture<String> f) {
    f.exceptionally(ex -> {
        System.out.println(label);
        System.out.println("  exceptionally received : " + ex.getClass().getName());
        System.out.println("  its getCause()         : " + ex.getCause());
        return "recovered";
    }).join();
}

A: thrown inside supplyAsync
  exceptionally received : java.util.concurrent.CompletionException
  its getCause()         : java.lang.IllegalStateException: boom
B: set with completeExceptionally
  exceptionally received : java.lang.IllegalStateException
  its getCause()         : null
C: orTimeout fired
  exceptionally received : java.util.concurrent.TimeoutException
  its getCause()         : null

Vậy nên gọi ex.getCause() bên trong handler là một bug chờ sẵn: nó là exception gốc ở case A và là null ở case B với C. handle và whenComplete hành xử y hệt. Hãy unwrap phòng thủ thay vì tin vào một dạng:

static Throwable root(Throwable ex) {
    return (ex instanceof CompletionException && ex.getCause() != null) ? ex.getCause() : ex;
}

handle raw       : CompletionException | unwrapped: IllegalStateException
whenComplete raw : CompletionException | unwrapped: IllegalStateException
handle raw       : IllegalStateException | unwrapped: IllegalStateException
whenComplete raw : IllegalStateException | unwrapped: IllegalStateException

Cùng một handler, cả hai nguồn lỗi, một câu trả lời duy nhất.

join ném CompletionException, get ném ExecutionException

Block ở cuối chain có hai dạng, và chúng khác nhau đúng ở một điểm đáng kể: thứ chúng ném ra là checked hay unchecked.

CompletableFuture<String> manual = new CompletableFuture<>();
boolean accepted = manual.completeExceptionally(new java.io.IOException("socket closed"));
System.out.println("completeExceptionally accepted = " + accepted
                   + ", isCompletedExceptionally = " + manual.isCompletedExceptionally());
 
try {
    manual.join();
} catch (CompletionException e) {
    System.out.println("join threw   " + e.getClass().getName());
    System.out.println("  getCause() " + e.getCause());
}
 
try {
    manual.get();
} catch (ExecutionException e) {
    System.out.println("get threw    " + e.getClass().getName());
    System.out.println("  getCause() " + e.getCause());
}

completeExceptionally accepted = true, isCompletedExceptionally = true
join threw   java.util.concurrent.CompletionException
  getCause() java.io.IOException: socket closed
get threw    java.util.concurrent.ExecutionException
  getCause() java.io.IOException: socket closed

join() ném CompletionException là unchecked, nên nó dùng được bên trong lambda mà không cần khối try. get() ném ExecutionException là checked, cộng thêm InterruptedException, nên compiler bắt bạn xử lý cả hai. Cả hai đều giữ lỗi gốc trong getCause(). Để ý là việc bọc vẫn xảy ra ngay cả ở đây, nơi future được complete bằng một IOException thô — wrapper do lời gọi block thêm vào, không phải do chỗ lưu trữ.

Để một join() fail lọt ra ngoài mà không bắt thì bạn nhận được thứ này, rất đáng đọc kỹ:

Exception in thread "main" java.util.concurrent.CompletionException: java.lang.IllegalArgumentException: bad id
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1770)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1760)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:387)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1312)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1843)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1808)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:188)
Caused by: java.lang.IllegalArgumentException: bad id
	at Wrapping.lambda$main$0(Wrapping.java:25)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768)
	... 6 more

Dòng đầu ghi main, nhưng mọi frame ở phần trên đều thuộc về thread worker của pool đã chạy supplier. Dòng code gọi join() không xuất hiện ở đâu cả. Đó là cái giá lớn nhất về mặt trải nghiệm của code bất đồng bộ: stack bạn nhận được là stack của thread bị lỗi, không phải stack của đoạn code đang đợi. Phần Caused by mới là phần duy nhất chỉ đúng vào lambda của bạn.

Checked exception không thoát ra khỏi supplier được

Supplier.get() không khai báo checked exception nào, nên body của supplier mà gọi thứ gì ném checked exception thì không compile được:

static String readFile() throws IOException { throw new IOException("disk gone"); }
 
CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> readFile());

Checked.java:8: error: unreported exception IOException; must be caught or declared to be thrown
        CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> readFile());
                                                                                  ^
1 error

Bạn tự bọc nó trong một CompletionException. JDK không bọc lại một CompletionException được ném từ body, nên lỗi gốc nằm đúng một tầng bên dưới:

CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
    try {
        return readFile();
    } catch (IOException e) {
        throw new CompletionException(e);
    }
});

received  : java.util.concurrent.CompletionException
getCause(): java.io.IOException: disk gone
result = fallback

orTimeout và completeOnTimeout

Java 9 thêm hai method timeout, và cả hai đều có mặt trên JDK 21. orTimeout làm future fail bằng TimeoutException nếu nó chưa complete kịp; completeOnTimeout thì complete nó bằng một giá trị dự phòng.

try {
    slowCall().orTimeout(200, TimeUnit.MILLISECONDS).join();
} catch (CompletionException e) {
    System.out.println("orTimeout        -> " + e.getCause());
}
 
String v = slowCall()
    .completeOnTimeout("cached answer", 200, TimeUnit.MILLISECONDS)
    .join();
System.out.println("completeOnTimeout -> " + v);
 
String fast = CompletableFuture.supplyAsync(() -> "quick")
    .completeOnTimeout("fallback", 500, TimeUnit.MILLISECONDS)
    .join();
System.out.println("no timeout needed -> " + fast);

orTimeout        -> java.util.concurrent.TimeoutException
completeOnTimeout -> cached answer
no timeout needed -> quick

Ở đây slowCall() là một supplyAsync có supplier sleep hai giây. Có hai điểm cần nói cho chính xác. Thứ nhất, TimeoutException từ orTimeout tới tay handler ở dạng không bọc, đúng như bảng ở phần trước, nên ex.getCause() trên nó là null — hãy kiểm tra bằng instanceof TimeoutException trên chính exception đó. Thứ hai, không method nào dừng công việc đang chạy:

CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
    System.out.println("supplier started");
    sleep(600);
    System.out.println("supplier finished its work anyway");
    return "done";
});
 
sleep(100);
System.out.println("cancel(true) returned " + f.cancel(true));
try {
    f.join();
} catch (CancellationException e) {
    System.out.println("join threw " + e.getClass().getName());
}
sleep(900);
System.out.println("main exits");

supplier started
cancel(true) returned true
join threw java.util.concurrent.CancellationException
supplier finished its work anyway
main exits

Giống hệt nhau qua ba lần chạy được lấy mẫu. cancel(true) complete cái future bằng một CancellationException — để ý là join() ném thẳng exception đó chứ không bọc trong CompletionException — nhưng tham số true được tài liệu ghi rõ là không có tác dụng, và supplier vẫn chạy tới hết. Timeout hay cancel là lời hứa về việc bạn sẽ đợi bao lâu, không bao giờ là cách dừng công việc đã bắt đầu.

Những cái bẫy

Future trả về là một future mới

Mọi method nối stage đều trả về một stage mới. Bỏ qua giá trị trả về đó là bug CompletableFuture phổ biến nhất, vì chẳng có gì cảnh báo bạn:

CompletableFuture<String> base = CompletableFuture.completedFuture("hello");
base.thenApply(String::toUpperCase);              // return value dropped
System.out.println("base.join() = " + base.join());
 
CompletableFuture<String> upper = base.thenApply(String::toUpperCase);
System.out.println("upper.join() = " + upper.join());
System.out.println("same object? " + (base == upper));

base.join() = hello
upper.join() = HELLO
same object? false

Function có chạy thật — chỉ là nó ghi kết quả vào một future không ai giữ lại. Đây vẫn là kỷ luật immutable quen thuộc của String.replace hay Stream.filter: object bạn gọi method lên không đổi, và câu trả lời nằm ở giá trị trả về.

Block bên trong một task của pool

Block một thread của pool để đợi một task khác cũng từ pool đó là cách một executor tự treo mình: với executor một thread, worker duy nhất nằm im trong join() đợi một task không bao giờ được lên lịch, vì chính worker đó mới là thứ phải chạy nó. Bài 22 nói về kiểu hỏng đó và cách nhận ra nó. Phần thuộc về bài này là chuyện common ForkJoinPool lại không hành xử như vậy, và biết chính xác khác biệt đó thì hơn hẳn nghe truyền miệng.

CompletableFuture block thông qua ForkJoinPool.managedBlock, cơ chế cho phép pool khởi động thêm thread bù, nên một task của common pool block trên join() không làm nghẽn pool:

ForkJoinPool cp = ForkJoinPool.commonPool();
int p = cp.getParallelism();
System.out.println("parallelism = " + p);
 
CompletableFuture<String> gate = new CompletableFuture<>();
List<CompletableFuture<Void>> tasks = new ArrayList<>();
for (int i = 0; i < p + 4; i++) tasks.add(CompletableFuture.runAsync(gate::join));
 
Thread.sleep(500);
System.out.println(tasks.size() + " tasks blocked in join(): poolSize = " + cp.getPoolSize());
gate.complete("go");
CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0])).join();

parallelism = 9
13 tasks blocked in join(): poolSize = 14

Ba lần chạy được lấy mẫu, kết quả giống nhau trên máy mười processor này. Pool đã nở ra vượt parallelism để tiếp tục làm việc. Cơ chế bù đó có giới hạn — java.util.concurrent.ForkJoinPool.common.maximumSpares mặc định là 256 — và nó không áp dụng cho kiểu block thông thường. Đổi sang sleep hoặc socket I/O thì pool không nở thêm chút nào:

parallelism = 9
13 tasks queued, blocked in Thread.sleep(): poolSize = 9

Cũng ba lần chạy được lấy mẫu. Pool không bao giờ vượt quá chín thread, nên nhiều nhất chỉ chín trong mười ba task chạy cùng lúc, số còn lại nằm trong queue cho tới khi có worker rảnh. Đó là phiên bản thô của mối nguy này, và cũng là thứ bạn gặp thật: một pool bận kín mà chẳng làm gì.

Common pool dùng chung cho cả process

ForkJoinPool.commonPool() là một pool duy nhất cho mỗi JVM, kích thước bằng availableProcessors() - 1. Parallel stream dùng nó. Mọi thư viện trong cây dependency của bạn mà gọi supplyAsync không kèm executor cũng dùng nó. Mười lời gọi HTTP đang nằm đợi trên một pool rộng chín nghĩa là một parallel stream ở chỗ khác trong process giờ phải xếp hàng sau chúng:

ExecutorService io = Executors.newFixedThreadPool(16, r -> new Thread(r, "io"));
CompletableFuture.supplyAsync(this::callTheApi, io);

Nguyên tắc thực dụng: common pool chỉ dành cho việc CPU-bound và ngắn. Cái gì có block thì được executor riêng, kích thước tính theo số việc có thể đang đợi chứ không theo số core, và được shutdown khi ứng dụng dừng.

Chain kiểu fire-and-forget nuốt mất lỗi

Nếu không ai quan sát một future thì lỗi của nó được lưu lại rồi rơi vào quên lãng. Ở đây không có thứ gì tương đương uncaught-exception handler:

CompletableFuture.supplyAsync(() -> { throw new IllegalStateException("nobody sees me"); })
                 .thenApply(v -> v);
Thread.sleep(300);
System.out.println("main finished normally, no stack trace printed above");

main finished normally, no stack trace printed above

Exit code 0, không output, không stack trace. Exception nằm trong một CompletableFuture mà không ai giữ reference, và nó bị thu hồi cùng với future đó. Mọi chain phải kết thúc ở một thứ quan sát được kết cục: một join(), một get(), một exceptionally, một handle, hoặc tối thiểu là một whenComplete có ghi log. Kết thúc chain bằng thenAccept rồi vứt kết quả đi chính là cách những lỗi biến mất trên production.

Ghép lại thành một pipeline

Một hình dạng sát thực tế: tra id, rồi lấy hai thứ độc lập nhau bằng id đó, gộp lại, chặn toàn bộ bằng một timeout, và recover thay vì ném lỗi. Để ý mọi lời gọi async đều nhận executor IO tường minh, và chương trình shutdown executor đó rồi đợi nó, nên JVM luôn kết thúc.

static final AtomicInteger SEQ = new AtomicInteger();
static final ExecutorService IO =
    Executors.newFixedThreadPool(4, r -> new Thread(r, "io-" + SEQ.incrementAndGet()));
 
public static void main(String[] args) throws Exception {
    String page = findUserId("a@example.com")
        .thenCompose(id -> loadProfile(id).thenCombine(countOrders(id),
                (profile, orders) -> profile + " with " + orders + " orders"))
        .orTimeout(2, TimeUnit.SECONDS)
        .exceptionally(ex -> "unavailable: " + ex)
        .join();
 
    System.out.println(page);
 
    IO.shutdown();
    System.out.println("pool terminated = " + IO.awaitTermination(5, TimeUnit.SECONDS));
}

profile(u-42) with 3 orders
pool terminated = true

thenCompose vì findUserId dẫn tiếp vào một lời gọi async; thenCombine vì profile và số order không phụ thuộc nhau; orTimeout đặt trên future đã ghép để giới hạn phủ cả thao tác chứ không chỉ một nhánh; exceptionally đặt cuối cùng để nó bắt lỗi từ mọi stage phía trên. Đúng một lời gọi block, ở tận cuối, trên một thread không còn việc gì khác.

FAQ

join() và get() khác nhau ở đâu?

Cả hai đều block cho tới khi future settle. join() ném CompletionException là unchecked, nên gọi được bên trong lambda mà không cần khối try; get() ném ExecutionException là checked cộng thêm InterruptedException, nên compiler bắt bạn xử lý. Cả hai đều đặt lỗi gốc vào getCause(). Dùng join() bên trong chain, và dùng get() khi bạn cần timeout, vì chỉ get mới có overload get(timeout, unit).

CompletableFuture có tạo thread mới không?

Không. supplyAsync và runAsync đẩy một task sang executor — mặc định là common ForkJoinPool, hoặc cái bạn truyền vào. completedFuture và new CompletableFuture<>() không chạy gì hết. Việc tạo thread là chuyện của executor, không phải của future.

Khi nào nên dùng thenCompose thay cho thenApply?

Mỗi khi function của bạn trả về một CompletableFuture hay bất kỳ CompletionStage nào. thenApply sẽ cho bạn CompletableFuture<CompletableFuture<T>>; thenCompose làm phẳng thành CompletableFuture<T>. Đúng quan hệ giữa flatMap và map trên Optional và Stream.

Vì sao exception của tôi biến mất?

Vì không ai quan sát future đó. Lỗi được lưu trong future và chỉ được báo ra khi có người gọi join, get, exceptionally, handle hoặc whenComplete trên chain đó. Một chain kết thúc bằng thenAccept bị vứt đi thì không báo gì và không in stack trace nào, đúng như phần trên đã minh họa.

Có nên chạy lời gọi database hay HTTP trên common pool không?

Không. Đó là một pool duy nhất cho cả JVM, kích thước bằng availableProcessors() - 1, và dùng chung với parallel stream cùng mọi thư viện gọi supplyAsync không kèm executor. Việc có block thì thuộc về executor riêng của bạn, kích thước tính theo số thao tác có thể đang đợi chứ không theo số core.

thenApplyAsync có luôn an toàn hơn thenApply không?

An toàn hơn ở một khía cạnh, nhưng không miễn phí. Async bảo đảm body được đẩy sang executor thay vì có thể chạy trên thread của caller, và đó là thứ bạn cần cho bất cứ gì có block hoặc nhạy cảm với thread. Nó cũng tốn một lần submit task cho mỗi stage, nên một chain mười bước map tí hon không cần mười lần như vậy. Dùng Async ở chỗ thread quan trọng, và dùng dạng thường ở chỗ nó thật sự không quan trọng.

Làm sao lấy được list kết quả từ allOf?

allOf trả về CompletableFuture<Void>, nên hãy giữ lại list ban đầu và map trên nó ngay khi cổng mở: allOf(array).thenApply(v -> list.stream().map(CompletableFuture::join).toList()). Các lời gọi join() ở đó không thể block, vì mọi source future đều đã complete trước khi cổng complete.

Có hủy được công việc phía sau một CompletableFuture không?

Không hẳn. cancel(true) complete future bằng một CancellationException, còn cờ mayInterruptIfRunning được tài liệu ghi là không có tác dụng — supplier đang chạy không bị interrupt và, như output ở trên cho thấy, nó chạy tới hết. orTimeout cũng vậy. Nếu cần hủy thật, hãy giữ lại Future gốc từ executor, hoặc để task tự poll một cờ do bạn quản lý.

Kết luận

CompletableFuture là một Future cho phép bạn mô tả việc gì xảy ra tiếp theo thay vì ngồi đợi việc đã xảy ra. supplyAsync và runAsync khởi chạy công việc trên thread khác, completedFuture cùng cách complete thủ công phủ những trường hợp bạn đã có câu trả lời hoặc người khác sẽ đưa nó cho bạn, còn thenApply, thenAccept và thenRun thêm các stage chạy khi giá trị về. thenCompose là flat map cho những function trả về future; thenCombine, allOf và anyOf ghép các phần việc độc lập, với hai trong ba trả về type mà bạn phải tìm cách đi vòng.

Những phần đáng nhớ nhất lại là những phần API không quảng cáo. Không có hậu tố Async, thread chạy một stage được quyết định bởi cuộc đua giữa thread hoàn tất source và thread đang gọi, và bảng đếm ở trên thay đổi giữa các lần chạy. Một lỗi bỏ qua mọi stage xử lý giá trị cho tới khi gặp handler, và thứ handler nhận được là CompletionException khi exception bị ném trong body của stage, nhưng là exception thô khi nó được set bằng completeExceptionally hoặc do orTimeout sinh ra. join bọc trong CompletionException còn get bọc trong ExecutionException, cả hai đều giữ lỗi gốc ở getCause(). Và một chain không ai quan sát thì vứt lỗi đi trong im lặng, nên chain nào cũng cần một điểm kết thúc.

Mối nguy còn lại chính là thứ mà phần về pool ở trên mới chỉ chỉ vào: hai thread, mỗi bên giữ đúng thứ bên kia cần, và cả hai đợi mãi mãi. Bài 22 nói về deadlock và livelock — chúng phát sinh thế nào, nhận ra chúng trong một thread dump ra sao, cùng những kỷ luật về thứ tự khóa và timeout giúp tránh chúng.