Future, FutureTask & CompletionService — nhận kết quả từ task async

TL;DR: submit một Callable cho executor, bạn không nhận kết quả ngay — bạn nhận về một Future<V>, tay cầm tới kết quả sẽ có trong tương lai. Future mô hình hóa vòng đời task một chiều qua bốn method: get() chặn tới khi xong, get(timeout, unit) chờ tối đa một khoảng, cancel(...) yêu cầu hủy, isDone()/isCancelled() hỏi trạng thái. get là một biên giới giữa hai thread: lỗi task bị gói trong ExecutionException (nguyên nhân thật ở getCause()), còn InterruptedException nói chính thread đang chờ bị interrupt. FutureTask vừa là Runnable vừa là Future nên ráp thẳng vào executor. CompletionService gom kết quả theo thứ tự hoàn tất thay vì submit. Nhưng Future thuần vấp hai giới hạn — blocking và không compose được — mở đường cho CompletableFuture.

Lấy trang xác nhận đặt vé của TicketFlow: để dựng nó, hệ thống gọi ba service — load hồ sơ user, tính giá cuối cho ghế đã chọn, rồi gửi email xác nhận sau khi hai bước trên xong. Mỗi lời gọi chừng 200ms. Gọi tuần tự là 600ms; gọi khéo — hai bước đầu song song, bước ba nối sau — chỉ còn hơn 400ms, và thread phục vụ request không đứng chờ giây nào.

Bài Executor & thread pool khép lại bằng một câu hỏi treo: submit một Callable xong, ta nhận về một Future — dùng thế nào cho đúng, và nó gánh được bao xa khi các bước bắt đầu phụ thuộc nhau? Bài này mổ trọn họ Future: cách lấy kết quả, cách nó chuyển lỗi qua biên giới thread, và hai giới hạn buộc nó nhường chỗ cho một mô hình mạnh hơn.

1. Callable và Future: tay cầm tới kết quả tương lai

Runnable.run() trả void và không được ném checked exception — hợp task "làm rồi thôi" nhưng vô dụng khi cần một giá trị. Từ Java 5, java.util.concurrent thêm Callable<V> lấp chỗ đó: call() trả về V và được phép ném checked exception.

Callable<Integer> countSeats = () -> {
    Thread.sleep(50);                 // gia lap truy van DB
    return 128;
};

submit một Callable cho executor, ta không nhận kết quả ngay — task còn chưa chắc đã chạy. Thứ nhận lại là một Future<V>: lời hứa rằng kết quả kiểu V sẽ có ở một thời điểm trong tương lai.

try (var pool = Executors.newFixedThreadPool(2)) {
    Future<Integer> seats = pool.submit(countSeats);
    // ... lam viec khac trong luc task chay ...
    int n = seats.get();              // chan o day toi khi task xong
    System.out.println("Còn " + n + " chỗ");
}

Hình dung Future như tấm phiếu gửi đồ ở quầy coat check: đưa áo, bạn nhận ngay một phiếu số — chưa có áo, nhưng có tay cầm để lấy nó về sau.

Future mô hình hóa vòng đời task async qua bốn method: get() lấy kết quả (chặn nếu chưa xong), get(timeout, unit) chờ tối đa một khoảng, cancel(mayInterruptIfRunning) yêu cầu hủy, và isDone()/isCancelled() hỏi trạng thái — một vòng đời chỉ đi một chiều, từ "chưa xong" sang "xong" (thành công, lỗi, hoặc bị hủy) và không bao giờ quay lại.

Không mũi tên nào quay ngược về "chưa xong" — đó là ý nghĩa của "một chiều", và là lý do một Future chỉ trả kết quả đúng một lần.

2. get là một biên giới: nó dồn mọi thứ về thread gọi

Điểm tinh tế nhất của Future là get làm gì với lỗi. Task chạy trên thread khác; nếu call() ném exception, nó không bay ngược về thread gọi theo cách thường được. Executor bắt lấy, gói vào ExecutionException, ném ra đúng lúc bạn gọi get — nguyên nhân thật nằm trong getCause().

Future<Integer> f = pool.submit(() -> { throw new IllegalStateException("kho vé tạm khóa"); });
try {
    f.get();
} catch (ExecutionException e) {
    Throwable cause = e.getCause();   // IllegalStateException goc nam o day
    log.warn("Task hỏng: {}", cause.getMessage());
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();   // khoi phuc co interrupt roi moi xu ly
}

get như cửa hải quan giữa hai thread: mọi thứ task sản sinh bị giữ lại bên kia cho tới khi ai đó qua cửa mang về — lỗi không bị nuốt, nhưng bạn buộc phải đứng chờ.

Hai checked exception get ném ra nói hai chuyện khác hẳn. ExecutionException nghĩa là task chạy và hỏng. InterruptedException nghĩa là chính thread đang chờ bị interrupt — task có thể vẫn chạy bình thường. Cách xử lý đúng cho InterruptedException gần như luôn là khôi phục cờ interrupt bằng Thread.currentThread().interrupt() rồi mới quyết định dừng, đúng interruption policy đã bàn ở bài Executor & thread pool.

3. FutureTask: bản thân Future cũng là một Runnable

Một mảnh ghép giải thích vì sao mô hình này ráp được vào executor: FutureTask. Nó vừa là Runnable (executor chạy được) vừa là Future (bạn lấy kết quả được). Khi bạn submit(callable), executor gói callable vào một FutureTask, chạy nó như Runnable, rồi trả về dưới mặt nạ Future.

FutureTask<Integer> task = new FutureTask<>(countSeats);
new Thread(task).start();             // chay nhu Runnable
int n = task.get();                   // lay ket qua nhu Future

FutureTask cũng dùng được làm "kết quả tính một lần, nhiều thread cùng chờ": gọi get nhiều lần từ nhiều thread đều an toàn, lần đầu chặn tới khi tính xong, các lần sau trả ngay kết quả đã cache.

4. CompletionService: gom kết quả theo thứ tự hoàn tất

Một tình huống thường gặp: bạn submit một loạt task và muốn xử lý kết quả ngay khi từng cái xong, chứ không đợi cả lô. Giữ một List<Future> rồi lặp get theo thứ tự submit sẽ ép bạn chờ theo đúng thứ tự đó: task đầu chậm nhất chặn bạn, không cho chạm tới những task đã xong từ lâu phía sau.

Hãy tưởng tượng bạn gửi quần áo tới năm tiệm giặt. Bạn muốn lấy đồ ngay khi tiệm nào xong, chứ không đứng lì trước tiệm số 1 trong khi đồ ở tiệm số 4 đã giặt xong. CompletionService chính là cái quầy "ai xong trước trả trước" đó.

ExecutorService pool = Executors.newFixedThreadPool(4);
var ecs = new ExecutorCompletionService<Integer>(pool);
for (String region : regions) {
    ecs.submit(() -> querySeatCount(region));   // 5 truy van song song
}
int total = 0;
for (int i = 0; i < regions.size(); i++) {
    Future<Integer> done = ecs.take();          // lay task DA xong som nhat
    total += done.get();                        // get() o day khong con chan lau
}

ExecutorCompletionService bọc quanh executor và đẩy mỗi task vừa hoàn tất vào một hàng đợi nội bộ; take() chặn tới khi có kết quả bất kỳ rồi trả về đúng Future đó. Bạn tiêu thụ kết quả theo thứ tự hoàn tất thay vì submit, nên latency tổng gần với task chậm nhất chứ không phải tổng các lần chờ xếp chồng. Khi cần "lấy cái xong đầu tiên rồi bỏ phần còn lại", poll() không chặn cộng cancel các future còn lại cho bạn hành vi đó.

5. Vì sao Future thuần là chưa đủ?

Future giải quyết gọn bài toán "chạy một task rồi lấy kết quả". Nhưng đặt nó vào một hệ thống thật, giới hạn lộ ra ngay. Đoạn dưới diễn đạt một yêu cầu đời thường: "lấy thông tin user, rồi dựa trên đó gọi service tính giá, rồi dựa trên đó gửi xác nhận."

// Future thuan: moi buoc phu thuoc buoc truoc deu phai get() — chan roi moi di tiep
Future<User>    fu = pool.submit(() -> loadUser(id));
User user        = fu.get();                                 // chan (1)
Future<Price>   fp = pool.submit(() -> priceFor(user));
Price price      = fp.get();                                 // chan (2)
Future<Receipt> fr = pool.submit(() -> sendReceipt(user, price));
Receipt receipt  = fr.get();                                 // chan (3)

Thứ nhất, get là blocking. Mỗi lần gọi, một thread — thường đắt đỏ — bị ghim chỉ để ngồi chờ. Trong service xử lý nghìn request, mỗi request một thread chờ get là đốt đúng thứ tài nguyên mà thread pool sinh ra để tiết kiệm.

Thứ hai, và sâu hơn, Future không compose được. Với Future thuần, bạn phải get kết quả bước một (chặn) mới có dữ liệu submit bước hai, rồi lại get (chặn), rồi mới submit bước ba. Chuỗi phụ thuộc lẽ ra chảy mượt như dây chuyền lại bị cắt vụn thành những lần chặn nối tiếp — thread phục vụ request bị ghim gần như suốt cả ba lần chờ.

Future không có chỗ gắn callback "khi xong thì làm tiếp việc này" — chỉ cho hỏi "xong chưa?" rồi đứng chờ "cho tôi kết quả". Đúng khoảng trống này là lý do CompletableFuture ra đời: bài kế tiếp dựng pipeline async khai báo, lấp cả hai giới hạn trên mà không thread nào phải ngồi chờ ở giữa.

6. 📚 Deep Dive Oracle

Liên hệ các bài khác

• Executor & thread pool — nơi submit/Callable xuất hiện lần đầu và interruption policy cho InterruptedException được đặt ra; bài này nối tiếp câu hỏi treo ở đó.
• Blocking queues & producer–consumer — ExecutorCompletionService bên trong dùng một BlockingQueue để gom task hoàn tất; cùng một cơ chế hàng đợi chặn.
• CompletableFuture — pipeline async — bài kế tiếp, lấp hai giới hạn của Future thuần bằng pipeline compose được.
• Fork/Join: chia để trị song song — một mô hình task khác, nơi task tự chẻ mình thành task con đệ quy thay vì chạy độc lập.

7. Tóm tắt

• Callable<V> trả giá trị và ném được checked exception; submit(callable) → Future<V> — vòng đời một chiều: pending → done (success/fail/cancel).
• get() là biên giới thread: lỗi task bọc trong ExecutionException (getCause() lấy lỗi thật); InterruptedException = thread chờ bị interrupt → Thread.currentThread().interrupt().
• FutureTask: vừa Runnable vừa Future — "tính một lần, nhiều thread cùng chờ": thread sau đến thì get() trả kết quả đã cache, không recompute.
• CompletionService lấy kết quả theo thứ tự hoàn tất; latency tổng ≈ task chậm nhất, không phải tổng dồn. Hai giới hạn Future thuần: get blocking ghim thread + không compose → CompletableFuture.

8. Tự kiểm tra

Bài tiếp theo: CompletableFuture: pipeline async compose được