CompletableFuture结合线程池初步使用-Toy模板网

这篇具有很好参考价值的文章主要介绍了CompletableFuture结合线程池初步使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

CompletableFuture详解

CompletableFuture 是 Java 8 引入的一个类，用于支持异步编程和函数式编程。CompletableFuture 的优点包括：

异步编程：CompletableFuture 支持异步编程，可以在异步任务完成之前继续执行其他任务，从而提高程序的效率和吞吐量。
链式调用：CompletableFuture 提供了丰富的方法来支持链式调用，可以使用函数式编程的风格编写代码，使代码更加简洁、易读、易于维护。
组合操作：CompletableFuture 提供了多个方法来支持组合操作，例如 thenCompose()、thenCombine()、allOf()、anyOf() 等，可以让开发者更加灵活地组合多个异步任务的执行结果。
异常处理：CompletableFuture 提供了多个方法来支持异常处理，例如 exceptionally()、handle() 等，可以方便地捕获和处理异步任务中可能出现的异常。
线程池控制：CompletableFuture 可以通过指定 Executor 来控制异步任务的执行线程池，可以灵活地调整线程池大小和线程数，从而更好地控制程序的性能和资源消耗。

总之，CompletableFuture 是一种强大而灵活的异步编程工具，可以极大地提高程序的效率和可维护性。

常用的方法

thenApply()：当 CompletableFuture 完成时，执行指定的函数并返回一个新的 CompletableFuture，该函数接收前一个 CompletableFuture 的结果，并返回新的结果。
thenAccept()：当 CompletableFuture 完成时，执行指定的操作但不返回任何结果，该操作接收前一个 CompletableFuture 的结果。
thenRun()：当 CompletableFuture 完成时，执行指定的操作并返回一个新的 CompletableFuture。
exceptionally()：在 CompletableFuture 异常时，执行指定的函数，该函数将异常转换为正常结果或抛出一个新的异常。
whenComplete()：在 CompletableFuture 完成时执行指定的操作，该操作接收前一个 CompletableFuture 的结果或异常。
handle()：当 CompletableFuture 完成时，执行指定的函数，该函数可以处理正常结果或异常，并返回一个新的结果。
thenCompose()：在 CompletableFuture 完成时，执行指定的函数并返回一个新的 CompletableFuture，该函数接收前一个 CompletableFuture 的结果，并返回一个新的 CompletableFuture。

常用的静态方法

CompletableFuture.runAsync()：在异步任务中执行没有返回值的操作。
CompletableFuture.supplyAsync()：在异步任务中执行有返回值的操作。
CompletableFuture.completedFuture()：创建一个已经完成的 CompletableFuture 实例。
CompletableFuture.anyOf()：返回一个 CompletableFuture，其结果是给定 CompletableFuture 中最快完成的一个。
CompletableFuture.allOf()：返回一个 CompletableFuture，其结果是给定 CompletableFuture 中所有任务完成后的结果。
CompletableFuture.exceptionally()：在 CompletableFuture 异常时，执行指定的函数。
CompletableFuture.thenApply()：在 CompletableFuture 完成后，执行指定的函数并返回一个新的 CompletableFuture。
CompletableFuture.thenAccept()：在 CompletableFuture 完成后，执行指定的操作但不返回任何结果。
CompletableFuture.thenRun()：在 CompletableFuture 完成后，执行指定的操作并返回一个新的 CompletableFuture。
CompletableFuture.thenCompose()：在 CompletableFuture 完成后，执行指定的函数并返回一个新的 CompletableFuture。

链式调用示例

    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 2)
            .thenApplyAsync(x -> x * 3)
            .thenApplyAsync(x -> x + 1)
            .thenApplyAsync(x -> x * 2);
    future.thenAccept(System.out::println);

使用thenApplyAsync方法将一系列任务串联起来，并在最后使用thenAccept方法处理最终的结果

组合线程池调用示例

// 创建一个自定义的线程池
ExecutorService executor = Executors.newFixedThreadPool(10);
// 使用CompletableFuture提交一个异步任务到线程池中
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 异步任务的代码
}, executor);

// 阻塞等待异步任务完成
future.get();

// 关闭线程池
executor.shutdown();

合并多个异步返回结果示例

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> getFirstResult());
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> getSecondResult());

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

allFutures.join();

String combinedResult = future1.join() + future2.join();


CompletableFuture.supplyAsync()方法接收一个Supplier类型的参数，该参数表示异步操作的执行体。在这里，使用lambda表达式将getFirstResult()和getSecondResult()作为Supplier传递给了supplyAsync()方法。

CompletableFuture.allOf()方法接收一个可变参数列表，每个参数都是一个CompletableFuture。在这里，将future1和future2传递给了allOf()方法，表示需要等待这两个CompletableFuture全部完成。

allFutures.join()方法会阻塞当前线程，直到所有的CompletableFuture都完成。

最后，将future1.join()和future2.join()的结果拼接成一个字符串，并将其存储在combinedResult变量中。join()方法会等待CompletableFuture完成并返回其结果。因为在调用allOf().join()方法之后，所有的CompletableFuture都已经完成，所以这里可以直接调用future1.join()和future2.join()方法获取它们的结果。

将CompletableFuture 带入业务实现简单的查询合并成map返回

文章来源地址https://www.toymoban.com/news/detail-484085.html

 //通过传入的不同产品获取不同的商品对应的数量,返回map类型的数据集
public CompletableFuture<Map<String, String>> getAsync(List<String> products, ExecutorService executorService) {
        List<CompletableFuture<Map.Entry<String, String>>> futures = new ArrayList<>();
        for (String product : products) {
            CompletableFuture<Map.Entry<String, String>> future = CompletableFuture.supplyAsync(() -> {
                String num = "";
                if (product.equals("xxx")) {
                    num = String.valueOf(xxxService.list().size());
                }
                if (product.equals("xxx")) {
                    num = String.valueOf(xxxService.list().size());
                }
                if (product.equals("xxx")) {
                    num = String.valueOf(xxxService.list().size());
                }
                if (product.equals("xxx")) {
                    num = String.valueOf(xxxService.list().size()));
                }
                return (Map.Entry<String, String>) new AbstractMap.SimpleEntry<String, String>(product, price);
            }, executorService);
            futures.add(future);
        }
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        return allFutures.thenApplyAsync(v -> {
            Map<String, String> resultMap = new HashMap<>();
            for (CompletableFuture<Map.Entry<String, String>> future : futures) {
                Map.Entry<String, String> entry = future.join();
                resultMap.put(entry.getKey(), entry.getValue());
            }
            return resultMap;
        }, executorService);
    
 // 具体调用如下:
            ExecutorService executorService = Executors.newFixedThreadPool(4);
            List<String> list = Arrays.asList("商品1", "商品2", "商品3", "商品4");
            CompletableFuture<Map<String, String>> future = getAsync(list, executorService);
            Map<String, String> map = future.get();
            map.forEach((key, value) -> System.out.println(key + ":" + value));
            executorService.shutdown();