在 Java 中实现异步编程：CompletableFuture 使用指南

在现代软件开发中，异步编程已成为构建高性能、高响应性应用程序的关键技术之一。Java 作为一门广泛使用的编程语言，在其发展过程中不断引入新的特性以支持异步编程。自 Java 8 起，CompletableFuture 的引入为开发者提供了一个强大而灵活的工具，用于处理异步操作和复杂的异步流程编排。本文将深入探讨如何在 Java 中使用 CompletableFuture 实现异步编程，通过详细代码实例，帮助读者全面掌握这一重要技术。

一、CompletableFuture 概述

CompletableFuture 是 Java 8 引入的一个类，位于 java.util.concurrent 包中。它实现了 Future 和 CompletionStage 接口，用于表示一个异步计算的过程，并且提供了丰富的 API 来链式调用和组合多个异步操作。与传统的 Future 相比，CompletableFuture 不仅可以获取异步任务的结果，还能更加灵活地处理任务的完成回调、异常处理以及多个任务之间的依赖关系等。

二、基本使用：创建 CompletableFuture 实例

在使用 CompletableFuture 之前，我们需要了解如何创建它的实例。以下是几种常见的创建方式：

1. 无返回值的异步任务

当我们需要执行一个没有返回值的异步任务时，可以使用 runAsync 方法。该方法接受一个 Runnable 类型的任务，并在异步线程中执行。

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 异步任务代码
    System.out.println("Running async task without return value...");
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("Async task completed.");
});

// 主线程继续执行其他操作
System.out.println("Main thread continues...");
// 等待异步任务完成（可选）
future.join();

在这个示例中，我们创建了一个无返回值的异步任务，它会在控制台打印一条消息，然后休眠两秒，最后打印完成消息。主线程在启动异步任务后，会继续执行自己的操作，最后通过 join 方法等待异步任务完成。

2. 有返回值的异步任务

如果异步任务需要返回一个结果，我们可以使用 supplyAsync 方法。该方法接受一个 Supplier 类型的任务，执行完成后会返回一个结果。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务代码
    System.out.println("Running async task with return value...");
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("Async task completed.");
    return 42; // 返回结果
});

// 主线程继续执行其他操作
System.out.println("Main thread continues...");
// 获取异步任务结果（会阻塞直到结果可用）
Integer result = future.join();
System.out.println("Result: " + result);

这里，异步任务在完成时返回了一个整数值 42。主线程通过 join 方法获取这个结果，并打印出来。

三、链式调用：处理异步任务的结果

CompletableFuture 的一大特点是它支持链式调用，可以在一个异步任务完成后，自动触发另一个任务，从而构建出复杂的异步流程。

1. thenApply：对结果进行转换

thenApply 方法用于对前一个异步任务的结果进行处理，并返回一个新的结果。它接受一个 Function 类型的参数，该函数将前一个任务的结果作为输入，返回一个新的结果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("Step 1: Generate number");
    return 42;
}).thenApply(num -> {
    System.out.println("Step 2: Convert number to string");
    return "Number is: " + num;
});

String result = future.join();
System.out.println(result);

在这个例子中，首先执行一个 supplyAsync 任务生成一个数字，然后通过 thenApply 将数字转换为字符串。最终结果是一个包含数字信息的字符串。

2. thenAccept：消费结果

如果只需要对结果进行消费而不需要返回新结果，可以使用 thenAccept 方法。它接受一个 Consumer 类型的参数，用于处理结果。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Step 1: Generate number");
    return 42;
}).thenAccept(num -> {
    System.out.println("Step 2: Print number");
    System.out.println("Number is: " + num);
}).join();

这里，thenAccept 方法只是简单地打印了生成的数字。

3. thenRun：无参数的后续任务

当不需要前一个任务的结果时，可以使用 thenRun 方法来执行一个无参数的任务。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Step 1: Generate number");
    return 42;
}).thenRun(() -> {
    System.out.println("Step 2: Log completion");
}).join();

在这个示例中，thenRun 方法用于在生成数字后记录任务完成的日志。

四、组合多个 CompletableFuture：并行与依赖关系

在实际应用中，我们经常需要组合多个 CompletableFuture 实例来处理复杂的业务逻辑。这些任务之间可能存在并行关系或依赖关系。

1. 并行执行多个异步任务

当多个异步任务之间没有依赖关系且可以并行执行时，我们可以使用 allOf 或 anyOf 方法来组合它们。

allOf：等待所有任务完成

allOf 方法会创建一个 CompletableFuture，它在所有给定的 CompletableFuture 实例完成后才会完成。

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 2;
});

CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 3;
});

CompletableFuture<Void> allFuture = CompletableFuture.allOf(future1, future2, future3);

allFuture.join();

System.out.println("All tasks completed.");

在这个示例中，三个异步任务分别有不同的执行时间。allOf 方法会等待所有任务完成后才继续执行。

anyOf：等待任意一个任务完成

anyOf 方法会创建一个 CompletableFuture，它在任何一个给定的 CompletableFuture 实例完成后就会立即完成。

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 1;
});

CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return 2;
});

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2);

anyFuture.join();

System.out.println("One of the tasks completed.");

这里，anyOf 方法会在第一个完成的任务完成后立即返回。

2. 处理多个依赖任务

当多个任务之间存在依赖关系时，可以通过链式调用的方式将它们组合起来。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Step 1: Generate number");
    return 42;
}).thenApply(num -> {
    System.out.println("Step 2: Multiply by 2");
    return num * 2;
}).thenApply(num -> {
    System.out.println("Step 3: Add 10");
    return num + 10;
}).thenAccept(result -> {
    System.out.println("Final result: " + result);
}).join();

在这个例子中，每个任务都依赖于前一个任务的结果，通过链式调用实现了任务的顺序执行。

五、异常处理：确保异步任务的健壮性

在异步编程中，异常处理是一个重要的方面。CompletableFuture 提供了多种方式来处理异步任务中可能出现的异常。

1. handle：处理正常结果和异常

handle 方法可以处理正常结果和异常情况。它接受一个 BiFunction 参数，该函数有两个参数：一个是正常结果，另一个是异常。如果任务正常完成，handle 方法会收到结果；如果任务抛出异常，handle 方法会收到异常。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Running async task...");
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    // 故意抛出异常
    throw new RuntimeException("Task failed");
}).handle((result, exception) -> {
    if (exception != null) {
        System.out.println("Exception occurred: " + exception.getMessage());
        return "Default result";
    } else {
        System.out.println("Result: " + result);
        return result;
    }
}).join();

在这个示例中，异步任务故意抛出了一个异常。handle 方法捕获了这个异常，并返回了一个默认结果。

2. whenComplete：处理任务完成（无论是否异常）

whenComplete 方法会在任务完成时执行，无论任务是正常完成还是因异常完成。它接受一个 BiConsumer 参数，该消费者有两个参数：一个是结果，另一个是异常。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Running async task...");
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    throw new RuntimeException("Task failed");
}).whenComplete((result, exception) -> {
    if (exception != null) {
        System.out.println("Exception occurred: " + exception.getMessage());
    } else {
        System.out.println("Result: " + result);
    }
}).join();

这里，whenComplete 方法在任务完成后执行，并打印了异常信息。

3. 异常恢复：exceptionally 和 recover

除了 handle 和 whenComplete，我们还可以使用 exceptionally 和 recover 方法来处理异常并恢复任务执行。

exceptionally：处理异常并返回替代结果

exceptionally 方法用于处理异常情况，并返回一个替代结果。它接受一个 Function 参数，该函数以异常为输入，返回一个替代结果。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Running async task...");
    throw new RuntimeException("Task failed");
}).exceptionally(exception -> {
    System.out.println("Exception occurred: " + exception.getMessage());
    return "Recovered result";
}).join();

在这个例子中，exceptionally 方法捕获了异常，并返回了一个恢复后的结果。

recover：根据异常类型返回替代结果

recover 方法可以根据抛出的异常类型，返回一个替代结果。它接受一个 Class 参数，指定要处理的异常类型，以及一个 Supplier 参数，用于生成替代结果。

CompletableFuture.supplyAsync(() -> {
    System.out.println("Running async task...");
    throw new RuntimeException("Task failed");
}).recover((RuntimeException e) -> {
    System.out.println("Recovered from exception: " + e.getMessage());
    return "Recovered result";
}).join();

这里，recover 方法捕获了 RuntimeException，并返回了一个恢复后的结果。

六、自定义线程池：优化异步任务的执行

默认情况下，CompletableFuture 的异步任务会在 ForkJoinPool.commonPool 中执行。然而，在某些情况下，我们可能需要自定义线程池来更好地控制任务的执行，例如调整线程池大小、设置线程优先级等。

ExecutorService executor = Executors.newFixedThreadPool(5);

CompletableFuture.supplyAsync(() -> {
    System.out.println("Running async task in custom thread pool...");
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Task result";
}, executor).thenAccept(result -> {
    System.out.println("Result: " + result);
}).exceptionally(ex -> {
    System.out.println("Exception: " + ex.getMessage());
    return null;
}).join();

executor.shutdown();

在这个示例中，我们创建了一个自定义的固定大小线程池，并将其作为第二个参数传递给 supplyAsync 方法。这样，异步任务就会在这个自定义线程池中执行。

七、实际应用案例：构建复杂的异步工作流

为了更好地理解 CompletableFuture 的实际应用，我们来看一个构建复杂异步工作流的案例。假设我们需要开发一个电商订单处理系统，其中包含多个异步任务，如验证用户信息、检查库存、计算运费、处理支付等。这些任务之间存在一定的依赖关系，需要合理地组合和处理。

// 验证用户信息
CompletableFuture<User> validateUserAsync(String userId) {
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("Validating user...");
        // 模拟网络延迟
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 返回用户对象
        return new User(userId);
    });
}

// 检查库存
CompletableFuture<Boolean> checkInventoryAsync(Product product, int quantity) {
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("Checking inventory for product: " + product.getId());
        // 模拟网络延迟
        try {
            Thread.sleep(1500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 返回是否有足够库存
        return product.getStock() >= quantity;
    });
}

// 计算运费
CompletableFuture<Double> calculateShippingCostAsync(User user, double totalWeight) {
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("Calculating shipping cost...");
        // 模拟网络延迟
        try {
            Thread.sleep(1200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 根据用户地址和总重量计算运费
        return totalWeight * 0.5;
    });
}

// 处理支付
CompletableFuture<Boolean> processPaymentAsync(User user, double amount) {
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("Processing payment...");
        // 模拟网络延迟
        try {
            Thread.sleep(1800);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 返回支付是否成功
        return true;
    });
}

// 构建订单处理工作流
public void processOrder(String userId, Product product, int quantity) {
    // 验证用户信息
    CompletableFuture<User> userFuture = validateUserAsync(userId);

    // 检查库存
    CompletableFuture<Boolean> inventoryFuture = userFuture
        .thenCompose(user -> checkInventoryAsync(product, quantity));

    // 计算运费
    CompletableFuture<Double> shippingCostFuture = inventoryFuture
        .thenCompose(hasInventory -> {
            if (hasInventory) {
                return calculateShippingCostAsync(userFuture.join(), product.getWeight() * quantity);
            } else {
                return CompletableFuture.completedFuture(0.0);
            }
        });

    // 处理支付
    CompletableFuture<Boolean> paymentFuture = shippingCostFuture
        .thenCompose(shippingCost -> {
            double totalAmount = product.getPrice() * quantity + shippingCost;
            return processPaymentAsync(userFuture.join(), totalAmount);
        });

    // 最终处理
    paymentFuture
        .thenAccept(paymentSuccess -> {
            if (paymentSuccess) {
                System.out.println("Order processed successfully.");
            } else {
                System.out.println("Payment failed, order not processed.");
            }
        })
        .exceptionally(ex -> {
            System.out.println("Error occurred during order processing: " + ex.getMessage());
            return null;
        })
        .join();
}

在这个案例中，我们通过 CompletableFuture 的链式调用和组合，构建了一个复杂的异步订单处理工作流。各个任务之间存在依赖关系，例如只有在用户信息验证成功后才会检查库存，库存检查通过后才会计算运费，最后在运费计算完成后处理支付。通过这种方式，我们能够高效地处理多个异步任务，并且在任务完成或出现异常时进行相应的处理。

八、性能优化与最佳实践

在使用 CompletableFuture 进行异步编程时，为了确保应用程序的性能和稳定性，需要注意一些性能优化技巧和最佳实践。

1. 合理设置线程池大小

根据应用程序的工作负载和服务器资源，合理设置自定义线程池的大小，避免线程过多导致资源耗尽或线程过少导致任务积压。可以通过实验和监控来找到合适的线程池大小。

2. 避免阻塞操作

在异步任务中，应尽量避免长时间的阻塞操作，如同步的 I/O 操作、长时间的计算等。如果无法避免，可以将这些操作放在单独的线程池中执行，或者使用响应式编程模型来处理。

3. 异常处理与监控

完善的异常处理机制能够确保应用程序在遇到异常时不会崩溃，并能够进行相应的恢复操作。同时，对异步任务的执行情况进行监控，及时发现和处理任务失败、超时等问题。

4. 避免过度使用 CompletableFuture

虽然 CompletableFuture 功能强大，但在某些简单场景下可能过度使用会增加代码的复杂性。在决定是否使用 CompletableFuture 时，应权衡异步编程带来的性能提升和代码维护成本。

九、总结

CompletableFuture 是 Java 中实现异步编程的强大工具，它提供了丰富的 API 来创建、组合和处理异步任务，能够帮助开发者构建高性能、高响应性的应用程序。通过本文的详细介绍和代码示例，读者应该对 CompletableFuture 的基本使用、链式调用、组合多个任务、异常处理以及实际应用等方面有了深入的了解。在实际开发中，灵活运用 CompletableFuture 的特性，结合性能优化和最佳实践，能够有效提升应用程序的性能和用户体验。随着 Java 技术的不断发展和应用场景的日益复杂，掌握 CompletableFuture 等异步编程技术对于 Java 开发者来说显得尤为重要。