【Java异步编程】CompletableFuture基础(1):创建不同线程的子任务、子任务链式调用与异常处理

文章目录

    • 1. 三种实现接口
    • 2. 链式调用:保证链的顺序性与异步性
    • 3. CompletableFuture创建CompletionStage子任务
    • 4. 处理异常
      • a. 创建回调钩子
      • b. 调用handle()方法统一处理异常和结果
    • 5. 如何选择线程池:不同的业务选择不同的线程池

CompletableFuture是JDK 1.8引入的实现类,该类实现了Future和CompletionStage两个接口。该类的实例作为一个异步任务,可以在自己异步执行完成之后触发一些其他的异步任务,从而达到异步回调的效果。

CompletionStage代表异步计算过程中的某一个阶段,一个阶段完成以后可能会进入另一个阶段。一个阶段可以理解为一个子任务,每一个子任务会包装一个Java函数式接口实例,表示该子任务所要执行的操作。

 

1. 三种实现接口

每个CompletionStage子任务所包装的可以是一个Function、Consumer或者Runnable函数式接口实例。

这三个常用的函数式接口的特点如下:

被包装接口功能描述
FunctionFunction接口的特点是:有输入、有输出。包装了Function实例的CompletionStage子任务需要一个输入参数,并会产生一个输出结果到下一步
RunnableRunnable接口的特点是:无输入、无输出。包装了Runnable实例的CompletionStage子任务既不需要任何输入参数,又不会产生任何输出。
ConsumerConsumer接口的特点是:有输入、无输出。包装了Consumer实例的CompletionStage子任务需要一个输入参数,但不会产生任何输出。

 

2. 链式调用:保证链的顺序性与异步性

多个CompletionStage构成了一条任务流水线,一个环节执行完成了可以将结果移交给下一个环节(子任务)​。多个CompletionStage子任务之间可以使用链式调用。

下面是一个顺序调用的例子:

使用 CompletionStage 及其方法构建了一个异步任务链,thenApply 用于对前一个阶段的结果进行计算并传递结果,thenAccept 用于消费前一个阶段的结果并执行操作,thenRun 用于执行无输入输出的操作。

oneStage//被thenApply包装CompletionStage子任务,由输入输出.thenApply(x -> square(x))  //消耗上游输出,但是没有输出.thenAccept(y -> System.out.println(y)) //不消耗上一个子任务的输出又不产生结果.thenRun(() -> System.out.println()) 

这种链式操作可以方便地将多个异步操作连接起来,同时保证了操作的顺序性和异步性,提高了代码的可维护性和并发性能。

 

接下来是一个异步调用的例子:

在这个例子中,task2 和 task3 都依赖于 task1 完成后执行,但它们可能并行执行,也就是说,task2 和 task3 的执行顺序是不确定的,它们不一定会按照 thenRunAsync 的顺序执行。

CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {System.out.println("Task 1");
});CompletableFuture<Void> task2 = task1.thenRunAsync(() -> {System.out.println("Task 2");
});CompletableFuture<Void> task3 = task1.thenRunAsync(() -> {System.out.println("Task 3");
});

 

3. CompletableFuture创建CompletionStage子任务

CompletableFuture定义了一组方法用于创建CompletionStage子任务(或者阶段性任务)​,基础的方法如下:

//子任务包装一个Runnable实例,并调用ForkJoinPool.commonPool()线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable)//子任务包装一个Runnable实例,并调用指定的executor线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)//子任务包装一个Supplier实例,并调用ForkJoinPool.commonPool()线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)//子任务包装一个Supplier实例,并使用指定的executor线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

其中主要注意的信息是:

  1. Supplier 表示一个无参数但有返回值的函数,Runnable表示无惨无返回值的函数
  2. 在使用CompletableFuture创建CompletionStage子任务时,如果没有指定Executor线程池,在默认情况下CompletionStage会使用公共的ForkJoinPool线程池。
  3. 它们都会交给线程池执行,get方法会堵塞主线程等待执行结果。

给一个例子:

//无返回值异步调用  
@Test  
public void runAsyncDemo() throws Exception {  CompletableFuture<Void> future = CompletableFuture.runAsync(() ->  {  sleepSeconds(1);//模拟执行1秒  Print.tco("run end ...");  });  //等待异步任务执行完成,最多等待2秒  future.get(2, TimeUnit.SECONDS);  
}  //有返回值异步调用  
@Test  
public void supplyAsyncDemo() throws Exception {  CompletableFuture<Long> future = CompletableFuture.supplyAsync(() ->  {  long start = System.currentTimeMillis();  sleepSeconds(1);//模拟执行1秒  Print.tco("run end ...");  return System.currentTimeMillis() - start;  });  //等待异步任务执行完成,现时等待2秒  long time = future.get(2, TimeUnit.SECONDS);  Print.tco("异步执行耗时(秒) = " + time / 1000);  
}

 

4. 处理异常

a. 创建回调钩子

可以为CompletionStage子任务设置特定的回调钩子,当计算结果完成或者抛出异常的时候,执行这些特定的回调钩子。

//设置子任务完成时的回调钩子
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)//设置子任务完成时的回调钩子,可能不在同一线程执行
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)//设置子任务完成时的回调钩子,提交给线程池executor执行
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action,Executor executor)//设置异常处理的回调钩子
public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn)
@Test  
public void whenCompleteDemo() throws Exception {  CompletableFuture<Void> future = CompletableFuture.runAsync(() ->  {  sleepSeconds(1);//模拟执行1秒  Print.tco("抛出异常!");  throw new RuntimeException("发生异常");  //Print.tco("run end ...");  });  //设置执行完成后的回调钩子  future.whenComplete(new BiConsumer<Void, Throwable>() {  @Override  public void accept(Void t, Throwable action) {  Print.tco("执行完成!");  }  });  //设置发生异常后的回调钩子  future.exceptionally(new Function<Throwable, Void>() {  @Override  public Void apply(Throwable t) {  Print.tco("执行失败!" + t.getMessage());  return null;  }  });  future.get();  
}[ForkJoinPool.commonPool-worker-9]:抛出异常!
[main]:执行完成!
[ForkJoinPool.commonPool-worker-9]:执行失败!java.lang.RuntimeException: 发生异常
java.util.concurrent.ExecutionException: java.lang.RuntimeException: 发生异常

有如下几个注意点:

  1. 调用cancel()方法取消CompletableFuture时,任务被视为异常完成,completeExceptionally()方法所设置的异常回调钩子也会被执行。
  2. 如果没有设置异常回调钩子,发生内部异常时会有两种情况发生:
    • 在调用get()时,如果遇到内部异常,get()方法就会抛出ExecutionException(执行异常)​。
    • 在调用join()和getNow(T)启动任务时,如果遇到内部异常,join()和getNow(T)方法就会抛出CompletionException。

 

b. 调用handle()方法统一处理异常和结果

//在执行任务的同一个线程中处理异常和结果
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);//可能不在执行任务的**同一个线程**中处理异常和结果
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);//在指定线程池executor中处理异常和结果
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);
@Test  
public void handleDemo() throws Exception {  CompletableFuture<Void> future = CompletableFuture.runAsync(() ->  {  sleepSeconds(1);//模拟执行1秒  Print.tco("抛出异常!");  throw new RuntimeException("发生异常");  //Print.tco("run end ...");  });  //设置执行完成后的回调钩子  future.handle(new BiFunction<Void, Throwable, Void>() {  @Override  public Void apply(Void input, Throwable throwable) {  if (throwable == null) {  Print.tcfo("没有发生异常!");  } else {  Print.tcfo("sorry,发生了异常!");  }  return null;  }  });  future.get();  
}//
//[ForkJoinPool.commonPool-worker-1]:抛出异常! 
//[ForkJoinPool.commonPool-worker-1|CompletableFutureDemo$3.apply]: sorry,发生了异常!

 

5. 如何选择线程池:不同的业务选择不同的线程池

默认情况下,通过静态方法runAsync()、supplyAsync()创建的CompletableFuture任务会使用公共的ForkJoinPool线程池,默认的线程数是CPU的核数。当然,它的线程数可以通过以下JVM参数设置

     option:-Djava.util.concurrent.ForkJoinPool.common.parallelism

 

如果所有CompletableFuture共享一个线程池,那么一旦有任务执行一些很慢的IO操作,就会导致线程池中的所有线程都阻塞在IO操作上,造成线程饥饿,进而影响整个系统的性能。所以,强烈建议大家根据不同的业务类型创建不同的线程池,以避免互相干扰。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/894269.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

自制虚拟机(C/C++)(一、分析语法和easyx运用,完整虚拟机实现)

网上对虚拟机的解释很多&#xff0c;其实本质就一句话 虚拟机就是机器语言解释器 我们今天要实现汇编语言解释器&#xff0c;下一次再加上ndisasm反汇编器就是真正虚拟机了 注:这里的虚拟机指的是VMware一类的&#xff0c;而不是JVM&#xff0c;python一样的高级语言解释器 …

如何自己设计一个类似 Dubbo 的 RPC 框架?

面试题 如何自己设计一个类似 Dubbo 的 RPC 框架&#xff1f; 面试官心理分析 说实话&#xff0c;就这问题&#xff0c;其实就跟问你如何自己设计一个 MQ 一样的道理&#xff0c;就考两个&#xff1a; 你有没有对某个 rpc 框架原理有非常深入的理解。 你能不能从整体上来思考…

python 使用Whisper模型进行语音翻译

目录 一、Whisper 是什么? 二、Whisper 的基本命令行用法 三、代码实践 四、是否保留Token标记 五、翻译长度问题 六、性能分析 一、Whisper 是什么? Whisper 是由 OpenAI 开源的一个自动语音识别(Automatic Speech Recognition, ASR)系统。它的主要特点是: 多语言…

36. printf

1. printf 格式化函数说的是 printf、 sprintf 和 scanf 这样的函数&#xff0c;分为格式化输入和格式化输出两类函数。学习 C 语言的时候常常通过 printf 函数在屏幕上显示字符串&#xff0c;通过 scanf 函数从键盘获取输入。这样就有了输入和输出了&#xff0c;实现了最基本…

实验八 JSP访问数据库

实验八 JSP访问数据库 目的&#xff1a; 1、熟悉JDBC的数据库访问模式。 2、掌握使用My SQL数据库的使用 实验要求&#xff1a; 1、通过JDBC访问mysql数据&#xff0c;实现增删改查功能的实现 2、要求提交实验报告&#xff0c;将代码和实验结果页面截图放入报告中 实验过程&a…

python学opencv|读取图像(四十六)使用cv2.bitwise_or()函数实现图像按位或运算

【0】基础定义 按位与运算&#xff1a;全1取1&#xff0c;其余取0。按位或运算&#xff1a;全0取0&#xff0c;其余取1。 【1】引言 前序学习进程中&#xff0c;已经对图像按位与计算进行了详细探究&#xff0c;相关文章链接如下&#xff1a; python学opencv|读取图像&…

Flink (十二) :Table API SQL (一) 概览

Apache Flink 有两种关系型 API 来做流批统一处理&#xff1a;Table API 和 SQL。Table API 是用于 Scala 和 Java 语言的查询API&#xff0c;它可以用一种非常直观的方式来组合使用选取、过滤、join 等关系型算子。Flink SQL 是基于 Apache Calcite 来实现的标准 SQL。无论输入…

爬虫基础(六)代理简述

目录 一、什么是代理 二、基本原理 三、代理分类 一、什么是代理 爬虫一般是自动化的&#xff0c;当我们自动运行时 爬虫自动抓取数据&#xff0c;但一会就出现了错误&#xff1a; 如&#xff0c;您的访问频率过高&#xff01; 这是因为网站的反爬措施&#xff0c;如果频…

「 机器人 」利用数据驱动模型替代仿真器:加速策略训练并降低硬件依赖

前言 在强化学习(Reinforcement Learning, RL)中,策略训练需要大量的交互数据(状态、动作、奖励、下一状态),而这些数据通常来自仿真器或真实硬件。传统高保真仿真器虽然能在一定程度上模拟飞行器的动力学,但往往计算量大、开发成本高,且仍可能与真实环境存在差距。为此…

使用vhd虚拟磁盘安装两个win10系统

使用vhd虚拟磁盘安装两个win10系统 前言vhd虚拟磁盘技术简介准备工具开始动手实践1.winX选择磁盘管理2.选择“操作”--“创建VHD”3.自定义一个位置&#xff0c;输入虚拟磁盘大小4.右键初始化磁盘5.选择GPT分区表格式6.右键新建简单卷7.给卷起个名字&#xff0c;用于区分8.打开…

基于云计算、大数据与YOLO设计的火灾/火焰目标检测

摘要&#xff1a;本研究针对火灾早期预警检测需求&#xff0c;采用在Kaggle平台获取数据、采用云计算部署的方式&#xff0c;以YOLOv11构建模型&#xff0c;使用云计算服务器训练模型。经训练&#xff0c;box loss从约3.5降至1.0&#xff0c;cls loss从约4.0降至1.0&#xff0c…

计算机毕业设计Python+CNN卷积神经网络考研院校推荐系统 考研分数线预测 考研推荐系统 考研爬虫 考研大数据 Hadoop 大数据毕设 机器学习

温馨提示&#xff1a;文末有 CSDN 平台官方提供的学长联系方式的名片&#xff01; 温馨提示&#xff1a;文末有 CSDN 平台官方提供的学长联系方式的名片&#xff01; 温馨提示&#xff1a;文末有 CSDN 平台官方提供的学长联系方式的名片&#xff01; 作者简介&#xff1a;Java领…

为什么推荐将静态资源放在CDN上?

1. CDN 是什么&#xff1f; CDN&#xff08;Content Delivery Network&#xff09;是一种分布式网络&#xff0c;由地理上分散的服务器节点组成。其主要功能是将静态资源缓存到各地的边缘服务器上&#xff0c;从而将内容更快地传递给用户。当用户请求资源时&#xff0c;CDN 会…

Web-3.0(Solidity)ERC-20

&#x1f680; 发行自己的加密货币&#xff08;ERC-20 代币&#xff09; 你可以使用 Solidity 编写 ERC-20 智能合约 来发行自己的加密货币&#xff0c;然后部署到 以太坊&#xff08;Ethereum&#xff09; 或 BNB/Polygon 等 EVM 兼容链。 &#x1f4cc; 1. ERC-20 代币是什么…

小程序-基础加强-自定义组件

前言 这次讲自定义组件 1. 准备今天要用到的项目 2. 初步创建并使用自定义组件 这样就成功在home中引入了test组件 在json中引用了这个组件才能用这个组件 现在我们来实现全局引用组件 在app.json这样使用就可以了 3. 自定义组件的样式 发现页面里面的文本和组件里面的文…

AI 的安全性与合规性:实践中的最佳安全策略

随着人工智能&#xff08;AI&#xff09;技术的不断进步&#xff0c;越来越多的企业将其应用于实际业务场景。然而&#xff0c;AI 系统的使用也伴随着安全性和合规性方面的挑战。特别是当 AI 模型处理敏感数据时&#xff0c;如何确保数据的安全、隐私保护、以及防止滥用成为企业…

docker安装emqx

emqx安装 拉取emqx镜像 docker pull emqx/emqx:v4.1.0 运行docker容器 docker run -tid --name emqx -p 1883:1883 -p 8083:8083 -p 8081:8081 -p 8883:8883 -p 8084:8084 -p 18083:18083 emqx/emqx:v4.1.0 放行端口 1、如果要是自己的虚拟机&#xff0c;并且关闭了防火墙&a…

在K8s中部署动态nfs存储provisioner

背景 之前&#xff0c;我已经在一台worker node上安装了local lvm 的provisioner来模拟需要本地高IOPS的数据库等stafeful应用的实现。 为了后续给虚拟机里的K8s集群安装可用的metrics和logs监控系统&#xff08;metrics和logs的时序数据库需要永久存储&#xff09;&#xff0…

【OpenGL】OpenGL游戏案例(二)

文章目录 特殊效果数据结构生成逻辑更新逻辑 文本渲染类结构构造函数加载函数渲染函数 特殊效果 为提高游戏的趣味性&#xff0c;在游戏中提供了六种特殊效果。 数据结构 PowerUp 类只存储存活数据&#xff0c;实际逻辑在游戏代码中通过Type字段来区分执行 class PowerUp …

OSCP:常见文件传输方法

在渗透测试过程中&#xff0c;文件传输是一个关键环节&#xff0c;涉及不同的协议和工具&#xff0c;本文整理了 Linux 和 Windows 系统下常见的文件传输方法&#xff0c;并提供相应的命令示例。 通用文件传输方式 Base64 编码传输 Base64 可用于跨平台传输文件&#xff0c;…