CompletableFuture: chạy bất đồng bộ và ghép kết quả

1. Tổng quan

Một API thường phải gọi nhiều service: lấy thông tin khách hàng, đơn hàng, phí vận chuyển. Nếu gọi lần lượt, tổng thời gian bằng tổng độ trễ của từng lần gọi. CompletableFuture cho phép chạy các lần gọi độc lập song song, rồi ghép kết quả lại khi tất cả đã xong.

Đọc xong bài này, bạn sẽ biết cách tạo tác vụ bất đồng bộ, nối các bước xử lý (thenApply, thenCompose) và ghép kết quả (thenCombine, allOf, anyOf). Bạn cũng sẽ biết cách xử lý lỗi, đặt timeout và viết test cho code bất đồng bộ.

2. Chuẩn bị ví dụ

Chúng ta dùng ba service giả lập. Mỗi service “ngủ” một lúc để mô phỏng độ trễ mạng:

import java.math.BigDecimal;
import java.util.List;

record Customer(long id, String name, String city) {}
record Order(long id, long customerId, BigDecimal total) {}

final class Delay {
    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }
}

class CustomerService {
    Customer findById(long id) {
        Delay.sleep(300);
        return new Customer(id, "Lan", "Hue");
    }
}

class OrderService {
    List<Order> findByCustomer(long customerId) {
        Delay.sleep(400);
        return List.of(
            new Order(1, customerId, new BigDecimal("150000")),
            new Order(2, customerId, new BigDecimal("90000")));
    }
}

class ShippingService {
    BigDecimal feeFor(String city) {
        Delay.sleep(200);
        return "Hue".equals(city) ? new BigDecimal("30000") : new BigDecimal("45000");
    }
}

Ba service này chỉ dùng JDK. Bạn không cần thêm thư viện nào để chạy chúng.

3. Tạo tác vụ bất đồng bộ

3.1. supplyAsync và runAsync

supplyAsync chạy một Supplier và trả về CompletableFuture<T>. runAsync chạy một Runnable không có kết quả và trả về CompletableFuture<Void>:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CreateDemo {
    public static void main(String[] args) {
        var customerService = new CustomerService();

        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            CompletableFuture<Customer> customerFuture =
                CompletableFuture.supplyAsync(() -> customerService.findById(42L), executor);

            CompletableFuture<Void> auditFuture =
                CompletableFuture.runAsync(() -> System.out.println("Ghi log truy cập"), executor);

            System.out.println("Main thread không bị chặn");
            System.out.println(customerFuture.join());
            auditFuture.join();
        }
    }
}

Main thread không bị chặn
Ghi log truy cập
Customer[id=42, name=Lan, city=Hue]

Thứ tự hai dòng đầu có thể đổi chỗ cho nhau, vì chúng chạy trên hai thread khác nhau. Bạn cần chú ý mấy điểm sau:

• Tham số thứ hai là Executor. Nếu bỏ qua, tác vụ chạy trên ForkJoinPool.commonPool().
• Executors.newVirtualThreadPerTaskExecutor() (Java 21) tạo một virtual thread cho mỗi tác vụ. Lựa chọn này hợp với tác vụ I/O như gọi HTTP hay truy vấn database.
• ExecutorService là AutoCloseable từ Java 19. Khối try-with-resources sẽ chờ các tác vụ xong rồi mới đóng executor.
• join() chặn thread hiện tại tới khi có kết quả. Nếu tác vụ lỗi, join() ném CompletionException (unchecked). Còn get() ném ExecutionException (checked).

3.2. Tự hoàn thành một future

Đôi khi kết quả đến từ callback chứ không từ một lambda. Lúc đó bạn tạo future rỗng và tự hoàn thành nó:

CompletableFuture<String> paymentResult = new CompletableFuture<>();

// Ở đâu đó, ví dụ trong callback của cổng thanh toán:
paymentResult.complete("PAID");

System.out.println(paymentResult.join()); // PAID

Nếu muốn báo lỗi, bạn gọi completeExceptionally(new RuntimeException("Thẻ bị từ chối")). Khi đã có giá trị sẵn, CompletableFuture.completedFuture(value) là cách gọn nhất để tạo một future đã hoàn thành.

4. Nối các bước xử lý

4.1. thenApply và thenAccept

thenApply biến đổi kết quả, tương tự map của Stream. thenAccept tiêu thụ kết quả và không trả về gì:

CompletableFuture<String> greeting = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenApply(Customer::name)
    .thenApply(name -> "Xin chào, " + name);

greeting.thenAccept(System.out::println).join();

Xin chào, Lan

Mỗi lời gọi thenApply trả về một CompletableFuture mới. Future gốc không bị thay đổi.

4.2. thenCompose khi bước sau cũng bất đồng bộ

Để lấy đơn hàng, ta cần customerId từ bước trước. Nếu dùng thenApply với một hàm trả về future, kết quả sẽ là CompletableFuture<CompletableFuture<List<Order>>>. thenCompose “làm phẳng” kiểu lồng nhau này, tương tự flatMap:

CompletableFuture<List<Order>> ordersFuture = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenCompose(customer -> CompletableFuture.supplyAsync(
        () -> orderService.findByCustomer(customer.id()), executor));

System.out.println(ordersFuture.join().size()); // 2

Hai bước này phụ thuộc nhau nên vẫn chạy tuần tự. Tổng thời gian khoảng 700 ms. thenCompose không làm nhanh hơn. Nó giúp bạn nối các bước mà không chặn thread nào.

5. Ghép kết quả

Với các tác vụ độc lập, CompletableFuture mới thật sự phát huy tác dụng.

5.1. thenCombine cho hai future

Giả sử chúng ta đã biết thành phố của khách và muốn lấy đồng thời danh sách đơn hàng và phí ship:

long start = System.currentTimeMillis();

CompletableFuture<List<Order>> orders =
    CompletableFuture.supplyAsync(() -> orderService.findByCustomer(42L), executor);
CompletableFuture<BigDecimal> shippingFee =
    CompletableFuture.supplyAsync(() -> shippingService.feeFor("Hue"), executor);

CompletableFuture<BigDecimal> grandTotal = orders.thenCombine(shippingFee,
    (orderList, fee) -> orderList.stream()
        .map(Order::total)
        .reduce(BigDecimal.ZERO, BigDecimal::add)
        .add(fee));

System.out.println(grandTotal.join());
System.out.println("Mất " + (System.currentTimeMillis() - start) + " ms");

270000
Mất 405 ms

Hai lời gọi chạy song song. Tổng thời gian xấp xỉ lời gọi chậm nhất (400 ms), thay vì 600 ms nếu gọi tuần tự. Hàm ở tham số thứ hai của thenCombine chỉ chạy khi cả hai future đã hoàn thành.

5.2. allOf cho nhiều future

Khi có nhiều hơn hai tác vụ, bạn dùng allOf. Nó trả về CompletableFuture<Void>, nên bạn phải tự lấy kết quả từ từng future con:

List<String> cities = List.of("Hue", "Hanoi", "Danang");

List<CompletableFuture<BigDecimal>> fees = cities.stream()
    .map(city -> CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor))
    .toList();

CompletableFuture<List<BigDecimal>> allFees = CompletableFuture
    .allOf(fees.toArray(CompletableFuture[]::new))
    .thenApply(ignored -> fees.stream()
        .map(CompletableFuture::join)
        .toList());

System.out.println(allFees.join());

[30000, 45000, 45000]

Bên trong thenApply, mọi future con đã hoàn thành. Vì vậy join() trả về ngay mà không chặn. Thứ tự kết quả giữ đúng thứ tự trong cities.

5.3. anyOf lấy kết quả nhanh nhất

anyOf hoàn thành ngay khi có một future con hoàn thành. Cách này hữu ích khi bạn gọi nhiều bản sao (replica) của cùng một service:

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> { Delay.sleep(500); return "replica-A"; }, executor),
    CompletableFuture.supplyAsync(() -> { Delay.sleep(100); return "replica-B"; }, executor));

System.out.println(fastest.join()); // replica-B

Kiểu trả về là CompletableFuture<Object>, nên bạn phải tự ép kiểu. Lưu ý thêm: nếu future nhanh nhất kết thúc bằng lỗi, anyOf cũng kết thúc bằng lỗi đó.

6. Xử lý lỗi và timeout

6.1. exceptionally và handle

Một exception trong bất kỳ bước nào sẽ lan xuống các bước sau. Những bước đó bị bỏ qua cho tới khi gặp một bước xử lý lỗi:

CompletableFuture<BigDecimal> fee = CompletableFuture
    .supplyAsync(() -> {
        throw new IllegalStateException("Shipping service down");
    }, executor)
    .thenApply(value -> (BigDecimal) value)          // bị bỏ qua
    .exceptionally(ex -> {
        System.out.println("Lỗi: " + ex.getMessage());
        return new BigDecimal("50000");              // giá trị dự phòng
    });

System.out.println(fee.join());

Lỗi: java.lang.IllegalStateException: Shipping service down
50000

Tham số ex thường là CompletionException bọc exception gốc. Vì vậy message có tiền tố tên lớp. Bạn dùng ex.getCause() để lấy lỗi thật.

handle nhận cả kết quả lẫn lỗi. Nó luôn được gọi, dù tác vụ thành công hay thất bại:

CompletableFuture<String> status = CompletableFuture
    .supplyAsync(() -> orderService.findByCustomer(42L), executor)
    .handle((orderList, ex) -> ex == null
        ? "Có " + orderList.size() + " đơn hàng"
        : "Không tải được đơn hàng");

Ngoài ra còn có whenComplete. Nó cũng nhận cả kết quả và lỗi, nhưng không thay đổi kết quả. Phương thức này phù hợp để ghi log.

6.2. orTimeout và completeOnTimeout

Bạn đừng bao giờ chờ một service bên ngoài vô thời hạn. Java 9+ có sẵn hai phương thức cho việc này:

import java.util.concurrent.TimeUnit;

CompletableFuture<BigDecimal> withDefault = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .completeOnTimeout(new BigDecimal("45000"), 500, TimeUnit.MILLISECONDS);

System.out.println(withDefault.join()); // 45000 sau ~500 ms

CompletableFuture<BigDecimal> strict = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .orTimeout(500, TimeUnit.MILLISECONDS);
// strict.join() ném CompletionException với cause là TimeoutException

completeOnTimeout trả về giá trị mặc định khi hết giờ. orTimeout kết thúc future bằng TimeoutException.

Lưu ý: Timeout chỉ hoàn thành future. Nó không dừng tác vụ bên dưới. Thread vẫn tiếp tục “ngủ” 2 giây. cancel(true) cũng không interrupt thread đang chạy. Nếu cần hủy thật, bạn phải tự kiểm tra cờ hủy trong tác vụ hoặc dùng timeout của chính HTTP client.

7. Viết test

Nếu dùng Spring Boot, spring-boot-starter-test đã có sẵn JUnit 5 và AssertJ. Với project thuần Java, bạn thêm:

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.assertj</groupId>
    <artifactId>assertj-core</artifactId>
    <version>3.26.3</version>
    <scope>test</scope>
</dependency>

Trước hết, gom logic vào một lớp để dễ test:

class CheckoutService {
    private final OrderService orderService;
    private final ShippingService shippingService;
    private final Executor executor;

    CheckoutService(OrderService o, ShippingService s, Executor executor) {
        this.orderService = o;
        this.shippingService = s;
        this.executor = executor;
    }

    CompletableFuture<BigDecimal> grandTotal(long customerId, String city) {
        var orders = CompletableFuture.supplyAsync(() -> orderService.findByCustomer(customerId), executor);
        var fee = CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor);
        return orders.thenCombine(fee, (list, f) -> list.stream()
            .map(Order::total)
            .reduce(BigDecimal.ZERO, BigDecimal::add)
            .add(f));
    }
}

Sau đó viết test cho cả trường hợp thành công lẫn thất bại:

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import org.junit.jupiter.api.Test;

class CheckoutServiceTest {

    @Test
    void shouldCombineOrdersAndShippingFee() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), new ShippingService(), executor);

            assertThat(service.grandTotal(42L, "Hue"))
                .succeedsWithin(Duration.ofSeconds(2))
                .isEqualTo(new BigDecimal("270000"));
        }
    }

    @Test
    void shouldPropagateShippingFailure() {
        ShippingService broken = new ShippingService() {
            @Override
            BigDecimal feeFor(String city) {
                throw new IllegalStateException("down");
            }
        };

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), broken, executor);

            assertThatThrownBy(() -> service.grandTotal(42L, "Hue").join())
                .isInstanceOf(CompletionException.class)
                .hasCauseInstanceOf(IllegalStateException.class);
        }
    }
}

succeedsWithin chờ tối đa 2 giây rồi kiểm tra giá trị. Nhờ vậy test không treo nếu code có lỗi. Trong test thứ hai, lỗi từ ShippingService lan qua thenCombine và được bọc trong CompletionException.

Trong test, bạn có thể truyền Runnable::run làm Executor để mọi thứ chạy đồng bộ trên thread của test. Cách này giúp test nhanh và kết quả ổn định.

8. Lỗi hay gặp

• Chạy I/O trên commonPool. Pool này có số thread bằng số CPU trừ một, và được dùng chung cho cả ứng dụng (kể cả parallel stream). Vài lời gọi HTTP chậm có thể làm nghẽn toàn bộ. Với I/O, bạn nên truyền executor riêng, tốt nhất là virtual thread.
• Gọi join() quá sớm. Viết supplyAsync(a).join() rồi mới supplyAsync(b).join() thì vẫn là tuần tự. Bạn cần khởi chạy tất cả future trước, rồi mới ghép hoặc chờ.
• Nuốt exception. Nếu không ai gọi join(), get() hay gắn exceptionally/whenComplete, lỗi sẽ biến mất mà không để lại dấu vết.
• Nhầm thenApply với thenApplyAsync. thenApply có thể chạy trên thread vừa hoàn thành bước trước, hoặc trên thread gọi nếu future đã xong. Nếu bước tiếp theo nặng hoặc blocking, bạn dùng biến thể ...Async(fn, executor).
• Không đặt timeout. Một service treo sẽ kéo theo request của bạn treo theo.
• Chương trình thoát sớm. Thread của commonPool là daemon. Nếu main kết thúc trước khi tác vụ xong, kết quả sẽ không bao giờ được in ra.

Lưu ý: Bạn không nên dùng CompletableFuture khi chỉ có một lời gọi duy nhất và bạn chờ ngay kết quả. Từ Java 21, với code chạy trên virtual thread, viết đồng bộ thông thường thường dễ đọc hơn mà không mất hiệu năng. CompletableFuture chỉ đáng dùng khi bạn cần chạy song song nhiều tác vụ độc lập, hoặc cần một pipeline không chặn.

9. Kết luận

CompletableFuture giúp bạn chạy các tác vụ độc lập song song và ghép kết quả mà không phải tự quản lý thread. Tóm tắt nhanh:

• Tạo tác vụ: supplyAsync/runAsync, luôn truyền một executor phù hợp.
• Nối bước: thenApply khi biến đổi giá trị, thenCompose khi bước sau trả về future.
• Ghép: thenCombine cho hai future, allOf cho nhiều future, anyOf để lấy kết quả nhanh nhất.
• Lỗi và timeout: exceptionally, handle, orTimeout, completeOnTimeout.

Trong test, bạn dùng succeedsWithin của AssertJ để kiểm tra kết quả mà không lo test bị treo.