详细分析Java8中的CompletableFuture异步编程类（附Demo）

前言

Java基本知识：

1. java框架 零基础从入门到精通的学习路线 附开源项目面经等（超全）
2. 【Java项目】实战CRUD的功能整理（持续更新）

1. 基本知识

CompletableFuture 是 Java 8 引入的异步编程类，属于 java.util.concurrent 包的一部分

1. 异步计算（async computation）

2. 事件驱动（callback）

3. 任务组合（composition）

4. 异常处理（exception handling）

核心思想是非阻塞地执行任务，并在任务完成后处理结果

核心特性

• 非阻塞：与 Future.get() 阻塞不同，CompletableFuture 可以通过回调非阻塞地处理结果

• 链式调用：可以通过方法链组合多个异步任务

• 异常处理：提供 exceptionally、handle 等方法处理异常

• 线程池支持：可以指定 Executor 来控制异步任务执行线程

常用的方法总结如下：

核心方法：

创建：supplyAsync(), runAsync()
链式处理：thenApply(), thenAccept(), thenRun()
组合：thenCompose(), thenCombine(), allOf(), anyOf()
异常处理：exceptionally(), handle()

2. Demo

示例 1：异步任务执行

publicstaticvoiddemoRunAsync(){CompletableFuture.runAsync(()->{System.out.println("执行任务: "+Thread.currentThread().getName());}).join();}

示例 2：有返回值异步任务

publicstaticvoiddemoSupplyAsync(){intresult=CompletableFuture.supplyAsync(()->{return42;}).thenApply(x->x*2).join();System.out.println("结果: "+result);}

示例 3：组合两个任务

publicstaticvoiddemoThenCombine(){CompletableFuture<Integer>f1=CompletableFuture.supplyAsync(()->10);CompletableFuture<Integer>f2=CompletableFuture.supplyAsync(()->20);intsum=f1.thenCombine(f2,Integer::sum).join();System.out.println("组合结果: "+sum);}

示例 4：异常处理

publicstaticvoiddemoExceptionally(){CompletableFuture<Integer>future=CompletableFuture.supplyAsync(()->{if(true)thrownewRuntimeException("出错了");return100;});intresult=future.exceptionally(ex->{System.out.println("捕获异常: "+ex.getMessage());return-1;}).join();System.out.println("最终结果: "+result);}

示例 5：等待多个任务完成

publicstaticvoiddemoAllOf(){CompletableFuture<Integer>f1=CompletableFuture.supplyAsync(()->10);CompletableFuture<Integer>f2=CompletableFuture.supplyAsync(()->20);CompletableFuture<Integer>f3=CompletableFuture.supplyAsync(()->30);CompletableFuture<Void>all=CompletableFuture.allOf(f1,f2,f3);all.join();// 等待所有任务完成intsum=f1.join()+f2.join()+f3.join();System.out.println("总和: "+sum);}

大的一个Demo：

importjava.util.concurrent.*;publicclassCompletableFutureDemo{publicstaticvoidmain(String[]args)throwsException{ExecutorServiceexecutor=Executors.newFixedThreadPool(3);// 异步任务1CompletableFuture<Integer>future1=CompletableFuture.supplyAsync(()->{System.out.println("任务1执行线程: "+Thread.currentThread().getName());return10;},executor);// 异步任务2CompletableFuture<Integer>future2=CompletableFuture.supplyAsync(()->{System.out.println("任务2执行线程: "+Thread.currentThread().getName());return20;},executor);// 组合任务：任务1和任务2完成后求和CompletableFuture<Integer>resultFuture=future1.thenCombine(future2,(a,b)->a+b);// 处理结果和异常resultFuture.thenAccept(sum->System.out.println("结果: "+sum)).exceptionally(ex->{System.out.println("异常："+ex.getMessage());returnnull;});// 等待所有任务完成resultFuture.join();executor.shutdown();}}

截图如下：

3. 实战

之所以有这个知识点，是因为我原先的异步写错了，导致有这个总结的思路

上述意思如下：

线程A： 算法1算法2算法3

实际应该如下：

线程A：算法1线程B：算法2线程C：算法3

实际如下:

AI 算法A→Future<TaskResult>AI 算法B→Future<TaskResult>AI 算法C→Future<TaskResult>↓ allOf 等待 ↓ 汇总结果给前端

大致的Demo如下:

publicclassAiRecognition{publicstaticStringTOKEN="xxx";// === 线程池（生产环境建议交给 Spring 管理）===privatestaticfinalExecutorServiceEXECUTOR=Executors.newFixedThreadPool(3);/** * 三个 AI 算法并发识别 */publicstaticCompletableFuture<AiCheckResult>aiCheckAll(StringimgUrlF,StringimgUrlL,StringimgUrlI){CompletableFuture<TaskResult>containerFuture=CompletableFuture.supplyAsync(()->safeContainerCheck(imgUrlF),EXECUTOR);CompletableFuture<TaskResult>doorFuture=CompletableFuture.supplyAsync(()->safeDoorCheck(imgUrlL),EXECUTOR);CompletableFuture<TaskResult>innerFuture=CompletableFuture.supplyAsync(()->safeInnerCheck(imgUrlI),EXECUTOR);returnCompletableFuture.allOf(containerFuture,doorFuture,innerFuture).thenApply(v->{List<TaskResult>list=Arrays.asList(containerFuture.join(),doorFuture.join(),innerFuture.join());AiCheckResultresult=newAiCheckResult();result.setDetails(list);result.setSuccess(list.stream().allMatch(TaskResult::isSuccess));returnresult;});}// === 以下是单个算法的安全封装 ===privatestaticTaskResultsafeContainerCheck(StringimgUrl){try{Stringdata=uploadImageAndRecognize(imgUrl,"http://xxxx");returnTaskResult.ok("containerNumber",data);}catch(Exceptione){returnTaskResult.fail("containerNumber",e.getMessage());}}privatestaticTaskResultsafeDoorCheck(StringimgUrl){try{Stringdata=uploadImageAndGetState(imgUrl,"http://xxxx");returnTaskResult.ok("doorState",data);}catch(Exceptione){returnTaskResult.fail("doorState",e.getMessage());}}privatestaticTaskResultsafeInnerCheck(StringimgUrl){try{Stringdata=uploadImageAndGetState(imgUrl,"http://xxxx");returnTaskResult.ok("innerState",data);}catch(Exceptione){returnTaskResult.fail("innerState",e.getMessage());}}}

1️⃣ 定义单个算法的返回结果

publicclassTaskResult{privateStringtaskName;privatebooleansuccess;privateStringdata;privateStringreason;publicstaticTaskResultok(StringtaskName,Stringdata){TaskResultr=newTaskResult();r.taskName=taskName;r.success=true;r.data=data;returnr;}publicstaticTaskResultfail(StringtaskName,Stringreason){TaskResultr=newTaskResult();r.taskName=taskName;r.success=false;r.reason=reason;returnr;}// getter / setter 省略}

2️⃣ 汇总返回给前端的结果

publicclassAiCheckResult{privatebooleansuccess;privateList<TaskResult>details;// getter / setter}