Kết hợp CompletableFuture: thenCombine, allOf, anyOf

1. Kết hợp CompletableFuture

Trong thực tế, hiếm khi mọi thứ đến từ một nguồn duy nhất: có thể tải song song hồ sơ người dùng và các đơn hàng của họ, hợp nhất dữ liệu từ hai microservice, và đôi khi bạn chỉ muốn xử lý phản hồi đến đầu tiên. Cách tiếp cận đồng bộ buộc phải chờ lần lượt; lớp CompletableFuture cho phép khởi chạy tất cả đồng thời và kết hợp kết quả một cách thanh thoát. Nó cung cấp các phương thức chuyên dụng cho việc này: thenCombine, allOf, anyOf. Hãy lần lượt tìm hiểu từng phương thức.

Các tác vụ song song: hai yêu cầu bất đồng bộ

Sẽ trông như thế nào theo cách đồng bộ:

String name = loadUserName();    // lâu
int balance = loadUserBalance(); // lâu
System.out.println("Tên: " + name + ", Số dư: " + balance);

Vấn đề: lời gọi thứ hai chỉ bắt đầu sau khi lời gọi đầu tiên hoàn tất.

Cách bất đồng bộ

Với CompletableFuture, có thể khởi chạy cả hai tác vụ đồng thời:

CompletableFuture<String> nameFuture = CompletableFuture.supplyAsync(() -> loadUserName());
CompletableFuture<Integer> balanceFuture = CompletableFuture.supplyAsync(() -> loadUserBalance());

Nhưng làm thế nào để lấy cả hai kết quả và xử lý chúng cùng nhau? Đó là nhiệm vụ của thenCombine.

2. thenCombine: hợp nhất kết quả của hai tác vụ

Phương thức thenCombine cho phép hợp nhất hai CompletableFuture và thực thi một hành động khi cả hai tác vụ đều hoàn tất. Nó trả về một CompletableFuture mới với kết quả hợp nhất.

Chữ ký (signature):

<A, B, C> CompletableFuture<C> thenCombine(
    CompletionStage<? extends B> other,
    BiFunction<? super A, ? super B, ? extends C> fn
)

• A — kiểu kết quả của future thứ nhất,
• B — kiểu của future thứ hai,
• C — kiểu của kết quả hợp nhất.

Ví dụ

CompletableFuture<String> nameFuture = CompletableFuture.supplyAsync(() -> loadUserName());
CompletableFuture<Integer> balanceFuture = CompletableFuture.supplyAsync(() -> loadUserBalance());

CompletableFuture<String> resultFuture = nameFuture.thenCombine(
    balanceFuture,
    (name, balance) -> "Tên: " + name + ", Số dư: " + balance
);

resultFuture.thenAccept(System.out::println);

Cách hoạt động:

• Cả hai future khởi chạy song song.
• Ngay khi cả hai hoàn tất, hàm (name, balance) -> ... được gọi.
• Future kết quả chứa chuỗi với kết quả.

Ví dụ nhỏ với số

CompletableFuture<Integer> f1 = CompletableFuture.supplyAsync(() -> 2);
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> 3);
CompletableFuture<Integer> sum = f1.thenCombine(f2, Integer::sum);

sum.thenAccept(result -> System.out.println("Tổng: " + result));

Kết quả:

Tổng: 5

Biến thể bất đồng bộ

Nếu việc hợp nhất là một phép tính nặng, hãy dùng thenCombineAsync:

f1.thenCombineAsync(f2, (a, b) -> a * b);

3. allOf: khi có nhiều tác vụ

Nếu không phải hai mà là cả chục tác vụ thì sao? Ví dụ, chúng ta muốn tải song song dữ liệu của mười người dùng cùng lúc. Cho trường hợp này có phương thức CompletableFuture.allOf.

Mô tả

CompletableFuture.allOf(f1, f2, ..., fn) trả về một future mới sẽ hoàn tất khi tất cả các tác vụ được truyền vào hoàn tất. Nhưng có một lưu ý: future này không mang kết quả — kiểu của nó luôn là CompletableFuture<Void>. Để lấy kết quả, cần riêng “lấy” chúng từ các future gốc.

Ví dụ

CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "Thứ nhất");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "Thứ hai");

CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2);

all.thenRun(() -> {
    // Tất cả tác vụ đã hoàn thành!
    String s1 = f1.join(); // join() — giống get(), nhưng ném unchecked exception
    String s2 = f2.join();
    System.out.println(s1 + " & " + s2);
});

Kết quả:

Thứ nhất & Thứ hai

Ví dụ với mảng tác vụ

List<CompletableFuture<String>> futures = new ArrayList<>();
for (int i = 0; i < 5; i++) {
    int id = i;
    futures.add(CompletableFuture.supplyAsync(() -> "Người dùng " + id));
}

CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

all.thenRun(() -> {
    for (CompletableFuture<String> f : futures) {
        System.out.println(f.join());
    }
});

Chuyện gì diễn ra:

• Tất cả tác vụ khởi chạy song song.
• allOf sẽ hoàn tất khi tất cả tác vụ hoàn tất.
• Trong khối thenRun chúng ta có thể lấy kết quả qua join().

Sơ đồ trực quan

[Future1]   \
[Future2] ----> [allOf] ---> thenRun
[Future3]   /

4. anyOf: chờ tác vụ hoàn thành đầu tiên

Đôi khi không cần chờ tất cả, mà chỉ cần kết quả của tác vụ nhanh nhất. Ví dụ, chúng ta yêu cầu dữ liệu từ hai máy chủ — máy chủ nào trả lời trước thì dùng dữ liệu của máy đó. Cho trường hợp này có CompletableFuture.anyOf.

Mô tả

CompletableFuture.anyOf(f1, f2, ..., fn) trả về một future sẽ hoàn tất khi bất kỳ một trong các tác vụ được truyền vào hoàn tất. Kiểu kết quả là CompletableFuture<Object>, vì kiểu của các tác vụ có thể khác nhau.

Ví dụ

CompletableFuture<String> fast = CompletableFuture.supplyAsync(() -> {
    sleep(500);
    return "Máy chủ nhanh";
});
CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
    sleep(2000);
    return "Máy chủ chậm";
});

CompletableFuture<Object> any = CompletableFuture.anyOf(fast, slow);

any.thenAccept(result -> System.out.println("Nhận được: " + result));

Kết quả:

Nhận được: Máy chủ nhanh

Ví dụ với các kiểu dữ liệu khác nhau
Bạn có thể kết hợp các tác vụ có kiểu khác nhau, nhưng khi đó kết quả sẽ có kiểu Object, và cần ép kiểu tường minh.

5. Những điểm hữu ích

Khuyến nghị

• allOf không trả về một mảng kết quả. Cần giữ tham chiếu tới các future gốc để lấy giá trị của chúng qua join() hoặc get().
• anyOf trả về kết quả đầu tiên hoàn thành, nhưng kiểu là Object. Nếu tất cả tác vụ cùng một kiểu, bạn có thể ép kiểu.
• Nếu một trong các tác vụ trong allOf thất bại — future kết quả cũng sẽ thất bại.
• Với thenCombine, cả hai tác vụ phải hoàn tất thành công, nếu không sẽ có ngoại lệ.

Bảng so sánh các phương thức

thenCombine

allOf

Void

anyOf

6. Những lỗi thường gặp khi kết hợp CompletableFuture

Lỗi số 1: Chờ kết quả bằng get()/join() trong luồng chính.
Nếu bạn viết mã bất đồng bộ nhưng cuối cùng vẫn gọi get() hoặc join(), bạn đang chặn luồng và đánh mất mọi lợi ích của tính bất đồng bộ. Tốt hơn hãy dùng thenAccept/thenRun để xử lý kết quả mà không chặn.

Lỗi số 2: Không lưu giữ tham chiếu tới các future gốc khi dùng allOf.
Nếu bạn gọi CompletableFuture.allOf(f1, f2, f3) nhưng không lưu f1, f2, f3 — bạn sẽ không thể lấy được kết quả của chúng. allOf chỉ trả về Void!

Lỗi số 3: Không xử lý lỗi trong chuỗi.
Nếu một trong các tác vụ kết thúc với lỗi, toàn bộ allOf hoặc thenCombine cũng sẽ kết thúc với lỗi. Hãy dùng các phương thức xử lý lỗi (exceptionally, handle, whenComplete) để không bỏ sót ngoại lệ.

Lỗi số 4: Không khớp kiểu với anyOf.
anyOf trả về Object. Nếu các future của bạn trả về những kiểu khác nhau, bạn sẽ phải xác định chính xác thứ gì đến trước. Tốt hơn hãy dùng các tác vụ có cùng kiểu nếu có thể.

Lỗi số 5: Chuỗi quá phức tạp mà không có chú thích.
Khi mã trở nên nhiều, các chuỗi future có thể biến thành “mì spaghetti”. Đừng ngại tách chuỗi thành các biến riêng và chú thích từng bước.

1

Nhiệm vụ

JAVA 25 SELF, mức độ, bài học

Đã khóa

Điều phối giao hàng: Chờ tất cả các lô hàng

Điều phối giao hàng: Chờ tất cả các lô hàng

Mở

1

Nhiệm vụ

JAVA 25 SELF, mức độ, bài học

Đã khóa

Ghép lời chào: Kết hợp các mảnh lời

Ghép lời chào: Kết hợp các mảnh lời

Mở