CompletableFuture的thenCombine/thenCompose：实现异步任务的精准编排

CompletableFuture的thenCombine/thenCompose：实现异步任务的精准编排

大家好，今天我们来深入探讨Java并发编程中CompletableFuture的两个重要方法：thenCombine和thenCompose。CompletableFuture是Java 8引入的强大工具，它极大地简化了异步编程，让我们能够以更清晰、更灵活的方式处理并发任务。thenCombine和thenCompose是CompletableFuture提供的两种组合异步任务的关键方法，理解它们对于构建高效、可维护的异步系统至关重要。

1. 异步编程的挑战与CompletableFuture的优势

在传统的同步编程模型中，程序的执行流程是线性的，一个任务必须等待前一个任务完成后才能开始。这种模型在处理耗时操作（例如网络请求、数据库查询）时会造成线程阻塞，导致程序性能下降。

异步编程则允许我们启动一个耗时任务，而无需等待其完成，可以继续执行其他操作。当耗时任务完成时，再通过回调或事件通知的方式处理结果。

CompletableFuture是Java对Future接口的增强，提供了更丰富的异步操作和组合能力，主要优势包括：

• 非阻塞执行： 异步执行任务，避免线程阻塞。
• 链式调用： 可以将多个CompletableFuture串联起来，形成一个复杂的异步流程。
• 异常处理： 提供了完善的异常处理机制，可以捕获和处理异步任务中的异常。
• 组合操作： 允许将多个CompletableFuture组合成一个，例如等待所有任务完成或选择最先完成的任务。

2. thenCombine：并行执行，合并结果

thenCombine方法用于将两个独立的CompletableFuture并行执行，并将它们的结果合并成一个新的CompletableFuture。其基本语法如下：

<U, V> CompletableFuture<V> thenCombine(
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn
)

• other：另一个需要并行执行的CompletableFuture。
• fn：一个BiFunction，接收两个CompletableFuture的结果作为参数，并返回合并后的结果。T是第一个CompletableFuture的结果类型，U是第二个CompletableFuture的结果类型，V是合并后的结果类型。

示例：并行获取用户信息和订单信息

假设我们需要同时获取用户的基本信息和用户的订单信息，这两个操作可以并行执行，然后将结果合并成一个包含用户完整信息的对象。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;

class User {
    String name;
    String address;
    String orderDetails;

    public User(String name, String address, String orderDetails) {
        this.name = name;
        this.address = address;
        this.orderDetails = orderDetails;
    }

    @Override
    public String toString() {
        return "User{" +
                "name='" + name + ''' +
                ", address='" + address + ''' +
                ", orderDetails='" + orderDetails + ''' +
                '}';
    }
}

public class ThenCombineExample {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> userInfoFuture = CompletableFuture.supplyAsync(() -> {
            // 模拟获取用户信息
            try {
                Thread.sleep(100); // 模拟网络延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "张三，北京";
        });

        CompletableFuture<String> orderInfoFuture = CompletableFuture.supplyAsync(() -> {
            // 模拟获取订单信息
            try {
                Thread.sleep(200); // 模拟网络延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "订单号：12345，金额：100元";
        });

        CompletableFuture<User> combinedFuture = userInfoFuture.thenCombine(orderInfoFuture, (userInfo, orderInfo) -> {
            String[] parts = userInfo.split("，");
            String name = parts[0];
            String address = parts[1];
            return new User(name, address, orderInfo);
        });

        User user = combinedFuture.get();
        System.out.println(user); // 输出: User{name='张三', address='北京', orderDetails='订单号：12345，金额：100元'}
    }
}

在这个例子中，userInfoFuture和orderInfoFuture并行执行，thenCombine方法接收这两个CompletableFuture的结果，并通过BiFunction将它们合并成一个User对象。

thenCombineAsync：指定执行线程池

thenCombine还有一个变体thenCombineAsync，它允许我们指定BiFunction在哪个线程池中执行。这对于避免阻塞默认的ForkJoinPool.commonPool()线程池非常有用。

<U, V> CompletableFuture<V> thenCombineAsync(
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn,
    Executor executor
)

• executor：用于执行BiFunction的线程池。

3. thenCompose：串行执行，依赖结果

thenCompose方法用于将两个CompletableFuture串行执行，其中第二个CompletableFuture的创建依赖于第一个CompletableFuture的结果。其基本语法如下：

<U> CompletableFuture<U> thenCompose(
    Function<? super T,? extends CompletionStage<U>> fn
)

• fn：一个Function，接收第一个CompletableFuture的结果作为参数，并返回一个新的CompletableFuture。T是第一个CompletableFuture的结果类型，U是第二个CompletableFuture的结果类型。

示例：根据用户ID获取用户详细信息

假设我们需要先根据用户ID获取用户基本信息，然后再根据用户基本信息中的账户ID获取用户的账户余额。这两个操作需要串行执行，因为获取账户余额需要依赖用户基本信息中的账户ID。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;

class UserInfo {
    int id;
    String accountId;

    public UserInfo(int id, String accountId) {
        this.id = id;
        this.accountId = accountId;
    }
}

class AccountBalance {
    String accountId;
    double balance;

    public AccountBalance(String accountId, double balance) {
        this.accountId = accountId;
        this.balance = balance;
    }

    @Override
    public String toString() {
        return "AccountBalance{" +
                "accountId='" + accountId + ''' +
                ", balance=" + balance +
                '}';
    }
}

public class ThenComposeExample {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<UserInfo> userInfoFuture = CompletableFuture.supplyAsync(() -> {
            // 模拟根据用户ID获取用户基本信息
            try {
                Thread.sleep(100); // 模拟数据库查询延迟
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return new UserInfo(1, "account123");
        });

        CompletableFuture<AccountBalance> accountBalanceFuture = userInfoFuture.thenCompose(userInfo -> {
            // 模拟根据账户ID获取账户余额
            return CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(150); // 模拟数据库查询延迟
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                return new AccountBalance(userInfo.accountId, 1000.0);
            });
        });

        AccountBalance accountBalance = accountBalanceFuture.get();
        System.out.println(accountBalance); // 输出: AccountBalance{accountId='account123', balance=1000.0}
    }
}

在这个例子中，userInfoFuture先执行，thenCompose方法接收userInfoFuture的结果（即UserInfo对象），并使用UserInfo对象中的accountId创建一个新的CompletableFuture，用于获取账户余额。

thenComposeAsync：指定执行线程池

与thenCombine类似，thenCompose也有一个变体thenComposeAsync，允许我们指定Function在哪个线程池中执行。

<U> CompletableFuture<U> thenComposeAsync(
    Function<? super T,? extends CompletionStage<U>> fn,
    Executor executor
)

• executor：用于执行Function的线程池。

4. thenCombine vs. thenCompose：关键区别

特性	thenCombine	thenCompose
执行方式	并行执行两个独立的CompletableFuture	串行执行两个CompletableFuture，第二个依赖第一个结果
参数	BiFunction，接收两个结果并返回合并后的结果	Function，接收第一个结果并返回一个新的CompletableFuture
适用场景	两个任务可以独立执行，需要将结果合并	第二个任务需要依赖第一个任务的结果才能执行

简单来说：

• 如果两个任务可以同时启动，并且你需要合并它们的结果，使用thenCombine。
• 如果第二个任务必须等待第一个任务完成后才能开始，并且第二个任务的启动需要依赖第一个任务的结果，使用thenCompose。

5. 异常处理

在使用thenCombine和thenCompose时，需要注意异常处理。如果任何一个CompletableFuture抛出异常，整个组合的CompletableFuture也会抛出异常。我们可以使用exceptionally或handle方法来处理异常。

示例：使用exceptionally处理异常

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class ExceptionallyExample {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟抛出异常
            throw new RuntimeException("Something went wrong!");
        });

        CompletableFuture<String> recoveredFuture = future.exceptionally(ex -> {
            System.err.println("Exception occurred: " + ex.getMessage());
            return "Default value";
        });

        String result = recoveredFuture.get();
        System.out.println("Result: " + result); // 输出: Result: Default value
    }
}

在这个例子中，如果future抛出异常，exceptionally方法会捕获异常，并返回一个默认值。

示例：使用handle处理异常

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class HandleExample {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟抛出异常
            throw new RuntimeException("Something went wrong!");
        });

        CompletableFuture<String> handledFuture = future.handle((result, ex) -> {
            if (ex != null) {
                System.err.println("Exception occurred: " + ex.getMessage());
                return "Default value";
            } else {
                return result;
            }
        });

        String result = handledFuture.get();
        System.out.println("Result: " + result); // 输出: Result: Default value
    }
}

在这个例子中，handle方法接收结果和异常作为参数。如果发生异常，我们可以处理异常并返回一个默认值；如果没有发生异常，我们可以直接返回结果。

6. 最佳实践

• 避免阻塞： 尽量使用异步操作，避免在CompletableFuture中执行耗时操作。
• 使用线程池： 为CompletableFuture指定合适的线程池，避免阻塞默认的ForkJoinPool.commonPool()线程池。
• 处理异常： 使用exceptionally或handle方法处理异常，保证程序的健壮性。
• 简化流程： 尽可能使用链式调用，简化异步流程的代码。
• 清晰命名： 为CompletableFuture命名，使其含义清晰，易于理解。

7. 总结

thenCombine 和 thenCompose 是 CompletableFuture 中非常强大的工具，帮助我们构建复杂、高效的异步流程。通过理解它们的区别和适用场景，我们可以更好地利用 CompletableFuture 提高程序的并发性能和响应速度。

8. 编排异步任务，提升程序效率

理解并熟练运用 thenCombine 和 thenCompose 方法，能够让我们更加灵活地编排异步任务，最大化地利用系统资源，显著提升程序的执行效率。

9. 结合异常处理，保证程序健壮

在异步任务编排中，合理的异常处理至关重要。通过 exceptionally 和 handle 方法，我们可以优雅地处理异步任务中的异常，保证程序的健壮性和可靠性。