CompletableFuture — async composition trong Java

public ProfileView getProfile(long userId) {
    User user          = userService.fetchUser(userId);          // 150ms
    List<Order> orders = orderService.fetchOrders(userId);       // 200ms
    List<Item> recs    = recommendationService.recommend(userId); // 180ms
    return new ProfileView(user, orders, recs);
}
// Tong: 150 + 200 + 180 = 530ms

public ProfileView getProfile(long userId) throws Exception {
    Future<User>        userF = exec.submit(() -> userService.fetchUser(userId));
    Future<List<Order>> ordF  = exec.submit(() -> orderService.fetchOrders(userId));
    Future<List<Item>>  recF  = exec.submit(() -> recommendationService.recommend(userId));
    // Tong: max(150, 200, 180) = 200ms
    return new ProfileView(userF.get(), ordF.get(), recF.get());
}

// supplyAsync -- tao CF tu supplier, chay tren pool
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> fetchData());

// supplyAsync voi executor tuong minh (best practice cho I/O)
ExecutorService ioPool = Executors.newFixedThreadPool(20);
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> fetchData(), ioPool);

// runAsync -- void task, khong co return value
CompletableFuture<Void> cf3 = CompletableFuture.runAsync(() -> sendEmail());

// completedFuture -- already done, dung de wrap gia tri sync
CompletableFuture<String> cf4 = CompletableFuture.completedFuture("cached_result");

// failedFuture -- already failed (Java 9+)
CompletableFuture<String> cf5 = CompletableFuture.failedFuture(new RuntimeException("DB down"));

// Kiem tra kich thuoc commonPool
System.out.println(ForkJoinPool.commonPool().getParallelism()); // 7 tren may 8 core

// I/O-bound pool rieng -- khong anh huong commonPool
ExecutorService ioPool = Executors.newFixedThreadPool(50);
// Hoac dung virtual thread (Java 21+) -- khong can size tuning
ExecutorService vtPool = Executors.newVirtualThreadPerTaskExecutor();

CompletableFuture<User> userCF = CompletableFuture.supplyAsync(
    () -> userService.fetchUser(id), ioPool
);

CompletableFuture<String> rawCF = CompletableFuture.supplyAsync(() -> fetchRawJson());
CompletableFuture<User>   userCF = rawCF.thenApply(json -> parseUser(json));

userCF.thenAccept(user -> auditLog.record(user.id(), "profile_viewed"));
// Return: CompletableFuture<Void>

CompletableFuture<Void> done = userCF.thenRun(() -> metrics.increment("profile.loaded"));

// DUNG khi callback tra ve CompletableFuture khac
CompletableFuture<Order> latestOrder = userCF
    .thenCompose(user -> orderService.fetchLatestOrder(user.id()));
// fetchLatestOrder tra ve CF<Order>, khong phai Order

CompletableFuture<User>        userCF   = CompletableFuture.supplyAsync(() -> fetchUser(id));
CompletableFuture<List<Order>> ordersCF = CompletableFuture.supplyAsync(() -> fetchOrders(id));

CompletableFuture<ProfileView> profileCF = userCF.thenCombine(
    ordersCF,
    (user, orders) -> new ProfileView(user, orders)  // BiFunction
);

// thenApply -- chay callback tren thread hoan thanh CF (co the main thread)
cf.thenApply(fn)

// thenApplyAsync -- chay callback tren ForkJoinPool.commonPool()
cf.thenApplyAsync(fn)

// thenApplyAsync voi executor -- chay tren pool chi dinh
cf.thenApplyAsync(fn, customPool)

CompletableFuture<User> userCF = CompletableFuture.supplyAsync(() -> fetchUser(id))
    .exceptionally(ex -> {
        log.warn("fetchUser failed: {}", ex.getMessage());
        return User.anonymous();   // fallback value
    });

CompletableFuture<User> userCF = CompletableFuture.supplyAsync(() -> fetchUser(id))
    .handle((user, ex) -> {
        if (ex != null) return User.anonymous();  // error path
        return user;                               // success path
    });

userCF.whenComplete((user, ex) -> {
    if (ex != null) metrics.increment("fetch.error");
    else metrics.increment("fetch.success");
    // return void -- khong transform result
});

CompletableFuture<User>        userCF   = CompletableFuture.supplyAsync(() -> fetchUser(id), pool);
CompletableFuture<List<Order>> ordersCF = CompletableFuture.supplyAsync(() -> fetchOrders(id), pool);
CompletableFuture<List<Item>>  recCF    = CompletableFuture.supplyAsync(() -> recommend(id), pool);

CompletableFuture<Void> allDone = CompletableFuture.allOf(userCF, ordersCF, recCF);

// Sau allOf.join(), collect ket qua:
allDone.thenRun(() -> {
    User         user   = userCF.join();    // da done, khong block
    List<Order>  orders = ordersCF.join();
    List<Item>   recs   = recCF.join();
    result.set(new ProfileView(user, orders, recs));
});

// Goi 3 service redundant, lay ket qua nhanh nhat
CompletableFuture<Object> firstResult = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> primaryDb.query(id), pool),
    CompletableFuture.supplyAsync(() -> replicaDb.query(id), pool),
    CompletableFuture.supplyAsync(() -> cacheService.get(id), pool)
);

// BAD -- get() block thread hien tai, defeat async
CompletableFuture<User> userCF = CompletableFuture.supplyAsync(() -> fetchUser(id), pool);
User user = userCF.get();   // BLOCK -- thread nay dung o day cho I/O

// Tiep tuc voi ket qua cua user -- sequential, khong async
CompletableFuture<List<Order>> ordersCF = CompletableFuture.supplyAsync(
    () -> fetchOrders(user.id()), pool
);

// DUNG -- chain thay vi block
CompletableFuture<List<Order>> ordersCF = CompletableFuture
    .supplyAsync(() -> fetchUser(id), pool)
    .thenCompose(user -> CompletableFuture.supplyAsync(
        () -> fetchOrders(user.id()), pool
    ));

// BAD -- task A submit task B vao cung commonPool
CompletableFuture.supplyAsync(() -> {
    // Task A: chay tren commonPool
    CompletableFuture<String> inner = CompletableFuture.supplyAsync(() -> {
        // Task B: cung commonPool, nhung commonPool co the da het thread
        return fetchSomething();
    });
    return inner.join();   // Task A block cho Task B -- nhung B khong co thread de chay
    // DEADLOCK CLUSTER: A block, B chua start, khong co thread moi
}, /* no executor -- uses commonPool */);

ExecutorService outerPool = Executors.newFixedThreadPool(10);
ExecutorService innerPool = Executors.newFixedThreadPool(20);

CompletableFuture.supplyAsync(() -> {
    CompletableFuture<String> inner = CompletableFuture.supplyAsync(
        () -> fetchSomething(),
        innerPool    // pool rieng, khong anh huong outerPool
    );
    return inner.join();
}, outerPool);

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    // Chay 30 giay
    return heavyCompute();
}, pool);

cf.cancel(true);   // Mark CF as cancelled
// heavyCompute() van chay den cuoi trong pool thread!

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    for (int i = 0; i < STEPS; i++) {
        if (Thread.currentThread().isInterrupted()) {
            throw new InterruptedException("Cancelled");
        }
        doStep(i);
    }
    return result;
}, pool);