99 Ngày Java — Ngày 65: Future và CompletableFuture

Ngày 64 đã gửi task vào ExecutorService. Future cho phép lấy kết quả sau, nhưng get() chặn luồng và khó ghép nhiều bước. CompletableFuture biến kết quả bất đồng bộ thành chuỗi xử lý và kết hợp các nhánh. Ta cần phân biệt ghép tuần tự với ghép song song, đồng thời đưa lỗi và timeout vào thiết kế.

1. Callable và Future

Callable<T> trả kết quả hoặc ném lỗi; submit() trả Future<T>. get() đợi đến khi xong, có thể có timeout. Lỗi từ task được bọc trong ExecutionException; nguyên nhân thật là getCause(). Hủy bằng cancel(true) chỉ yêu cầu interrupt, task phải hợp tác. Không gọi get() trong từng vòng submit nếu mục đích là chạy đồng thời.

ExecutorService pool = Executors.newFixedThreadPool(2);
Future<Integer> answer = pool.submit(() -> 21 * 2);
try { System.out.println(answer.get(2, TimeUnit.SECONDS)); }
finally { pool.shutdown(); }

2. thenApply, thenCompose và thenCombine

thenApply biến giá trị T thành U. Nếu hàm trả một CompletionStage<U>, thenCompose làm phẳng chuỗi phụ thuộc, tránh CompletableFuture<CompletableFuture<U>>. thenCombine đợi hai nhánh độc lập rồi kết hợp kết quả. Tạo cả hai nhánh trước khi combine để chúng có cơ hội chạy đồng thời.

CompletableFuture<User> user = fetchUser(id);
CompletableFuture<Order> order = fetchOrder(id);
CompletableFuture<String> summary = user
    .thenCombine(order, (u, o) -> u.name() + ":" + o.total());
CompletableFuture<Profile> profile = user
    .thenCompose(u -> fetchProfile(u.profileId()));

3. Lỗi, timeout và luồng thực thi

exceptionally trả giá trị dự phòng khi stage trước lỗi; handle nhìn được cả giá trị và lỗi. orTimeout() hoàn tất stage bằng TimeoutException, còn completeOnTimeout() cung cấp fallback. join() ném CompletionException không checked; get() ném checked ExecutionException/InterruptedException. Callback không có hậu tố Async có thể chạy trên luồng hoàn thành stage; nhánh Async không truyền executor thường dùng common pool. Với blocking I/O, truyền executor do bạn quản lý và luôn đóng nó.

CompletableFuture<String> result = CompletableFuture
    .supplyAsync(() -> blockingLookup(id), pool)
    .orTimeout(2, TimeUnit.SECONDS)
    .exceptionally(error -> "fallback");
// Timeout does not guarantee the underlying blocking call stopped.

4. Bẫy khi ghép tác vụ

Đừng chặn bằng join() ngay trong một callback chạy trên pool nhỏ nếu callback đang đợi task cũng cần chính pool đó. Luôn quyết định chỗ nào được fallback, chỗ nào phải báo lỗi. allOf() chỉ báo các nhánh đã xong; muốn giá trị phải lấy từ từng future. Khi cần hủy cả luồng công việc, CompletableFuture không tự lan truyền cancel đến mọi I/O nền.

• Bài tập: ghép hai API độc lập bằng thenCombine rồi đo thời gian so với gọi tuần tự.

• Bài tập: thêm orTimeout và kiểm tra tài nguyên nền có thực sự dừng không.

Future là tay cầm cho một kết quả; CompletableFuture là đồ thị các bước phụ thuộc hoặc song song. Dùng timeout, xử lý lỗi và executor như một phần hợp đồng chứ không thêm sau cùng. Ngày 66 sẽ quay lại khóa và deadlock.

Tài liệu đối chiếu

• https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/Future.html

• https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/CompletableFuture.html