Java 中 CompletableFuture 异步任务执行结果存储

CompletableFuture 基础概述

在Java编程中，CompletableFuture 是Java 8引入的一个强大工具，用于处理异步计算。它实现了 Future 接口和 CompletionStage 接口，提供了一种更为灵活和便捷的方式来管理异步任务及其结果。

Future 接口在Java早期就已存在，它允许我们异步执行任务并获取任务的执行结果。然而，Future 存在一些局限性，比如在获取结果时，如果任务还未完成，调用 get() 方法会阻塞当前线程，而且它缺乏对异步任务完成后的链式操作支持。CompletableFuture 则很好地解决了这些问题。

CompletableFuture 支持异步任务的创建、组合、转换以及结果的获取等操作。它可以通过多种方式创建异步任务，例如 CompletableFuture.supplyAsync() 用于创建有返回值的异步任务，CompletableFuture.runAsync() 用于创建无返回值的异步任务。

下面是一个简单的使用 CompletableFuture.supplyAsync() 创建异步任务的示例：

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

public class CompletableFutureExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Hello, CompletableFuture!";
        });

        String result = future.get();
        System.out.println(result);
    }
}

在上述代码中，CompletableFuture.supplyAsync() 方法接收一个 Supplier 作为参数，在一个新的线程中执行该 Supplier 的 get() 方法，并返回一个 CompletableFuture 对象。通过调用 future.get() 方法来获取异步任务的执行结果，这里会阻塞主线程直到异步任务完成。

CompletableFuture 异步任务执行结果的存储方式

直接获取结果存储

如前面示例中所示，最直接的方式就是通过 get() 方法获取异步任务的执行结果并进行存储。get() 方法有两种形式：

V get()：阻塞当前线程，直到异步任务完成并返回结果。如果任务执行过程中抛出异常，get() 方法会将异常重新抛出，需要在调用处进行捕获处理。
V get(long timeout, TimeUnit unit)：在指定的时间内阻塞当前线程获取结果。如果在规定时间内任务未完成，会抛出 TimeoutException。

以下是使用 get(long timeout, TimeUnit unit) 的示例：

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

public class CompletableFutureTimeoutExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task completed";
        });

        try {
            String result = future.get(2, TimeUnit.SECONDS);
            System.out.println(result);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，我们尝试在2秒内获取异步任务的结果。由于任务实际需要3秒完成，所以会抛出 TimeoutException。

使用 `whenComplete` 回调存储结果

CompletableFuture 提供了 whenComplete 方法，它允许我们在异步任务完成（无论成功还是失败）时执行一个回调函数。whenComplete 方法接收两个参数：任务的结果（如果任务成功完成）和任务抛出的异常（如果任务执行失败）。

示例代码如下：

import java.util.concurrent.CompletableFuture;

public class CompletableFutureWhenCompleteExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Hello from async task";
        });

        future.whenComplete((result, ex) -> {
            if (ex == null) {
                System.out.println("Task completed successfully: " + result);
            } else {
                System.out.println("Task failed: " + ex.getMessage());
            }
        });

        // 主线程不会阻塞，这里可以执行其他操作
        System.out.println("Main thread continues execution");
    }
}

在上述代码中，whenComplete 回调函数在异步任务完成后被执行。如果任务成功，result 会包含任务的返回值；如果任务失败，ex 会包含抛出的异常。这种方式不会阻塞主线程，适合在异步任务完成后进行一些后续处理，同时将结果存储在合适的变量中以备后续使用。

使用 `thenAccept` 处理并存储结果

thenAccept 方法用于在异步任务成功完成时执行一个消费操作。它接收一个 Consumer 作为参数，该 Consumer 会处理异步任务的结果。与 whenComplete 不同的是，thenAccept 只在任务成功完成时被调用。

示例如下：

import java.util.concurrent.CompletableFuture;

public class CompletableFutureThenAcceptExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Data from async task";
        });

        future.thenAccept(result -> {
            System.out.println("Received result: " + result);
            // 这里可以将结果存储到某个变量中
            String storedResult = result;
            System.out.println("Stored result: " + storedResult);
        });

        System.out.println("Main thread continues execution");
    }
}

在这个例子中，当异步任务成功完成后，thenAccept 中的 Consumer 会被调用，处理并可以存储任务的结果。主线程同样不会阻塞。

使用 `handle` 方法处理并存储结果

handle 方法允许我们在异步任务完成（成功或失败）时，对结果进行处理并返回一个新的 CompletableFuture。它接收一个 BiFunction 作为参数，该 BiFunction 会处理任务的结果和可能抛出的异常，并返回一个新的结果。

示例代码如下：

import java.util.concurrent.CompletableFuture;

public class CompletableFutureHandleExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟可能失败的操作
            if (Math.random() < 0.5) {
                throw new RuntimeException("Task failed");
            }
            return "Task completed successfully";
        });

        CompletableFuture<String> newFuture = future.handle((result, ex) -> {
            if (ex == null) {
                return "Processed result: " + result;
            } else {
                return "Error occurred: " + ex.getMessage();
            }
        });

        newFuture.thenAccept(finalResult -> System.out.println(finalResult));
    }
}

在上述代码中，handle 方法根据任务的执行情况返回不同的处理结果，并创建一个新的 CompletableFuture。我们可以进一步对这个新的 CompletableFuture 进行操作，例如通过 thenAccept 处理最终结果并存储。

复杂场景下的结果存储

多个异步任务结果的合并存储

在实际应用中，常常需要同时执行多个异步任务，并将它们的结果合并存储。CompletableFuture 提供了 allOf 和 anyOf 方法来处理这种情况。

allOf 方法接收多个 CompletableFuture 作为参数，返回一个新的 CompletableFuture。这个新的 CompletableFuture 在所有传入的 CompletableFuture 都完成时才会完成。我们可以通过遍历每个 CompletableFuture 的结果来合并存储。

示例代码如下：

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

public class CompletableFutureAllOfExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result from future1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result from future2";
        });

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

        allFutures.join(); // 等待所有任务完成

        List<String> results = new ArrayList<>();
        try {
            results.add(future1.get());
            results.add(future2.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        System.out.println("Combined results: " + results);
    }
}

在这个示例中，我们创建了两个异步任务 future1 和 future2，使用 CompletableFuture.allOf 等待它们都完成，然后分别获取它们的结果并存储到一个 List 中。

anyOf 方法同样接收多个 CompletableFuture 作为参数，返回的新 CompletableFuture 在任何一个传入的 CompletableFuture 完成时就会完成。我们可以获取这个最先完成的任务的结果进行存储。

示例代码如下：

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

public class CompletableFutureAnyOfExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result from future1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result from future2";
        });

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

        String result = (String) anyFuture.get();
        System.out.println("First completed result: " + result);
    }
}

在这个例子中，future2 由于耗时较短会先完成，anyFuture.get() 会获取到 future2 的结果并进行存储和处理。

异步任务结果的链式存储与处理

CompletableFuture 支持链式调用，这使得我们可以对异步任务的结果进行一系列的转换和处理，并在不同阶段存储中间结果。例如，我们可以使用 thenApply 方法对异步任务的结果进行转换，再使用后续的方法进行进一步处理和存储。

示例代码如下：

import java.util.concurrent.CompletableFuture;

public class CompletableFutureChainingExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> "Initial result")
               .thenApply(result -> {
                    String processedResult = result + " - processed";
                    System.out.println("Intermediate processed result: " + processedResult);
                    return processedResult;
                })
               .thenAccept(finalResult -> {
                    System.out.println("Final result: " + finalResult);
                    // 存储最终结果
                    String storedFinalResult = finalResult;
                    System.out.println("Stored final result: " + storedFinalResult);
                });

        System.out.println("Main thread continues execution");
    }
}

在上述代码中，thenApply 方法对初始异步任务的结果进行了处理，生成一个中间处理结果并打印。接着 thenAccept 方法接收这个中间处理结果，进一步处理并存储最终结果。整个过程通过链式调用实现了异步任务结果的逐步处理和存储。

异常处理与结果存储

在异步任务执行过程中，异常处理是非常重要的一部分。CompletableFuture 提供了多种方式来处理异常，并在异常处理过程中合理地存储相关信息。

exceptionally 方法用于在异步任务抛出异常时，返回一个替代结果。它接收一个 Function 作为参数，该 Function 会处理异常并返回一个结果。

示例代码如下：

import java.util.concurrent.CompletableFuture;

public class CompletableFutureExceptionallyExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (Math.random() < 0.5) {
                throw new RuntimeException("Task failed");
            }
            return "Task completed successfully";
        });

        CompletableFuture<String> newFuture = future.exceptionally(ex -> {
            System.out.println("Exception occurred: " + ex.getMessage());
            return "Default result due to exception";
        });

        newFuture.thenAccept(result -> System.out.println("Final result: " + result));
    }
}

在这个例子中，如果异步任务抛出异常，exceptionally 中的 Function 会被调用，返回一个默认结果并进行存储和后续处理。

另外，whenComplete 和 handle 方法也可以用于处理异常并存储相关信息。例如，在 whenComplete 中可以根据异常情况记录错误日志并存储默认结果。

示例代码如下：

import java.util.concurrent.CompletableFuture;
import java.util.logging.Logger;

public class CompletableFutureWhenCompleteExceptionExample {
    private static final Logger LOGGER = Logger.getLogger(CompletableFutureWhenCompleteExceptionExample.class.getName());

    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (Math.random() < 0.5) {
                throw new RuntimeException("Task failed");
            }
            return "Task completed successfully";
        });

        future.whenComplete((result, ex) -> {
            if (ex == null) {
                System.out.println("Task completed successfully: " + result);
            } else {
                LOGGER.severe("Exception occurred: " + ex.getMessage());
                String defaultResult = "Default result due to exception";
                System.out.println("Stored default result: " + defaultResult);
            }
        });

        System.out.println("Main thread continues execution");
    }
}

在这个示例中，whenComplete 回调函数在任务完成时检查是否有异常。如果有异常，记录错误日志并存储一个默认结果。

实际应用场景中的结果存储考量

在 Web 应用中的结果存储

在 Web 应用开发中，经常会遇到需要异步处理请求并返回结果的场景。例如，一个电商应用可能需要异步查询多个库存系统来获取商品库存信息，然后将这些结果合并返回给用户。

假设我们有一个简单的 Spring Boot 应用，使用 CompletableFuture 异步查询多个库存服务：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.HashMap;
import java.util.Map;

@RestController
public class InventoryController {

    @GetMapping("/inventory")
    public Map<String, Integer> getInventory() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> inventory1Future = CompletableFuture.supplyAsync(() -> {
            // 模拟调用库存服务1
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 10;
        });

        CompletableFuture<Integer> inventory2Future = CompletableFuture.supplyAsync(() -> {
            // 模拟调用库存服务2
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 15;
        });

        CompletableFuture.allOf(inventory1Future, inventory2Future).join();

        Map<String, Integer> inventoryMap = new HashMap<>();
        inventoryMap.put("Inventory1", inventory1Future.get());
        inventoryMap.put("Inventory2", inventory2Future.get());

        return inventoryMap;
    }
}

在上述代码中，我们通过 CompletableFuture 异步调用两个库存服务，等待它们都完成后，将结果存储在一个 Map 中并返回给客户端。这样可以提高应用的响应速度，避免阻塞主线程。

在数据处理任务中的结果存储

在大数据处理任务中，可能需要异步处理多个数据块，并将处理结果合并存储。例如，一个日志分析系统可能需要异步分析不同时间段的日志文件，然后将分析结果汇总存储。

假设我们有一个简单的日志分析任务，使用 CompletableFuture 异步处理不同时间段的日志文件：

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

public class LogAnalyzer {
    public static List<String> analyzeLogs() throws ExecutionException, InterruptedException {
        CompletableFuture<String> analysis1Future = CompletableFuture.supplyAsync(() -> {
            // 模拟分析上午的日志
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Analysis result for morning logs";
        });

        CompletableFuture<String> analysis2Future = CompletableFuture.supplyAsync(() -> {
            // 模拟分析下午的日志
            try {
                Thread.sleep(2500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Analysis result for afternoon logs";
        });

        CompletableFuture.allOf(analysis1Future, analysis2Future).join();

        List<String> analysisResults = new ArrayList<>();
        analysisResults.add(analysis1Future.get());
        analysisResults.add(analysis2Future.get());

        return analysisResults;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<String> results = analyzeLogs();
        System.out.println("Combined analysis results: " + results);
    }
}

在这个例子中，我们异步分析上午和下午的日志文件，将分析结果存储在一个 List 中，以便后续进一步处理或存储到数据库等持久化存储中。

性能与资源管理

在使用 CompletableFuture 进行异步任务结果存储时，性能和资源管理是需要考虑的重要因素。过多的异步任务可能导致线程池资源耗尽，影响系统性能。

为了优化性能和资源管理，可以合理配置线程池。例如，使用 ForkJoinPool 作为 CompletableFuture 的默认线程池，通过调整线程池的大小和参数来适应不同的应用场景。

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

public class CompletableFutureThreadPoolExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool forkJoinPool = new ForkJoinPool(4); // 设置线程池大小为4
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 异步任务
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Task result";
        }, forkJoinPool);

        String result = future.get();
        System.out.println(result);
        forkJoinPool.shutdown();
    }
}

在上述代码中，我们创建了一个大小为4的 ForkJoinPool，并将其作为 CompletableFuture.supplyAsync 的线程池参数。这样可以有效地管理异步任务的执行线程，避免资源过度消耗。

另外，合理地使用缓存也可以提高性能。例如，如果异步任务的结果经常被使用，可以将结果缓存起来，下次需要时直接从缓存中获取，而不需要重新执行异步任务。

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

public class CompletableFutureCacheExample {
    private static final ConcurrentHashMap<String, CompletableFuture<String>> cache = new ConcurrentHashMap<>();

    public static CompletableFuture<String> getResult(String key) {
        return cache.computeIfAbsent(key, k -> CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Result for key " + key;
        }));
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> result1 = getResult("key1");
        System.out.println(result1.get());

        CompletableFuture<String> result2 = getResult("key1");
        System.out.println(result2.get()); // 从缓存中获取，不会重新执行异步任务
    }
}

在这个示例中，我们使用 ConcurrentHashMap 作为缓存，computeIfAbsent 方法确保只有当缓存中不存在指定键的结果时，才会执行异步任务并将结果存入缓存。这样可以显著提高性能，减少重复计算。

与其他异步框架的对比及优势

与 Guava 的 ListenableFuture 对比

Guava 的 ListenableFuture 也是一个用于异步计算的框架，它提供了一种异步获取结果并注册回调的方式。然而，与 CompletableFuture 相比，CompletableFuture 具有更丰富的功能和更简洁的语法。

CompletableFuture 支持链式调用，使得异步任务的组合和处理更加直观。例如，在 CompletableFuture 中可以轻松地通过 thenApply、thenAccept 等方法对结果进行转换和处理，而在 ListenableFuture 中实现类似功能相对复杂。

另外，CompletableFuture 内置了对多个异步任务的合并操作，如 allOf 和 anyOf，而 ListenableFuture 没有直接提供这样的功能，需要开发者自己实现一些辅助方法来完成类似任务。

与 RxJava 的对比

RxJava 是一个强大的响应式编程框架，提供了丰富的操作符来处理异步数据流。与 CompletableFuture 相比，RxJava 更侧重于处理连续的异步事件流，适用于复杂的异步场景，如事件驱动的系统。

CompletableFuture 则更简洁直接，适合处理简单的异步任务和结果获取。在需要处理单个异步任务并获取其结果的场景下，CompletableFuture 的代码更易于理解和维护。例如，使用 CompletableFuture.supplyAsync 创建一个异步任务并获取结果，代码非常简洁明了。而在 RxJava 中，需要创建 Observable 等对象，使用更多的操作符来实现相同功能，代码相对复杂。

然而，在处理多个异步任务之间复杂的依赖关系和事件流时，RxJava 的优势就体现出来了。它可以通过操作符轻松地实现任务的合并、转换、过滤等操作。但对于简单的异步任务结果存储，CompletableFuture 是一个更轻量级的选择。

总结 `CompletableFuture` 结果存储的要点

在使用 CompletableFuture 进行异步任务执行结果存储时，我们需要根据具体的应用场景选择合适的方法。直接获取结果存储适用于简单场景，但可能会阻塞线程。使用回调方法如 whenComplete、thenAccept 和 handle 可以在不阻塞主线程的情况下处理和存储结果，并且能够更好地处理异常情况。

在复杂场景中，如多个异步任务结果的合并存储、链式处理以及异常处理，CompletableFuture 提供了丰富的方法来满足需求。同时，要注意性能和资源管理，合理配置线程池和使用缓存，以提高系统的整体性能。

与其他异步框架相比，CompletableFuture 具有简洁易用的特点，在处理简单异步任务结果存储方面具有明显优势。但在面对复杂的异步事件流处理时，可能需要结合其他框架如 RxJava 来满足需求。

通过深入理解和掌握 CompletableFuture 的各种特性和使用方法，我们能够更高效地编写异步代码，提升应用程序的性能和响应能力。

Java 中 CompletableFuture 异步任务执行结果存储

CompletableFuture 基础概述

CompletableFuture 异步任务执行结果的存储方式

直接获取结果存储

使用 whenComplete 回调存储结果

使用 thenAccept 处理并存储结果

使用 handle 方法处理并存储结果

复杂场景下的结果存储

多个异步任务结果的合并存储

异步任务结果的链式存储与处理

异常处理与结果存储

实际应用场景中的结果存储考量

在 Web 应用中的结果存储

在数据处理任务中的结果存储

性能与资源管理

与其他异步框架的对比及优势

与 Guava 的 ListenableFuture 对比

与 RxJava 的对比

总结 CompletableFuture 结果存储的要点

使用 `whenComplete` 回调存储结果

使用 `thenAccept` 处理并存储结果

使用 `handle` 方法处理并存储结果

总结 `CompletableFuture` 结果存储的要点