当前位置：首页 > news >正文

Java 异步计算

news 2025/9/14 7:18:56

CompletableFuture（可完成的Future）

一个可完成的Future，在我们调用他的get方法的时候，他会阻塞等待这个任务完成来获取他的结果。
当然也可以为这个任务注册一些回调，类似于完成时，出现异常时，或者执行超时等额外处理。

使用

CompletableFuture.suppleAsync

异步执行一个任务并返回结果

CompletableFuture.runAsync

异步执行一个任务不返回结果

这两方法都可以为我们快速的创建一个CompletableFuture对象

CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + ":supplyAsync start");return "supplyAsync";
});CompletableFuture<String> completableFuture = CompletableFuture.runAsync(() -> {System.out.println(Thread.currentThread().getName() + ":runAsync start");return "runAsync";
});

这两个方法都是可以有第二个参数的，也就是可以执行线程池，这里默认是**Fork-Join-Pool**(一个可以将一个大任务很自然的分解为多个子任务的线程池)。

回调有哪些

方法	参数	描述
thenApply	T -> U	对结果进行处理，并返回一个新的结果
thenAccpet	T -> void	对结果进行处理，返回结果为Void
thenCompose	T -> CompletableFuture	对结果调用这个函数来进行处理，并返回一个新的结果
handle	(T,Throwable) -> U	与thenApply类似，但是他可以处理异常，根据异常对象是否为null，可以判断是否出现异常。不报错也会执行
whenCompletable	(T,Throwable) -> void	类似handle 但是不返回结果不报错也会执行
exceptionally	Throwable -> U	出现异常时，返回一个结果报错时才会执行。
completableOnTimeout	T, long, TimeUnit	如果超时返回一个指定的结果。超时后的链式操作都不执行
orTimeout	long, TimeUnit	超时返回一个异常 TimeOutException 超时后的链式操作都不执行
thenRun	Runable	执行Runable,返回void，对于不需要任务的返回结果

示例

private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(10, 10, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(50));public static void main(String[] args) {CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + ":supplyAsync");return "supplyAsync";});completableFuture.thenApply((result) -> {System.out.println(Thread.currentThread().getName() + ":thenApply1 ");return result + "\nthenApply1";}).thenAccept((result) -> {System.out.println(Thread.currentThread().getName() + ":thenAccept ");}).thenCompose(result ->CompletableFuture.supplyAsync(() -> {try {System.out.println(Thread.currentThread().getName() + ":thenCompose ");TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {throw new RuntimeException(e);}return result + "\nthenCompose";}))// thenCompose 超时了，orTimeout 之前的的操作都将被忽略 之后的还会执行.thenAccept((result) -> {System.out.println(Thread.currentThread().getName() + ":ignore ");})// .completeOnTimeout("completeOnTimeout", 1, TimeUnit.SECONDS)// 这里会抛出一个TimeOutException.orTimeout(1, TimeUnit.SECONDS)// throwable则可以获取到 TimeOutException .handleAsync((result, throwable) -> {System.out.println(Thread.currentThread().getName() + ":handleAsync  ");if (throwable == null) {return result;}throwable.printStackTrace();return result + "\nhandleAsync";}, THREAD_POOL_EXECUTOR)// 因为TimeOutException 被handleAsync处理了，所以这里也没有异常了throwable为null.whenComplete((result, throwable) -> {System.out.println(Thread.currentThread().getName() + ":whenComplete  ");if (throwable == null) {return;}throwable.printStackTrace();}).thenRun(() -> {System.out.println(Thread.currentThread().getName() + ":thenRun ");});try {Thread.sleep(5000); // 添加短暂的延迟  因为是异步任务这里不等待一下的话main线程就终止了} catch (InterruptedException e) {e.printStackTrace();}THREAD_POOL_EXECUTOR.shutdownNow();
}#### 执行结果
ForkJoinPool.commonPool-worker-3:supplyAsync
ForkJoinPool.commonPool-worker-3:thenApply1 
ForkJoinPool.commonPool-worker-3:thenAccept 
ForkJoinPool.commonPool-worker-3:thenCompose 
pool-1-thread-1:handleAsync  
pool-1-thread-1:whenComplete  
pool-1-thread-1:thenRun 
java.util.concurrent.TimeoutException

注意回调时机即可。

Async回调

我们这里使用一个叫handleAsync的方法与普通的handle相比，他是执行的线程发送了变化。
使用Async在大多数情况下都会是在一个新的线程下去帮我们执行这个回调，而普通的则是在原有由原有执行任务的线程去执行这个回调。
这里的大多数情况是指我们在使用自定义线程池的时候。而我们的Fork-Join-Pool可能会为一些短暂的任务重用相同的线程，以减少线程的创建和销毁开销。

get、join

当我们的CompletableFuture提供了返回值的时候，我们可以通过get或者join方法来阻塞的得到这个结果
与之不同是get他可能会抛出异常，而join不会。通常我们使用join

组合CompletableFuture

可以根据某种条件去执行两个或者多个CompletableFuture
因为组合太多，这里就简单描述下我自己比较常用的

方法	参数	描述
static allOf	CompletableFuture<?>…	所以任务都执行完成后完成，返回结果为void
static anyOf	CompletableFuture<?>…	任意任务都执行完成后完成，返回结果为void

示例

public static void main(String[] args) {// 1.两个任务都执行完成后才完成CompletableFuture.allOf(CompletableFuture.runAsync(()->{System.out.println("supplyAsync1");}),CompletableFuture.runAsync(()->{// 异步任务等待1秒try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("supplyAsync2");}));// 任意一个完成则完成CompletableFuture.anyOf(CompletableFuture.runAsync(()->{System.out.println("supplyAsync3");}),CompletableFuture.runAsync(()->{// 异步任务等待2秒try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("supplyAsync4");}));// 主线程只等待一秒 try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}
}### 结果
supplyAsync1
supplyAsync3
supplyAsync2