背景
Java异步编程离不开Future
接口,但是Future
接口提供的方法使用起来不够灵活。为了判断一个Future
是否已经完成,我们可以:
- 通过调用
get
方法,以阻塞的形式获取执行结果。 - 调用有最大等待时间的
get
方法,以阻塞的形式获取执行结果。 - 反复轮询
isDone
方法,直到任务完成,获取结果。
以上3种方法都不够灵活,会造成线程阻塞或耗费CPU资源。需要用户过多的参与异步编程逻辑,对业务代码的侵入性较强。
另外用户如果需要异步执行多个任务,并且这些任务具有先后依赖关系。基于传统的方式我们需要大量使用锁,CountDownLatch
和CyclicBarrier
和阻塞队列等,编程十分复杂。
CompletableFuture
是Future
的增强版,提供了一系列的同步或异步任务执行操作。除此之外还能够对异步任务多个阶段的前后依赖关系进行控制。使用起来十分方便。
注意事项
所有的以async结尾的方法都为异步执行。它们都可以传入一个Executor线程池,用于异步执行。如果不指定线程池,默认使用ForkJoinPool.commonPool()
下面以所有的async异步方法为例,说明下CompletableFuture的使用方法。
静态方法
supplyAsync
提供初始数据。接收一个Supplier类型参数。
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> 1);
completableFuture.thenApplyAsync((i) -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return i + 1;
// 这里打印出2
}).thenAcceptAsync((i) -> System.out.println(i));
runAsync
异步执行任务,接收Runnable类型参数。
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("runAsync");
});
completedFuture
直接返回一个已经完成的CompletableFuture。接收一个任意类型数据为参数。
CompletableFuture<Integer> completableFuture = CompletableFuture.completedFuture(500);
// 此处可以立刻返回500
System.out.println(completableFuture.get());
allOf
参数接收任意多个CompletableFuture。该方法返回一个新的CompletableFuture,只有在参数所有的CompletableFuture完成的时候它才会完成。
long start = System.currentTimeMillis();
CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("runAsync1");
});
CompletableFuture<Void> completableFuture2 = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("runAsync2");
});
// 等待两个CompletableFuture都完成的时候,future才会完成
CompletableFuture<Void> future = CompletableFuture.allOf(completableFuture1, completableFuture2);
// 阻塞到CompletableFuture1完成
future.get();
// 耗时大于3000毫秒
System.out.println(System.currentTimeMillis() - start);
anyOf
和allOf一样,返回一个新的CompletableFuture,参数中任意一个CompletableFuture完成时它就能完成。
long start = System.currentTimeMillis();
CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("runAsync1");
});
CompletableFuture<Void> completableFuture2 = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("runAsync2");
});
// 两个CompletableFuture有任何一个完成,future就可以完成
CompletableFuture<Object> future = CompletableFuture.anyOf(completableFuture1, completableFuture2);
// 阻塞到CompletableFuture2完成
future.get();
// 耗时略大于1000毫秒
System.out.println(System.currentTimeMillis() - start);
普通方法
thenApplyAsync
进行数据处理,接收前一步骤传递的数据,处理加工后返回。返回数据类型可以和前一步骤返回的数据类型不同。
接收参数为Function类型。
CompletableFuture.supplyAsync(() -> 1).thenApplyAsync((i) -> "Hi: " + i).whenCompleteAsync(((s, throwable) -> {
// 返回 Hi: 1
System.out.println(s);
}));
thenAcceptAsync
接收上游传递过来的数据并消费。接收一个Consumer类型参数。
CompletableFuture.supplyAsync(() -> 1).thenAcceptAsync(System.out::println);
acceptEitherAsync
接收另一个CompletableFuture和一个Consumer。含义为2个CompletableFuture哪个先运行完成,就采用谁的执行结果。
CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
}).acceptEitherAsync(completableFuture1, integer -> {
// completableFuture1最早返回,所以integer的值为1
System.out.println(integer);
});
这个例子中第一个CompletableFuture
比第二个先执行完毕,因此acceptEitherAsync
输出第一个CompletableFuture
的结果。
applyToEitherAsync
接收一个CompletableFuture和一个Function。和acceptEitherAsync
类似,只不过第二个参数从Consumer变成Function。
CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
}).applyToEitherAsync(completableFuture1, (integer -> {
// 这里CompletableFuture1首先完成,所以integer为1
return integer;
})).whenCompleteAsync((integer, throwable) -> {
System.out.println(integer);
});
thenAcceptBothAsync
接收另一个CompletableFuture和BiConsumer,用于在两个CompletableFuture都执行完的时候,获取他们的执行结果并处理。
CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
}).thenAcceptBothAsync(completableFuture1, (integer, integer2) -> {
// 等到两个CompletableFuture都完成之后回调
// 在这个例子中需要等待5秒钟
回调的两个参数分别对应两个CompletableFuture的执行结果
System.out.println(integer + integer2);
});
runAfterEitherAsync
接收另一个CompletableFuture和Runnable类型参数。两个CompletableFuture有任意一个完成的时候,执行Runnable。
long start = System.currentTimeMillis();
CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
}).runAfterEitherAsync(completableFuture1, () -> {
// 任何一个CompletableFuture完成的时候都会执行Runnable
// 这个例子等待1秒后会执行Runnable,此处打印的值略大于1000
System.out.println(System.currentTimeMillis() - start);
});
runAfterBothAsync
接收另一个CompletableFuture和Runnable类型参数。两个CompletableFuture都执行完成的时候,执行Runnable。
long start = System.currentTimeMillis();
CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
}).runAfterBothAsync(completableFuture1, () -> {
// 和runAfterEitherAsync不同的是,这里必须等到两个CompletableFuture都完成的时候才会执行Runnable
// 这里需要等待5秒钟后才会执行,打印的值略大于5000
System.out.println(System.currentTimeMillis() - start);
});
complete
接收任意类型数据作为参数。如果CompletableFuture没有完成的时候调用complete()方法,后续再调用这个future的get()方法时返回这个值。
// 睡眠2秒再返回结果
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
// 注意,此处如果睡眠3秒,调用complete时completableFuture已经执行完毕返回1,complete方法不会修改返回值。此时调用get方法返回1
// 如果此处没有睡眠3秒,调用complete时completableFuture尚未执行完毕,下面调用get的时候方法返回1000
Thread.sleep(3000);
completableFuture.complete(1000);
Integer integer = completableFuture.get();
System.out.println(integer);
completeExceptionally
接收一个Throwable类型参数。如果CompletableFuture没有完成的时候调用completeExceptionally()方法,后续再调用这个future的get()方法时会抛出这个异常。
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
// 如果调用completeExceptionally的时候completableFuture没有执行完成,那么下面调用get的时候会抛出IllegalArgumentException
Thread.sleep(3000);
completableFuture.completeExceptionally(new IllegalArgumentException());
Integer integer = completableFuture.get();
System.out.println(integer);
cancel
相当于completeExceptionally(new CancellationException())
,不再赘述。接收的参数对行为没有影响。
如果CompletableFuture没有执行完成的时候调用了cancel,cancel方法返回true。
如果CompletableFuture执行完成的时候调用了cancel,cancel方法返回false。
exceptionally
设置当CompletableFuture执行抛出异常时候的返回值。用于处理异常情况。
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
// 抛出一个异常
if (true) {
throw new RuntimeException("Error Message");
}
return 1;
});
// 如果遇到异常,返回500
completableFuture.exceptionally((t) -> 500)
.whenCompleteAsync(((integer, throwable) -> {
// 打印出500
System.out.println(integer);
// 异常已被处理,返回null
System.out.println(throwable.getMessage());
}));
handleAsync
相比exceptionally
更为复杂的处理exception方式。接收一个BiFunction类型的参数。
handleAsync
也可以视为回调方法具有返回值的whenCompleteAsync
。
注意和exceptionally
的区别,handleAsync
没有自动发现是否抛出exception的能力,需要手工编写相关逻辑。
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
if (true) {
throw new RuntimeException("Error Message");
}
return 1;
});
completableFuture
.handleAsync((integer, throwable) -> {
System.out.println(integer);
System.out.println(throwable);
return null == throwable ? 1 : 500;
})
.whenCompleteAsync(((integer, throwable) -> {
System.out.println(integer);
System.out.println(throwable.getMessage());
}));
whenCompleteAsync
完成时异步回调,接收一个BiConsumer<? super T, ? super Throwable>
类型参数。
例子在前面的代码中已有体现,不再赘述。
thenCombineAsync
含义相当于"thenApplyBothAsync",但是JDK不叫这个名字。
接收另一个CompletableFuture和BiFunction作为参数。目的为等待两个CompletableFuture都完成的时候,执行BiFunction对两个CompletableFuture的返回值进行处理。
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
return 1;
});
CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
});
completableFuture.thenCombineAsync(completableFuture2, (integer, integer2) -> {
// 等两个CompletableFuture都执行完毕后,返回他们两个返回值的和
return integer + integer2;
}).whenCompleteAsync(((integer, throwable) -> {
System.out.println(integer);
}));
thenComposeAsync
将两个CompletableFuture组合起来。接收的参数类型为Function。
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
return 1;
});
completableFuture.thenComposeAsync(integer -> {
// 这里返回另一个CompletableFuture,该CompletableFuture可以使用前面CompletableFuture的返回值进行计算
return CompletableFuture.supplyAsync(() -> {
return 10 + integer;
});
}).whenCompleteAsync(((integer, throwable) -> {
// 返回11
System.out.println(integer);
}));