Chi Tiết và Ví Dụ về API CompletableFuture trong Java

Tổng Quan về Các Đặc Tính Chính

CompletableFuture là một khung lập trình bất đồng bộ được giới thiệu trong Java 8, triển khai các giao diện `Future` và `CompletionStage`. Nó cung cấp các tính năng như xử lý kết quả không chặn, gọi chuỗi, kết hợp tác vụ và xử lý ngoại lệ. Các lợi ích chính bao gồm:

• Xử lý kết quả không chặn: Sử dụng hàm callback để xử lý kết quả, tránh việc chặn luồng.
• Gọi chuỗi: Hỗ trợ các phương thức như `thenApply`, `thenAccept` để nối các tác vụ.
• Kết hợp tác vụ: Cung cấp các phương thức như `thenCombine`, `allOf` để hợp nhất nhiều tác vụ.
• Xử lý ngoại lệ: Hỗ trợ các phương thức như `exceptionally`, `handle` để xử lý ngoại lệ tập trung.

Chi Tiết về Các API Chính

1. Tạo Tác Vụ Bất Đồng Bộ

(1) `supplyAsync` / `runAsync`

// Tạo tác vụ bất đồng bộ có giá trị trả về (sử dụng ForkJoinPool.commonPool() mặc định)
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(1000); } catch (InterruptedException e) {}
    return "Hello";
});

// Tạo tác vụ bất đồng bộ không có giá trị trả về
CompletableFuture<Void> voidFuture = CompletableFuture.runAsync(() -> 
    System.out.println("Đang chạy trong luồng nền")
);

// Chỉ định pool luồng tùy chỉnh
ExecutorService executor = Executors.newFixedThreadPool(5);
CompletableFuture<String> customFuture = CompletableFuture.supplyAsync(
    () -> "Sử dụng pool luồng tùy chỉnh", 
    executor
);

(2) `completedFuture`

// Tạo CompletableFuture đã hoàn thành
CompletableFuture<String> immediateFuture = CompletableFuture.completedFuture("Kết quả ngay lập tức");

2. Xử Lý và Chuyển Đổi Kết Quả

(1) Chuyển đổi đồng bộ

// thenApply: Chuyển đổi kết quả (có giá trị trả về)
CompletableFuture<Integer> lengthFuture = future.thenApply(String::length);

// thenAccept: Xử lý kết quả (không có giá trị trả về)
future.thenAccept(result -> System.out.println("Kết quả: " + result));

// thenRun: Thực hiện tác vụ mới không phụ thuộc vào kết quả trước đó
future.thenRun(() -> System.out.println("Tác vụ hoàn thành"));

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

// thenApplyAsync: Chuyển đổi kết quả bất đồng bộ
CompletableFuture<Integer> asyncLengthFuture = future.thenApplyAsync(String::length);

// Chỉ định pool luồng cho thực thi bất đồng bộ
CompletableFuture<Integer> customAsyncFuture = future.thenApplyAsync(
    String::length, 
    executor
);

3. Kết Hợp Tác Vụ

(1) Kết hợp hai chiều

// thenCombine: Hợp nhất kết quả của hai tác vụ
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 20);
future1.thenCombine(future2, (a, b) -> a + b)
       .thenAccept(System.out::println); // In ra 30

(2) Kết hợp nhiều tác vụ

// allOf: Chờ tất cả tác vụ hoàn thành
CompletableFuture<Void> allFutures = CompletableFuture.allOf(
    CompletableFuture.runAsync(() -> System.out.println("Tác vụ 1")),
    CompletableFuture.runAsync(() -> System.out.println("Tác vụ 2"))
);
allFutures.thenRun(() -> System.out.println("Tất cả tác vụ hoàn thành"));

// anyOf: Kích hoạt khi bất kỳ tác vụ nào hoàn thành
CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> "Kết quả 1"),
    CompletableFuture.supplyAsync(() -> "Kết quả 2")
);
anyFuture.thenAccept(System.out::println); // In ra kết quả của tác vụ đầu tiên hoàn thành

4. Xử Lý Ngoại Lệ

(1) `exceptionally`

CompletableFuture<Integer> failedFuture = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Lỗi");
}).exceptionally(ex -> {
    System.out.println("Ngoại lệ: " + ex.getMessage());
    return 0;
});

(2) `handle`

future.handle((result, ex) -> {
    if (ex != null) {
        System.out.println("Lỗi: " + ex.getMessage());
        return -1;
    }
    return result;
});

(3) `whenComplete`

future.whenComplete((result, ex) -> {
    if (ex != null) System.out.println("Tác vụ thất bại");
});

5. Chuỗi Thực Thi

CompletableFuture.supplyAsync(() -> "Dữ liệu")
    .thenApply(data -> data + " đã xử lý")
    .thenApply(String::toUpperCase)
    .thenAccept(System.out::println); // In ra "DỮ LIỆU ĐÃ XỬ LÝ"

Ví Dụ Thực Tế

Ví dụ 1: Xử Lý Đơn Hàng Thương Mại Điện Tử Đồng Thời

// Mô phỏng xử lý đồng thời nhiều bước của đơn hàng
CompletableFuture<Void> orderProcess = CompletableFuture.runAsync(() -> 
    System.out.println("Xử lý thanh toán")
);

CompletableFuture<Void> inventoryUpdate = CompletableFuture.runAsync(() -> 
    System.out.println("Cập nhật kho hàng")
);

CompletableFuture<Void> notification = CompletableFuture.runAsync(() -> 
    System.out.println("Gửi thông báo")
);

// Chờ tất cả tác vụ hoàn thành
CompletableFuture.allOf(orderProcess, inventoryUpdate, notification)
    .thenRun(() -> System.out.println("Đơn hàng hoàn thành"));

Ví dụ 2: Xử Lý Yêu Cầu Mạng Bất Đồng Bộ

// Mô phỏng lấy dữ liệu và xử lý bất đồng bộ
CompletableFuture<String> fetchData = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(500); } catch (InterruptedException e) {}
    return "Dữ liệu thô";
});

fetchData.thenApply(data -> data + " đã xử lý")
         .thenApply(String::toUpperCase)
         .thenAccept(System.out::println); // In ra "DỮ LIỆU THÔ ĐÃ XỬ LÝ"

Ví dụ 3: Kiểm Soát Thời Gian Hoàn Thành Tác Vụ

// Đặt thời gian hoàn thành tác vụ
CompletableFuture<String> timeoutFuture = new CompletableFuture<>();

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.schedule(() -> {
    if (!timeoutFuture.isDone()) {
        timeoutFuture.completeExceptionally(new TimeoutException("Thao tác quá hạn"));
    }
}, 3, TimeUnit.SECONDS);

// Hoàn thành tác vụ bình thường
new Thread(() -> {
    try { Thread.sleep(5000); } catch (InterruptedException e) {}
    timeoutFuture.complete("Thành công");
}).start();

timeoutFuture.exceptionally(ex -> {
    System.out.println("Lỗi: " + ex.getMessage());
    return "Kết quả dự phòng";
});

Lời Khuyên Tốt Nhất

1. Lựa chọn pool luồng:
   • Sử dụng `ForkJoinPool.commonPool()` mặc định.
   • Trong môi trường sản xuất, nên sử dụng pool luồng tùy chỉnh (như `Executors.newFixedThreadPool()`).
2. Xử lý ngoại lệ:
   • Luôn xử lý bằng `exceptionally` hoặc `handle`.
   • Tránh bỏ qua ngoại lệ dẫn đến tác vụ thất bại im lặng.
3. Phụ thuộc tác vụ:
   • Đảm bảo thứ tự tác vụ đúng, tránh điều kiện cạnh tranh.
   • Sử dụng `thenCompose` để xử lý tác vụ bất đồng bộ lồng nhau.
4. Giải phóng tài nguyên:
   • Tác vụ chạy lâu cần xem xét đóng pool luồng.
   • Sử dụng `try-with-resources` để quản lý tài nguyên.
5. Tối ưu hóa hiệu suất:
   • Đặt kích thước pool luồng hợp lý.
   • Tránh phân chia tác vụ quá nhỏ gây tốn chi phí điều độ.