首页 > 其他分享 >深入解析CompletableFuture的功能和用法

深入解析CompletableFuture的功能和用法

时间:2024-01-31 09:47:20浏览次数:32  
标签:supplyAsync 代码 用法 任务 CompletableFuture 线程 解析 方法

https://zhuanlan.zhihu.com/p/650390185?utm_id=0

1. CompletableFuture 简介

1.1 概述

CompletableFuture是 Java 8 中引入的一个类,它实现了CompletionStage接口,提供了一组丰富的方法来处理异步操作和多个任务的结果。它支持链式操作,可以方便地处理任务的依赖关系和结果转换。相比于传统的Future接口,CompletableFuture更加灵活和强大。

1.2 优势与特点

CompletableFuture的使用具有以下优势和特点:

  • 异步执行:CompletableFuture允许任务在后台线程中异步执行,不会阻塞主线程,提高了应用程序的响应性和性能。
  • 链式操作:通过CompletableFuture提供的方法,可以方便地对任务进行链式操作,构建复杂的任务依赖关系,实现高效的任务调度和执行。
  • 异常处理:CompletableFuture提供了丰富的异常处理方法,可以处理任务执行过程中可能发生的异常,并实现灵活的错误处理和回退机制。
  • 多任务组合:CompletableFuture支持多个任务的并发执行和结果组合。可以轻松地实现多任务并发处理的场景,提高应用程序的效率和并发性。

2. CompletableFuture 的基本用法

2.1 创建 CompletableFuture 对象

使用CompletableFuture创建异步任务非常简单。可以使用
CompletableFuture.supplyAsync()
CompletableFuture.runAsync() 方法来创建CompletableFuture对象。

2.1.1 使用CompletableFuture.supplyAsync() 方法

使用
CompletableFuture.supplyAsync() 方法来创建 CompletableFuture 对象的示例。该方法用于执行具有返回值的任务,并在任务完成时返回结果。

arduino复制代码CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 执行具有返回值的任务
    return "任务结果";
});

在上述示例中,我们使用
CompletableFuture.supplyAsync() 方法创建一个具有返回值的 CompletableFuture 对象,任务会在默认的 ForkJoinPool 中异步执行。

2.1.2 使用CompletableFuture.runAsync() 方法

除了
CompletableFuture.supplyAsync() 方法,CompletableFuture 还提供了
CompletableFuture.runAsync() 方法用于执行没有返回值的任务。

java复制代码CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 执行没有返回值的任务
});

在上述示例中,我们使用
CompletableFuture.runAsync() 方法创建一个没有返回值的 CompletableFuture 对象,任务会在默认的 ForkJoinPool 中异步执行。

2.1.3 指定自定义线程池

我们还可以通过指定自定义线程池来创建 CompletableFuture 对象,以满足特定的并发需求。

java复制代码ExecutorService customExecutor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 执行任务的代码
}, customExecutor);

在上述示例中,我们通过
Executors.newFixedThreadPool(10) 创建了一个固定大小为 10 的自定义线程池,并将其传递给
CompletableFuture.supplyAsync() 方法来执行异步任务。

2.2 获取任务结果

获取CompletableFuture任务的结果有多种方式。最常用的方式是使用join() 方法阻塞当前线程,直到任务完成并返回结果。

2.2.1 使用join() 方法

join() 方法是 CompletableFuture 类提供的一种获取任务结果的方式,它会阻塞当前线程,直到任务完成并返回结果。

java复制代码CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 执行任务的代码
    return "任务结果";
});

String result = future.join();

在上述示例中,我们使用join() 方法获取任务的结果,并将结果赋值给result变量。如果任务还未完成,join() 方法会阻塞当前线程,直到任务完成。

join() 方法和get() 方法非常相似,但join() 方法不会抛出InterruptedExceptionExecutionException异常,而是将异常包装在CompletionException中抛出。因此,它更适合在 Lambda 表达式或流式操作中使用。

2.2.2 使用get() 方法

get() 方法也是 CompletableFuture 类提供的一种获取任务结果的方式,它会阻塞当前线程,直到任务完成并返回结果。与join() 方法不同的是,get() 方法会抛出InterruptedExceptionExecutionException异常,需要进行异常处理。

java复制代码CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 执行任务的代码
    return "任务结果";
});

try {
    String result = future.get();
} catch (InterruptedException | ExecutionException e) {
    // 异常处理逻辑
}

在上述示例中,我们使用get() 方法获取任务的结果,并在可能抛出异常的情况下进行异常处理。如果任务还未完成,get() 方法会阻塞当前线程,直到任务完成。

get() 方法的异常处理较为繁琐,需要捕获InterruptedExceptionExecutionException异常,并进行相应的处理。因此,在 Lambda 表达式或流式操作中,推荐使用join() 方法。

2.3 异步回调方法

CompletableFuture 提供了一系列方法来处理任务的完成事件,实现异步回调。我们将逐一介绍这些方法的区别和用法。

thenApply()

方法签名:thenApply(Function<? super T, ? extends U> fn)

  • 输入参数:上一阶段的任务结果类型为 T。
  • 返回值:新阶段的任务结果类型为 U。
  • 功能:对上一阶段的任务结果进行转换操作,并返回一个新的 CompletableFuture 对象。
java复制代码CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 42)
    .thenApply(result -> result * 2)
    .thenApply(result -> result + 1);

在上述示例中,我们使用 thenApply()方法对上一阶段的结果进行转换,将结果乘以 2,并将转换后的结果加 1。每个 thenApply()方法都返回一个新的 CompletableFuture 对象,可以继续链式调用。

thenAccept()

方法签名:thenAccept(Consumer<? super T> action)

  • 输入参数:上一阶段的任务结果类型为 T。
  • 返回值:CompletableFuture,没有返回值。
  • 功能:对上一阶段的任务结果进行消费操作,没有返回值。
java复制代码CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 42)
    .thenAccept(result -> System.out.println("任务结果:" + result));

在上述示例中,我们使用 thenAccept()方法对上一阶段的结果进行消费,将结果打印输出。thenAccept()方法没有返回值,仅用于消费任务结果。

thenRun()

方法签名:thenRun(Runnable action)

  • 输入参数:无。
  • 返回值:CompletableFuture,没有返回值。
  • 功能:在上一阶段任务完成后执行给定的 Runnable 任务,没有输入参数和返回值。
java复制代码CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 42)
    .thenRun(() -> System.out.println("任务执行完毕"));

在上述示例中,我们使用 thenRun()方法在上一阶段任务完成后执行一个 Runnable 任务,输出一条任务执行完毕的消息。

2.4 多任务组合回调

CompletableFuture 还提供了一些方法来组合多个任务的结果,实现更复杂的异步处理逻辑。

thenCombine()

方法签名:thenCombine(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn)

  • 输入参数:另一个 CompletionStage 对象和一个 BiFunction 函数,函数的输入参数分别为上一阶段的任务结果类型 T 和另一个 CompletionStage 对象的任务结果类型 U,函数的返回值类型为 V。
  • 返回值:新阶段的任务结果类型为 V。
  • 功能:当两个 CompletionStage 对象都完成时,将它们的任务结果传递给给定的 BiFunction 函数进行组合处理,并返回一个新的 CompletableFuture 对象。
java复制代码CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 20);

CompletableFuture<Integer> combinedFuture = future1.thenCombine(future2, (result1, result2) -> result1 + result2);

在上述示例中,我们使用 thenCombine()方法将两个任务的结果进行组合,将它们的结果相加并返回新的 CompletableFuture 对象。

thenCompose()

方法签名:thenCompose(Function<? super T, ? extends CompletionStage> fn)

  • 输入参数:一个 Function 函数,函数的输入参数为上一阶段的任务结果类型 T,函数的返回值为另一个 CompletionStage 对象。
  • 返回值:新阶段的任务结果类型为 U。
  • 功能:当上一阶段的任务完成后,将结果传递给给定的 Function 函数,该函数返回一个新的 CompletionStage 对象,新阶段的任务结果类型为 U。
java复制代码CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);

CompletableFuture<Integer> future2 = future1.thenCompose(result -> CompletableFuture.supplyAsync(() -> result * 2));

在上述示例中,我们使用 thenCompose()方法将上一阶段的结果传递给一个 Function 函数,该函数返回一个新的 CompletionStage 对象。新阶段的任务结果为上一阶段结果的两倍。

allOf()

方法签名:allOf(CompletableFuture<?>... cfs)

  • 输入参数:多个 CompletableFuture 对象。
  • 返回值:CompletableFuture,没有返回值。
  • 功能:等待所有给定的 CompletableFuture 对象都完成,返回一个新的 CompletableFuture 对象。
java复制代码CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 20);

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

在上述示例中,我们使用 allOf()方法等待所有的 CompletableFuture 对象都完成,返回一个新的 CompletableFuture 对象。这样我们就可以在该对象上进行进一步的处理,例如获取各个 CompletableFuture 的结果。

3. 异常处理与错误处理

3.1 异常处理方法

CompletableFuture 提供了多种方法来处理异步任务执行中可能发生的异常。常用的方法有:

  • exceptionally(Function<Throwable, ? extends T> fn) :当 CompletableFuture 执行过程中发生异常时,使用指定的函数进行异常处理,并返回一个新的 CompletableFuture 对象,其中包含处理结果或默认值。
java复制代码CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("任务执行异常");
});

CompletableFuture<Integer> handledFuture = future.exceptionally(ex -> {
    System.out.println("异常处理:" + ex.getMessage());
    return 0; // 默认值
});
  • handle(BiFunction<? super T, Throwable, ? extends U> fn) :当 CompletableFuture 执行完成时,使用指定的函数处理结果或异常,并返回一个新的 CompletableFuture 对象。
java复制代码CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 42);

CompletableFuture<String> handledFuture = future.handle((result, ex) -> {
    if (ex != null) {
        System.out.println("异常处理:" + ex.getMessage());
        return "默认值";
    } else {
        return "结果:" + result;
    }
});

3.2 错误处理与异常链

CompletableFuture 还支持异常链,可以将多个 CompletableFuture 的异常连接起来,形成异常链。可以使用exceptionally()handle() 方法来实现异常链的处理。

java复制代码CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("任务执行异常");
});

CompletableFuture<Integer> handledFuture = future.exceptionally(ex -> {
    System.out.println("异常处理:" + ex.getMessage());
    throw new CustomException("自定义异常", ex);
});

在上述示例中,我们通过exceptionally() 方法处理任务的异常,并抛出一个自定义异常,并将原始异常作为异常链的一部分传递下去。

4. 自定义线程池与资源管理

4.1 默认线程池与 ForkJoinPool

CompletableFuture 默认使用 ForkJoinPool 线程池来执行异步任务。ForkJoinPool 是一种基于工作窃取算法的线程池,适用于任务分解和并行计算。

java复制代码CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务的代码
});

在上述代码中,CompletableFuture 会在默认的 ForkJoinPool 中异步执行任务。

4.2 自定义线程池

除了使用默认线程池,我们还可以自定义线程池来满足特定的需求。自定义线程池可以通过Executors类来创建。

java复制代码ExecutorService customExecutor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务的代码
}, customExecutor);

在上述代码中,我们创建了一个固定大小为 10 的自定义线程池,并将其传递给 CompletableFuture 来执行异步任务。

4.3 资源管理与关闭线程池

在使用自定义线程池时,需要注意及时关闭线程池以释放资源。可以使用ExecutorServiceshutdown()shutdownNow() 方法来关闭线程池。

java复制代码ExecutorService customExecutor = Executors.newFixedThreadPool(10);
// 异步任务代码
customExecutor.shutdown();

在上述代码中,我们在任务完成后调用了shutdown() 方法来关闭线程池。

5. 并发任务的调度与控制

5.1 异步任务的并发度控制

CompletableFuture 允许我们控制并发任务的执行数量。可以通过自定义线程池的大小来限制并发度。

java复制代码ExecutorService customExecutor = Executors.newFixedThreadPool(5);
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    // 异步任务1的代码
}, customExecutor);
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    // 异步任务2的代码
}, customExecutor);
// ...

在上述代码中,我们通过自定义线程池的大小为 5 来限制并发任务的数量。

5.2 任务的超时处理

CompletableFuture 还提供了超时处理的功能,可以控制任务的最大执行时间。可以使用completeOnTimeout() 方法来实现超时处理。

java复制代码CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务的代码
}).completeOnTimeout("默认值", 5, TimeUnit.SECONDS);

在上述代码中,我们指定任务的最大执行时间为 5 秒,如果任务在规定时间内没有完成,将返回默认值。

5.3 中断与取消任务

在某些情况下,我们可能需要中断或取消正在执行的任务。CompletableFuture 提供了cancel() 方法来取消任务的执行。

java复制代码CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务的代码
});

boolean canceled = future.cancel(true);

在上述代码中,我们调用cancel() 方法来取消任务的执行,并传递一个布尔值表示是否中断正在执行的任务。

6. CompletableFuture 的进阶应用

6.1 CompletableFuture 与 IO 操作

CompletableFuture 在处理 IO 操作时非常有用。可以将 IO 操作封装为 CompletableFuture 任务,利用 CompletableFuture 的异步特性提高 IO 操作的效率。

java复制代码CompletableFuture<String> readData = CompletableFuture.supplyAsync(() -> {
    // 执行读取数据的IO操作
    return "读取的数据";
});

CompletableFuture<Void> processData = readData.thenAccept(data -> {
    // 处理读取到的数据
    System.out.println("读取到的数据:" + data);
    // 执行处理数据的操作
});

CompletableFuture<Void> writeData = processData.thenRun(() -> {
    // 执行写入数据的IO操作
    System.out.println("数据写入完成");
});

writeData.join();

在上述代码中,我们使用 CompletableFuture 处理了一个包含读取数据、处理数据和写入数据的 IO 操作流程。通过异步执行和链式操作,可以有效地利用 CPU 和 IO 资源,提高程序的响应性和吞吐量。

6.2 CompletableFuture 与网络请求

CompletableFuture 也可以很好地与网络请求结合使用。我们可以使用 CompletableFuture 发起多个网络请求,并在所有请求完成后处理结果。

java复制代码CompletableFuture<String> request1 = CompletableFuture.supplyAsync(() -> {
    // 发起网络请求1
    return "请求1结果";
});

CompletableFuture<String> request2 = CompletableFuture.supplyAsync(() -> {
    // 发起网络请求2
    return "请求2结果";
});

CompletableFuture<String> request3 = CompletableFuture.supplyAsync(() -> {
    // 发起网络请求3
    return "请求3结果";
});

CompletableFuture<Void> allRequests = CompletableFuture.allOf(request1, request2, request3);

allRequests.thenRun(() -> {
    // 所有请求完成后的处理逻辑
    String result1 = request1.join();
    String result2 = request2.join();
    String result3 = request3.join();
    // 对请求结果进行处理
});

在上述代码中,我们使用 CompletableFuture 发起了三个网络请求,并通过allOf() 方法等待所有请求完成。在所有请求完成后,我们可以使用join() 方法获取各个请求的结果,并进行后续处理。

7. 实战案例

业务背景: 在电商项目的售后业务中,当客服接收到用户的售后申请时,需要进行一系列操作,包括查询订单信息、查询 ERP 中的商品信息、查询用户信息,以及创建售后工单。

代码实现:

java复制代码public CompletableFuture<Void> processAfterSalesRequest(String orderId, String customerId) {
    CompletableFuture<Order> orderFuture = CompletableFuture.supplyAsync(() -> getOrderInfo(orderId));
    CompletableFuture<Inventory> inventoryFuture = CompletableFuture.supplyAsync(() -> getInventoryInfo(orderId));
    CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> getUserInfo(customerId));

    return CompletableFuture.allOf(orderFuture, inventoryFuture, userFuture)
        .thenApplyAsync(ignored -> {
            Order order = orderFuture.join();
            Inventory inventory = inventoryFuture.join();
            User user = userFuture.join();

            // 创建售后工单
            createAfterSalesTicket(order, inventory, user);

            return null;
        });
}

private Order getOrderInfo(String orderId) {
    // 查询订单信息的逻辑
    // ...
    return order;
}

private Inventory getInventoryInfo(String orderId) {
    // 查询ERP中商品信息的逻辑
    // ...
    return inventory;
}

private User getUserInfo(String customerId) {
    // 查询用户信息的逻辑
    // ...
    return user;
}

private void createAfterSalesTicket(Order order, Inventory inventory, User user) {
    // 创建售后工单的逻辑
    // ...
}

在上述代码中,我们使用
CompletableFuture.supplyAsync() 方法分别查询订单信息、ERP 中的商品信息和用户信息,然后使用CompletableFuture.allOf() 方法等待所有查询任务完成。完成后,我们可以通过join() 方法获取各个查询任务的结果,并将结果传递给createAfterSalesTicket() 方法来创建售后工单。

8. 总结

CompletableFuture 是提供了丰富的功能和方法。它能简化并发任务处理,提高系统性能和响应性。通过了解其基本用法、进阶应用和最佳实践,我们可以灵活处理异步回调、任务组合、异常处理和资源管理。

编辑于 202

标签:supplyAsync,代码,用法,任务,CompletableFuture,线程,解析,方法
From: https://www.cnblogs.com/leiqun123/p/17998567

相关文章

  • 解决gpt返回json Python没法解析的情况
    importreimportjsondefreplace_newlines(match):#在匹配的字符串中替换\n和\rreturnmatch.group(0).replace('\n','\\n').replace('\r','\\r')defclean_json_str(json_str:str)->str:""&......
  • 如何对混合日志进行自动化解析
    本文分享自华为云社区《【AIOps】日志分析领域难题:如何对混合日志进行自动化解析?》,作者:DevAI。AIOps智能运维,是指将AI应用于运维领域,基于已有的运维数据通过AI的方式来解决传统运维没办法解决的问题;本文聚焦在AIOps领域的日志解析子课题。现有的日志解析器(logparser)在实际......
  • nginx-go-crossplane crossplane golang 版本的nginx 配置解析包
    nginx-go-crossplane属于python版本crossplanenginx配置解析包的golang移植可以实现nginx配置解析转换为json格式的数据,当然也支持将json转换为nginx配置格式说明对于希望基于nginx搞自己的流量统一平台,同时希望基于api管理的,nginx-go-crossplane是一个很不错的选择......
  • MyBatis 源码系列:MyBatis 解析配置文件、二级缓存、SQL
    解析全局配置文件启动流程分析Stringresource="mybatis-config.xml";//将XML配置文件构建为Configuration配置类reader=Resources.getResourceAsReader(resource);//通过加载配置文件流构建一个SqlSessionFactoryDefaultSqlSessionFactorySqlSessionFactorysqlMapp......
  • C#中问号(?)运算符的历代新增用法
    本文是B站up主十月的寒流的相应视频的学习笔记,可以直接访问该视频获取更详细的讲解:C#中问号(?)运算符的历代新增用法C#1.0三目运算符stringres=x>5?"goood":"no"C#2.0可为空的值类型(NullableValueTypes)int?x=null;上述代码本质上是Nullable<int>x=nul......
  • c# ComboBox控件的常用一些属性和用法、事件及数据绑定方法
    一、常用属性和用法1、Text:获取或设置与此控件关联的文本。//设置默认值this.comboBox1.Text="请选择内容";//orcomboBox1.Items.Add("请选择内容");comboBox1.SelectedIndex=0;2、SelectedIndex:获取或设置指定当前选定项的索引。(设置新索引会SelectedIndexChanged......
  • Docker Volume 的经常用法区别
    对于使用NFS的Docker数据卷,配置示例应当类似于这样:version:'3'services:my_service:image:your_imagevolumes:#挂载命名的数据卷到容器内的路径(详细模式)-type:volumesource:my_nfs_volumetarget:/path/in/containe......
  • jenkins常见用法
    jenkins的一些设置工具设置指定maven配置文件配置jdk配置maven邮件配置系统邮件配置新建maven项目需要安装MavenIntegrationplugin才有这个风格的项目构建触发器git配置构建打包构建脚本jenkins的构建基本大概都是这5部分git地址#代码仓库......
  • vs+qt中使用opengl及关键报错“无法打开包括文件: no such file or directory”与“err
    参考链接https://blog.csdn.net/qq_22533607/article/details/79792083http://t.csdnimg.cn/T8II5http://t.csdnimg.cn/JP8k7基础准备:vs中配置qt插件(略)关键步骤:创建QtWidgetApplication项目将BaseClass修改成QWidget,方框中的内容可以不勾,个人习惯ui文件中添加open......
  • Java开发的SRM供应商、在线询价、招投标采购一体化系统源码功能解析
    前言:随着全球化和信息化的发展,企业采购管理面临越来越多的挑战。传统的采购方式往往涉及到多个繁琐的步骤,包括供应商筛选、询价、招投标等,这些过程不仅耗时,而且容易出错。为了解决这些问题,供应商、询价、招投标一体化系统应运而生。该系统通过集成供应商管理、询价管理、招投标......