Module 04 · Executable

CompletableFuture & Async Composition

Học cách compose computation bất đồng bộ mà không biến pipeline thành chuỗi get()/join() blocking.

1. Mục tiêu

  • Hiểu Future là “handle cho kết quả chưa sẵn sàng”.
  • Thấy vì sao Future.get() dễ kéo code trở lại blocking.
  • Dùng CompletionStage để compose thay vì chờ.
  • Nắm chắc thenApply vs thenCompose.
  • Hiểu thread/executor semantics của continuation.

2. Future: có async nhưng chưa compose tốt

ExecutorService pool = Executors.newFixedThreadPool(4);

Future<User> future = pool.submit(() -> repository.loadUser(42));

User user = future.get(); // block nếu chưa hoàn thành

Future cho biết computation có thể chưa xong, hỗ trợ cancellation và lấy kết quả; nhưng API cổ điển khiến ta thường phải gọi get(), tức chờ.

3. CompletableFuture

CompletableFuture<User> userFuture =
    CompletableFuture.supplyAsync(() -> repository.loadUser(42));

userFuture
    .thenApply(User::name)
    .thenAccept(System.out::println);

Thay vì “đợi rồi làm tiếp”, ta mô tả stage tiếp theo sẽ chạy khi stage trước hoàn thành.

4. thenApply vs thenCompose

thenApply — synchronous transform của value

CompletableFuture<User> user = findUserAsync(42);

CompletableFuture<String> name =
    user.thenApply(User::name);

thenCompose — transform trả về một async stage khác

CompletableFuture<Order> order =
    findUserAsync(42)
        .thenCompose(user -> findLatestOrderAsync(user.id()));

Sai mental model

CompletableFuture<CompletableFuture<Order>> nested =
    findUserAsync(42)
        .thenApply(user -> findLatestOrderAsync(user.id()));
Cầu nối tương lai: thenCompose ↔ Reactor flatMap ↔ Vert.x compose.

5. Chạy các async task độc lập

CompletableFuture<User> user = findUserAsync(id);
CompletableFuture<List<Order>> orders = findOrdersAsync(id);

CompletableFuture<UserProfile> profile =
    user.thenCombine(orders, UserProfile::new);

thenCombine phù hợp khi hai stage độc lập và ta cần cả hai kết quả.

allOf

CompletableFuture<Void> all =
    CompletableFuture.allOf(a, b, c);

allOf hoàn tất khi tất cả futures hoàn tất nhưng trả Void; cần lấy result từ từng future sau khi aggregate hoàn thành.

6. Error handling

findUserAsync(id)
    .thenCompose(this::loadOrdersAsync)
    .exceptionally(ex -> {
        log.error("flow failed", ex);
        return List.of();
    });
APIDùng khi
exceptionallyRecover từ exception thành value.
handleMuốn xử lý cả success và failure và transform kết quả.
whenCompleteSide effect quan sát success/failure, thường không thay đổi result.

7. Continuation chạy thread nào?

future.thenApply(v -> transform(v));

Non-async dependent action có thể chạy trên thread hoàn tất current future hoặc thread khác đang gọi completion method.

future.thenApplyAsync(v -> transform(v));

Async variant dùng executor mặc định của CompletableFuture nếu không truyền executor riêng; với các static async factory thông thường đây thường là ForkJoinPool.commonPool() khi environment cho phép.

future.thenApplyAsync(this::transform, myExecutor);
Không “Async” vô tội vạ. Mỗi boundary có scheduling cost. Hãy dùng executor rõ ràng cho blocking/CPU workloads khi kiến trúc yêu cầu.

8. Timeout và cancellation

CompletableFuture<User> result =
    findUserAsync(id)
        .orTimeout(2, TimeUnit.SECONDS);
CompletableFuture<User> fallback =
    findUserAsync(id)
        .completeOnTimeout(User.unknown(), 2, TimeUnit.SECONDS);

Timeout của wrapper không đồng nghĩa downstream I/O chắc chắn đã bị hủy. Cancellation cần được propagate tới resource/API cụ thể nếu muốn tiết kiệm công việc thật sự.

9. End-to-end async flow

CompletableFuture<OrderResponse> createOrder(CreateOrder cmd) {
    return findUserAsync(cmd.userId())
        .thenCompose(user -> checkInventoryAsync(cmd)
            .thenCombine(
                loadPricingAsync(cmd),
                (inventory, pricing) ->
                    new ValidatedOrder(user, inventory, pricing)
            ))
        .thenCompose(this::chargePaymentAsync)
        .thenCompose(this::saveOrderAsync)
        .thenApply(OrderResponse::from)
        .orTimeout(3, TimeUnit.SECONDS);
}

Điểm cần quan sát: không có get()/join() trong flow.

10. Lab

Xây aggregator gọi song song User Service và Order Service giả lập bằng ScheduledExecutorService. Yêu cầu:

  1. Không dùng get() giữa pipeline.
  2. Dùng thenCombine cho hai call độc lập.
  3. Timeout toàn flow ở 1500 ms.
  4. Nếu orders fail thì trả danh sách rỗng, nhưng nếu user fail thì fail cả request.
  5. In thread name ở mỗi stage để quan sát execution.

11. Bài tập

D1. Refactor nested CompletableFuture<CompletableFuture<T>> thành flat flow.
Dùng thenCompose ở stage trả Future.
D2. Cho ba API A/B/C: B phụ thuộc A, C độc lập. Hãy thiết kế flow tối ưu concurrency.
D3. Viết ví dụ chứng minh thenApply có thể chạy trên thread hoàn tất future trước.
D4. Giải thích tại sao gọi join() ngay sau supplyAsync() có thể làm mất lợi ích composition.

12. Interview checkpoint

  1. Future và CompletableFuture khác nhau thế nào?
  2. thenApply vs thenCompose?
  3. thenCombine khác allOf?
  4. exceptionally vs handle vs whenComplete?
  5. Async continuation chạy executor nào?
  6. Timeout có tự hủy network call không?
  7. Tại sao CompletableFuture là bước đệm tốt trước Reactor?

Tài liệu và code thực hành

Đọc chapter chi tiết [legacy source: modules/04-completable-future.md] · Mở foundation labs [legacy source: labs/foundation/README.md] · Xem Java source [legacy source: examples/README.md]

← Module 03Module 05 — NIO/Event Loop →