Java 异步编程：内置功能与三方库

1.引言

伴随着对非阻塞式编程需求的不断增长，各种编程语言都需要支持异步编程。

本文将介绍几种使用 Java 实现异步编程方法,还会介绍一些开箱即用的 Java 异步编程开发库。

2.Java 异步编程

2.1.线程

创建新线程可以实现异步执行操作。Java 8 引入了 lambda 表达式，让多线程代码更整洁可读性更强。

下面创建了一个新线程计算并输出阶乘：

int number = 20;Thread newThread = new Thread(() -> { System.out.println('Factorial of ' + number + ' is: ' + factorial(number));});newThread.start();

2.2.FutureTask

Java 5 开始，Future 接口供了 FutureTask 执行异步操作。

可以使用ExecutorService 的 submit 方法异步执行任务并返回 FutureTask 实例。

下面通过 FutureTask 计算阶乘：

ExecutorService threadpool = Executors.newCachedThreadPool();Future<Long> futureTask = threadpool.submit(() -> factorial(number)); while (!futureTask.isDone()) {    System.out.println('FutureTask is not finished yet...');}long result = futureTask.get(); threadpool.shutdown();

这里使用 Future 接口提供的 isDone方法检查任务是否完成。任务执行完，可以使用 get 方法获取计算结果。

2.3.CompletableFuture

Java 8 引入了 CompletableFuture，结合了 Future 和 CompletionStage 功能。它提供了诸如 supplyAsync、runAsync 和 thenApplyAsync这样的异步方法。

下面用 CompletableFuture 替代 FutureTask 计算阶乘：

CompletableFuture<Long> completableFuture = CompletableFuture.supplyAsync(() -> factorial(number));while (!completableFuture.isDone()) { System.out.println('CompletableFuture is not finished yet...');}long result = completableFuture.get();

不必显式调用 ExecutorService。CompletableFuture 内部采用 ForkJoinPool 异步执行任务。代码因此变得更整洁。

3.Guava

Guava 提供了 ListenableFuture 类用来执行异步操作。

首先，在项目中加入 guava 最新的 Maven 依赖：

<dependency>    <groupId>com.google.guava</groupId>    <artifactId>guava</artifactId>    <version>28.2-jre</version></dependency>

然后，使用 ListenableFuture 计算阶乘：

ExecutorService threadpool = Executors.newCachedThreadPool();ListeningExecutorService service = MoreExecutors.listeningDecorator(threadpool);ListenableFuture<Long> guavaFuture = (ListenableFuture<Long>) service.submit(()-> factorial(number));long result = guavaFuture.get();

这里，MoreExecutors 类提供了 ListeningExecutorService 类的实例。接着，ListeningExecutorService.submit 方法会异步执行任务并返回 ListenableFuture 实例。

Guava 还提供了 Futures 类，包含 submitAsync、scheduleAsync 和 transformAsync 方法，实现类似 CompletableFuture 对 ListenableFutures 进行的链式调用。

让我们看看如何使用 Futures.submitAsync 代替 ListeningExecutorService.submit 方法：

ListeningExecutorService service = MoreExecutors.listeningDecorator(threadpool);AsyncCallable<Long> asyncCallable = Callables.asAsyncCallable(new Callable<Long>() {    public Long call() {        return factorial(number);    }}, service);ListenableFuture<Long> guavaFuture = Futures.submitAsync(asyncCallable, service);

这里 submitAsync 方法需要传入 AsyncCallable 参数，后者由 Callables 类创建。

此外，Futures 类供了 addCallback 方法可以用来注册成功与失败回调：

Futures.addCallback( factorialFuture, new FutureCallback<Long>() { public void onSuccess(Long factorial) { System.out.println(factorial); } public void onFailure(Throwable thrown) { thrown.getCause(); } }, service);

4.EA Async

Electronic Arts 把 .NET async-await 功能引入了 Java，称作 ea-async。

该开发库支持顺序编写（非阻塞）异步代码。通过它进行异步编程不仅很容易，而且可以很自然地扩展。

首先，把 ea-async 最新的 Maven 依赖加入 pom.xml：

<dependency>    <groupId>com.ea.async</groupId>    <artifactId>ea-async</artifactId>    <version>1.2.3</version></dependency>

接着把之前 CompletableFuture 的代码改用 await 方法实现，后者由 Async 类提供：

static { Async.init();} public long factorialUsingEAAsync(int number) { CompletableFuture<Long> completableFuture = CompletableFuture.supplyAsync(() -> factorial(number)); long result = Async.await(completableFuture);}

这里，在 static 代码块中调用 Async.init 方法，对 Async 运行时进行初始化。

Async instrumentation 会在运行时转换代码并重写调用 await 方法，调用方式与 CompletableFuture 链式调用类似。

调用 await 方法与 Future.join 很像。

可以使用 -javaagent JVM 参数进行编译时检测。这是也是一种 Async.init 替代方案：

java -javaagent:ea-async-1.2.3.jar -cp <claspath> <MainClass>

让我们再来看一个顺序编写异步代码的例子。

首先，使用 CompletableFuture 中的组合方法，比如 thenComposeAsync 和 thenAcceptAsync 链式调用执行异步操作：

CompletableFuture<Void> completableFuture = hello() .thenComposeAsync(hello -> mergeWorld(hello)) .thenAcceptAsync(helloWorld -> print(helloWorld)) .exceptionally(throwable -> { System.out.println(throwable.getCause()); return null; });completableFuture.get();

然后，可以使用 EA Async.await() 进行代码转换：

try {    String hello = await(hello());    String helloWorld = await(mergeWorld(hello));    await(CompletableFuture.runAsync(() -> print(helloWorld)));} catch (Exception e) {    e.printStackTrace();}

这种实现类似于顺序阻塞式调用，但 await 并不会阻塞调用。

正如前面提到的，所有对 await 方法的调用都会被 Async instrumentation 重写，结果与 Future.join 方法类似。

因此，当异步执行的 hello 方法完成，Future 执行的结果会传给 mergeWorld 方法。接着把结果交给最后一步 CompletableFuture.runAsync 方法。

5.Cactoos

Cactoos 是一个面向对象的 Java 开发库。

它是 Google Guava 和 Apache Commons 的替代方案，提供了各种操作的通用对象。

首先，添加最新的 cactoos Maven 依赖：

<dependency> <groupId>org.cactoos</groupId> <artifactId>cactoos</artifactId> <version>0.43</version></dependency>

Cactoos 针对异步操作提供了 Async 类。

因此，可以用 Cactoos Async 计算阶乘：

Async<Integer, Long> asyncFunction = new Async<Integer, Long>(input -> factorial(input));Future<Long> asyncFuture = asyncFunction.apply(number);long result = asyncFuture.get();

这里 apply 方法会执行 ExecutorService.submit 并返回一个 Future 接口实例。

与之类似，Async 类的 exec 方法也提供了类似功能，区别在于没有返回值。

注意：Cactoos 库尚处于开发初期，可能不适合生产环境使用。

6.Jcabi-Aspects

Jcabi-Aspects 为异步编程提供了 @Async 注解，基于 AspectJ 的 AOP 切面技术。

首先，添加 jcabi-aspects 最新的 Maven 依赖：

<dependency> <groupId>com.jcabi</groupId> <artifactId>jcabi-aspects</artifactId> <version>0.22.6</version></dependency>

Jcabi-aspects 需要 AspectJ 运行时支持。因此还要添加 aspectjrt Maven 依赖：

<dependency>    <groupId>org.aspectj</groupId>    <artifactId>aspectjrt</artifactId>    <version>1.9.5</version></dependency>

接下来添加 jcabi-maven-plugin 插件，该插件使用 AspectJ 切面编织二进制文件。插件提供了 ajc goal，可以帮助我们完成所有工作：

<plugin> <groupId>com.jcabi</groupId> <artifactId>jcabi-maven-plugin</artifactId> <version>0.14.1</version> <executions> <execution> <goals> <goal>ajc</goal> </goals> </execution> </executions> <dependencies> <dependency> <groupId>org.aspectj</groupId> <artifactId>aspectjtools</artifactId> <version>1.9.1</version> </dependency> <dependency> <groupId>org.aspectj</groupId> <artifactId>aspectjweaver</artifactId> <version>1.9.1</version> </dependency> </dependencies></plugin>

准备就绪，可以用 AOP 切面开始异步开发：

@Async@Loggablepublic Future<Long> factorialUsingAspect(int number) {    Future<Long> factorialFuture = CompletableFuture.completedFuture(factorial(number));    return factorialFuture;}

编译代码时，Jcabi-Aspects 将为 @Async 注解通过 AspectJ 编织注入 AOP advice，进而调用 factorialUsingAspect 异步方法。

使用 Maven 命令编译：

mvn install

jcabi-maven-plugin 输出结果像下面这样：

--- jcabi-maven-plugin:0.14.1:ajc (default) @ java-async ---[INFO] jcabi-aspects 0.18/55a5c13 started new daemon thread jcabi-loggable for watching of @Loggable annotated methods[INFO] Unwoven classes will be copied to /tutorials/java-async/target/unwoven[INFO] jcabi-aspects 0.18/55a5c13 started new daemon thread jcabi-cacheable for automated cleaning of expired @Cacheable values[INFO] ajc result: 10 file(s) processed, 0 pointcut(s) woven, 0 error(s), 0 warning(s)

可以通过检查 Maven 插件生成的 jcabi-ajc.log 中的日志来验证我们的类是否编织正确：

Join point 'method-execution(java.util.concurrent.Futurecom.baeldung.async.JavaAsync.factorialUsingJcabiAspect(int))' in Type 'com.baeldung.async.JavaAsync' (JavaAsync.java:158)advised by around advice from 'com.jcabi.aspects.aj.MethodAsyncRunner' (jcabi-aspects-0.22.6.jar!MethodAsyncRunner.class(from MethodAsyncRunner.java))

然后，作为一个简单的 Java 应用运行，结果如下：

17:46:58.245 [main] INFO com.jcabi.aspects.aj.NamedThreads -jcabi-aspects 0.22.6/3f0a1f7 started new daemon thread jcabi-loggable for watching of @Loggable annotated methods17:46:58.355 [main] INFO com.jcabi.aspects.aj.NamedThreads -jcabi-aspects 0.22.6/3f0a1f7 started new daemon thread jcabi-async for Asynchronous method execution17:46:58.358 [jcabi-async] INFO com.baeldung.async.JavaAsync -#factorialUsingJcabiAspect(20): 'java.util.concurrent.CompletableFuture@14e2d7c1[Completed normally]' in 44.64µs

可以看到 Jcabi-Aspects 创建了一个新的 jcabi-async 守护线程执行异步任务。

同样，通过开发库提供的 @Loggable 注解可以启用日志。

7.总结

本文介绍了 Java 异步编程的几种方法。

首先探索了 Java 内置的异步编程功能，例如 FutureTask 和 CompletableFuture。接着介绍了一些提供了开箱即用异步编程解决方案的开发库，例如 EA Async 和 Cactoos。

此外，还讨论了如何使用 Guava 中的 ListenableFuture 和 Future 类异步执行任务。最后探索了jcabi-AspectJ，该库通过 @Async 注解基于 AOP 功能执行异步调用。