CompletableFuture: gọi nhiều API song song mà không block

Đây là bài toán kinh điển để học CompletableFuture. Mình tiếp tục với project xuyên suốt: Order Service chạy Spring Boot trên Tomcat, có API checkout.

1. Bài toán

Client → Tomcat (order-service) → GET /checkout/42
                                     |-- userClient.get(userId)        1s
                                     |-- orderClient.getCart(userId)   1s
                                     |-- paymentClient.methods(userId) 1s

Code hiện tại:

User user        = userClient.get(userId);          // 1s
Cart cart        = orderClient.getCart(userId);     // 1s
PaymentInfo pay  = paymentClient.methods(userId);   // 1s
return new CheckoutView(user, cart, pay);           // tổng 3s

Ba lời gọi này không phụ thuộc nhau, nhưng lại chạy tuần tự. Tomcat thread phần lớn thời gian chỉ ngồi chờ mạng. Nếu chạy song song, tổng thời gian ≈ lời gọi chậm nhất, tức khoảng 1 giây.

2. Vấn đề: Future cũ chưa đủ

Java 5 đã có ExecutorService + Future, và nó chạy song song thật:

Future<User> uf        = executor.submit(() -> userClient.get(userId));
Future<Cart> cf        = executor.submit(() -> orderClient.getCart(userId));
Future<PaymentInfo> pf = executor.submit(() -> paymentClient.methods(userId));

User user = uf.get();        // block tới khi xong
Cart cart = cf.get();
PaymentInfo pay = pf.get();  // tổng ≈ 1s

Lỗi hay gặp: gọi get() ngay sau mỗi submit. Khi đó task sau chỉ được submit khi task trước xong, quay về tuần tự. Phải submit hết trước, get sau. Exception trong task được giữ trong Future, đến get() mới lộ ra dưới dạng ExecutionException.

Hạn chế của Future:

• get() block thread gọi. Muốn làm gì tiếp với kết quả thì phải chờ.
• Không nối được bước tiếp theo kiểu "xong thì làm X".
• Không gộp được nhiều future, không có xử lý lỗi kiểu callback, không có timeout dạng khai báo.

3. Giải pháp: CompletableFuture

CompletableFuture (Java 8) cho bạn mô tả một pipeline: chạy async, xong thì biến đổi, nối, gộp, xử lý lỗi.

4. Code từng bước

Bước 1: Executor riêng cho IO. Không truyền executor thì supplyAsync chạy trên ForkJoinPool.commonPool(): số thread ≈ số CPU − 1 (máy 4 core chỉ 3 thread) và dùng chung cho cả JVM (parallel stream cũng dùng). Nhét lời gọi HTTP/DB vào đó, 3 thread bị chiếm là mọi thứ khác đứng theo.

@Configuration
class AsyncConfig {
    @Bean(destroyMethod = "shutdown")
    ExecutorService ioPool() {
        return new ThreadPoolExecutor(
                20, 50, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(500),               // queue có giới hạn
                new ThreadPoolExecutor.CallerRunsPolicy());  // đầy thì thread gọi tự chạy
    }
}

Bước 2: Timeout ở HTTP client (application.yml). Timeout của future không hủy được socket bên dưới, nên client phải có timeout riêng:

spring:
  cloud:
    openfeign:
      client:
        config:
          default:
            connectTimeout: 500
            readTimeout: 1500

Bước 3: Service

@Service
public class CheckoutService {
    private final UserClient userClient;
    private final OrderClient orderClient;
    private final PaymentClient paymentClient;
    private final ExecutorService ioPool;

    // constructor injection bỏ qua cho gọn

    public CheckoutView checkout(Long userId) {
        var uf = CompletableFuture
                .supplyAsync(() -> userClient.get(userId), ioPool)
                .orTimeout(1, TimeUnit.SECONDS);                     // bắt buộc có user

        var cf = CompletableFuture
                .supplyAsync(() -> orderClient.getCart(userId), ioPool)
                .orTimeout(1, TimeUnit.SECONDS);

        var pf = CompletableFuture
                .supplyAsync(() -> paymentClient.methods(userId), ioPool)
                .completeOnTimeout(PaymentInfo.defaultOnly(), 800, TimeUnit.MILLISECONDS)
                .exceptionally(ex -> {                               // payment lỗi: vẫn hiện trang
                    log.warn("payment methods lỗi, dùng mặc định", ex);
                    return PaymentInfo.defaultOnly();
                });

        return CompletableFuture.allOf(uf, cf, pf)
                .thenApply(v -> new CheckoutView(uf.join(), cf.join(), pf.join()))
                .join();                                             // ~1s thay vì 3s
    }

    // thenCompose: bước sau cần kết quả bước trước
    public CompletableFuture<List<Order>> ordersOf(Long userId) {
        return CompletableFuture.supplyAsync(() -> userClient.get(userId), ioPool)
                .thenCompose(user -> orderClient.findAsync(user.id()));   // không bị CF lồng CF
    }

    // thenCombine: 2 việc độc lập rồi gộp
    public CompletableFuture<Summary> summary(Long userId) {
        var u = CompletableFuture.supplyAsync(() -> userClient.get(userId), ioPool);
        var b = CompletableFuture.supplyAsync(() -> paymentClient.balance(userId), ioPool);
        return u.thenCombine(b, Summary::new);
    }
}

• exceptionally: payment chết thì trang vẫn hiện, chỉ thiếu phương thức thanh toán. Đây là graceful degradation.
• join() ném CompletionException (unchecked), get() ném ExecutionException/InterruptedException (checked). Lỗi gốc nằm ở getCause().
• Đừng gọi get()/join() giữa các bước pipeline; chỉ join ở cuối (hoặc trả thẳng CompletableFuture cho Spring MVC).

Bước 4: Test bằng curl

time curl -s http://localhost:8080/checkout/42
# real 0m1.05s   (trước đây 3.0s)

# Tắt payment-service rồi gọi lại:
curl -s http://localhost:8080/checkout/42
# {"user":{...},"cart":{...},"payment":{"methods":["COD"]}}   ← fallback

5. Cách hoạt động

Tomcat thread (http-nio-8080-exec-1)
   |
   |-- supplyAsync(user) ----> ioPool-1  [==== 1s ====]--+
   |-- supplyAsync(cart) ----> ioPool-2  [==== 1s ====]--+--> allOf xong
   |-- supplyAsync(pay)  ----> ioPool-3  [=== 0.8s ==]---+        |
   |                                                              v
   |-- join() chờ ............................................ thenApply(CheckoutView)
   |
   v  response sau ≈ 1s

Tuần tự:   [user 1s][cart 1s][pay 1s]       = 3s
Song song: [user 1s]
           [cart 1s]
           [pay  1s]                        = 1s

Lưu ý: Tomcat thread vẫn join() chờ. Cái ta được là latency giảm (3s → 1s), chứ chưa giải phóng Tomcat thread.

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: một request cần gọi nhiều service độc lập; việc nền cần nối bước tiếp khi xong (gửi mail xong thì lưu lịch sử).
• Không nên: các bước phụ thuộc nhau hoàn toàn (song song không được gì); việc CPU rất nhỏ (overhead lớn hơn lợi ích); dùng commonPool cho IO.
• Luôn có timeout ở HTTP client. orTimeout chỉ kết thúc future, thread bên dưới vẫn chờ socket.

Hai câu này hay bị hiểu nhầm. Về thread:

• Bản không Async (thenApply, thenCombine): chạy trên thread vừa hoàn tất future trước đó, hoặc chính thread đăng ký nếu future đã xong sẵn. Không nên giả định chính xác là thread nào.
• Bản Async (thenApplyAsync(fn, executor)): luôn submit sang executor (mặc định commonPool nếu không truyền).

CompletableFuture.supplyAsync(() -> 10, ioPool)
        .thenApply(n -> n * 2)                        // thread hoàn tất bước trước
        .thenApplyAsync(n -> "Total: " + n, ioPool)  // worker của ioPool
        .thenAccept(System.out::println);

Về JDBC: không. CompletableFuture chỉ giúp sắp xếp các task. Nếu lambda gọi JDBC hoặc HTTP client blocking, worker chạy nó vẫn đứng chờ suốt thời gian query:

Tomcat thread              ioPool worker
   | submit --------------->|
   |                        |---- JDBC query ----|   ← worker vẫn bị block
   | (rảnh hoặc join chờ)   |

Async ở đây chủ yếu chuyển việc chờ sang thread khác, không xóa được việc chờ. Muốn non-blocking thật phải có I/O non-blocking (WebClient, R2DBC...).

Đúng ví dụ để thấy rõ khác biệt. Cách 1, blocking: worker gọi ngân hàng rồi ngồi chờ 2s.

ExecutorService workers = Executors.newFixedThreadPool(2);
List<CompletableFuture<String>> flows = IntStream.rangeClosed(1, 10)
        .mapToObj(id -> CompletableFuture.supplyAsync(() -> {
            sleep(2000);                 // giả lập gọi bank blocking
            return "DONE P" + id;
        }, workers))
        .toList();

0s  pool-1-thread-1: P1   pool-1-thread-2: P2   (P3..P10 xếp hàng)
2s  pool-1-thread-1: P3   pool-1-thread-2: P4
...
10s xong hết  → 5 đợt × 2s

Cách 2, mô phỏng non-blocking: worker chỉ gửi yêu cầu rồi trả thread ngay; "ngân hàng" hoàn tất future sau 2s bằng ScheduledExecutorService:

ScheduledExecutorService bank = Executors.newScheduledThreadPool(1);

CompletableFuture<String> sendToBankAsync(int id) {
    CompletableFuture<String> reply = new CompletableFuture<>();
    bank.schedule(() -> reply.complete("OK P" + id), 2, TimeUnit.SECONDS);
    return reply;                        // trả về ngay, không chờ
}

List<CompletableFuture<String>> flows = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
    final int id = i;
    flows.add(CompletableFuture.supplyAsync(() -> "P" + id, workers)
            .thenCompose(x -> sendToBankAsync(id))
            .thenApply(x -> "DONE " + x)
            .exceptionally(ex -> "FAILED P" + id));
}
CompletableFuture.allOf(flows.toArray(new CompletableFuture[0])).join();   // ≈ 2s
workers.shutdown();
bank.shutdown();

10 flow cùng "đang chờ ngân hàng" nhưng không có thread nào đứng chờ, nên tổng chỉ ≈ 2s.

Lưu ý: schedule chỉ mô phỏng async I/O. Thực tế cần HTTP client non-blocking (WebClient), hoặc virtual threads để có hiệu ứng tương tự mà vẫn viết blocking (Bài 3.11).

Đúng rồi, nếu method @Async trả về void thì đó là fire-and-forget:

Tomcat thread (http-nio-8080-exec-1)
   |-- saveOrder()
   |-- emailService.sendEmail()  → đẩy task vào emailExecutor, trả về ngay
   |-- return 200 OK             → Tomcat thread rảnh

emailExecutor-1
   |-- sendEmail() ... xong → hết, không ai được báo

• createOrder() xong luôn, không biết email thành công hay thất bại.
• Exception trong method void không quay về caller; chỉ đi tới AsyncUncaughtExceptionHandler (mặc định là log).
• Email quan trọng thì phải tự lo retry/lưu trạng thái, hoặc đưa vào message queue.

Muốn "biết khi xong" thì trả về CompletableFuture thay vì void.

Đúng một nửa. Nếu request chờ kết quả trước khi trả response thì UX không nhanh hơn:

public Response createOrder() {
    saveOrder();                                                // 1s
    CompletableFuture<Boolean> f = emailService.sendEmail();   // @Async, 5s
    f.join();                                                   // WAIT
    return Response.ok();                                       // user chờ ≈ 6s
}

Và cũng không chắc tốt hơn cho hệ thống: Tomcat thread đứng chờ join(), worker thread đứng chờ SMTP. Một request chiếm 2 thread thay vì 1.

Async chỉ có lợi cho UX khi user không cần kết quả ngay. Bước tiếp theo vẫn làm được, nhưng nối vào future chứ không join:

@Service
class EmailService {
    @Async("emailExecutor")
    public CompletableFuture<String> sendEmail(String to) { ... }
}

public Response createOrder() {
    saveOrder();
    emailService.sendEmail(to)
            .thenAccept(msgId -> historyRepo.save(msgId))      // chạy nền khi email xong
            .exceptionally(ex -> { log.error("email lỗi", ex); return null; });
    return Response.ok();                                      // user chờ ≈ 1s
}

Nhớ thêm: @Async chạy qua Spring proxy, nên gọi this.sendEmail() trong cùng class sẽ không async. Tomcat có pool riêng (200 thread), emailExecutor là pool khác; Tomcat 200 mà app pool 20 thì nút thắt nằm ở app pool.

Q1. CompletableFuture là gì, dùng khi nào?

🇻🇳 Tiếng Việt:

CompletableFuture là Future có thể nối pipeline bất đồng bộ: supplyAsync để chạy, thenApply/thenCompose để biến đổi và nối bước phụ thuộc, thenCombine/allOf để gộp việc độc lập, exceptionally/handle để fallback. Em dùng khi một request gọi nhiều service độc lập, ví dụ checkout gọi user, order, payment mỗi cái 1 giây thì từ 3s xuống khoảng 1s. Với việc IO em luôn truyền executor riêng thay vì commonPool và đặt timeout cả ở future lẫn HTTP client.

🇬🇧 English:

CompletableFuture is a Future that lets us build an asynchronous pipeline: supplyAsync to start work, thenApply and thenCompose to transform and chain dependent steps, thenCombine and allOf to join independent work, and exceptionally or handle for fallbacks. I use it when one request calls several independent services, for example checkout calling user, order and payment services, which cuts latency from about three seconds to one. For I/O work I always pass a dedicated executor instead of the common pool, and set timeouts on both the future and the HTTP client.

Q2. CompletableFuture có biến JDBC thành non-blocking không?

🇻🇳 Tiếng Việt:

Không. Nó chỉ chuyển việc chờ sang worker thread khác; worker vẫn bị block suốt thời gian query. Non-blocking thật cần I/O non-blocking như WebClient, R2DBC. Ví dụ 10 payment với 2 worker blocking mất khoảng 10 giây, còn non-blocking chỉ khoảng 2 giây vì không thread nào đứng chờ.

🇬🇧 English:

No. It only moves the waiting to another worker thread, and that worker is still blocked during the query. Real non-blocking behaviour needs non-blocking I/O such as WebClient or R2DBC. For example, ten payments on two blocking workers take about ten seconds, while a non-blocking version takes about two seconds because no thread sits waiting.

Q3. @Async có làm user nhanh hơn không?

🇻🇳 Tiếng Việt:

Chỉ khi request không chờ kết quả. Nếu controller gọi @Async rồi join() thì user vẫn chờ, còn tốn thêm một thread. Muốn nhanh thì trả response trước và nối bước tiếp bằng thenAccept. @Async là của Spring, chạy qua proxy nên self-invocation không có tác dụng, và nên cấu hình executor riêng.

🇬🇧 English:

Only when the request does not wait for the result. If the controller calls an @Async method and then joins, the user still waits and we use an extra thread. To gain latency we return the response first and chain the next step with thenAccept. @Async is a Spring feature that works through a proxy, so self-invocation does not work, and it should use a dedicated executor.

Cách nhớ ngắn gọn

• Độc lập → allOf/thenCombine; phụ thuộc → thenCompose.
• IO → executor riêng, không commonPool; timeout ở cả future và HTTP client.
• Async ≠ non-blocking: worker vẫn chờ JDBC.
• @Async + join() = user vẫn chờ, tốn 2 thread.