Java CompletableFuture thenRun轻量级任务后置处理实践
Java CompletableFuture thenRun 概述
在Java的并发编程领域中，CompletableFuture 是一个强大的工具，它为异步编程提供了丰富的功能。thenRun 是 CompletableFuture 类中的一个方法，用于在 CompletableFuture 完成时执行一个无返回值的任务。
从本质上讲，CompletableFuture 代表一个异步计算的结果，它可以在计算完成后触发一系列后续操作。thenRun 方法允许我们在 CompletableFuture 正常完成（即没有抛出异常）时，安排一个新的任务去执行。这个新任务并不会关心 CompletableFuture 的计算结果，它只专注于在计算完成后执行一些收尾工作，比如清理资源、记录日志等轻量级的操作。
方法签名与原理剖析
thenRun 方法的签名如下：
public CompletableFuture<Void> thenRun(Runnable action)
它接受一个 Runnable 类型的参数 action，这个 action 就是我们要在 CompletableFuture 完成后执行的任务。该方法返回一个新的 CompletableFuture，这个新的 CompletableFuture 在 action 执行完毕后完成，并且其结果为 null（因为 Runnable 没有返回值）。
当 CompletableFuture 完成时（无论是正常完成还是因异常而完成），会检查是否有通过 thenRun 注册的 Runnable 任务。如果有，那么这个任务会被提交到一个默认的 ForkJoinPool.commonPool() 中执行（如果没有显式设置自定义的执行器）。如果 CompletableFuture 已经完成，那么 Runnable 任务会立即被提交执行。
简单代码示例
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class ThenRunExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture.supplyAsync(() -> {
// 模拟一些耗时操作
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "任务完成";
}).thenRun(() -> {
System.out.println("后置处理任务执行，不关心前面任务的结果");
}).get();
}
}
在上述代码中，首先通过 supplyAsync 创建了一个异步任务，这个任务模拟了一个耗时2秒的操作，并返回字符串 “任务完成”。然后通过 thenRun 注册了一个后置处理任务，这个任务仅仅打印一条日志，表明后置处理任务执行了，并且它并不关心前面异步任务的返回结果。最后通过 get 方法等待整个异步操作链完成。
结合复杂异步任务的应用
在实际应用中，我们可能会有多个异步任务，并且在这些任务完成后需要进行一些统一的后置处理。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ComplexThenRunExample {
private static final ExecutorService executor = Executors.newFixedThreadPool(3);
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
// 模拟一些耗时操作
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "任务1完成";
}, executor);
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
// 模拟一些耗时操作
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "任务2完成";
}, executor);
CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2);
allFutures.thenRun(() -> {
try {
System.out.println("任务1结果: " + future1.get());
System.out.println("任务2结果: " + future2.get());
System.out.println("所有任务完成，进行后置处理");
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}).get();
executor.shutdown();
}
}
在这段代码中，我们创建了两个异步任务 future1 和 future2，分别模拟了3秒和2秒的耗时操作。通过 CompletableFuture.allOf 方法等待这两个任务都完成。然后通过 thenRun 注册了一个后置处理任务，在这个后置处理任务中，我们获取了前面两个任务的结果并打印，同时表明所有任务完成并进行后置处理。这里使用了自定义的线程池 executor 来执行异步任务，在程序结束时关闭线程池。
异常处理与 thenRun 的关系
当 CompletableFuture 在执行过程中抛出异常时，通过 thenRun 注册的后置处理任务将不会被执行。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class ExceptionAndThenRunExample {
public static void main(String[] args) {
CompletableFuture.supplyAsync(() -> {
// 模拟抛出异常
throw new RuntimeException("任务执行出错");
}).thenRun(() -> {
System.out.println("后置处理任务执行");
}).exceptionally(ex -> {
System.out.println("捕获到异常: " + ex.getMessage());
return null;
}).join();
}
}
在上述代码中，异步任务抛出了一个运行时异常。由于异常的存在，thenRun 注册的任务不会执行。而通过 exceptionally 方法，我们捕获了这个异常并进行了相应的处理，打印出异常信息。
thenRun 与其他 CompletableFuture 方法的组合使用
thenRun 可以与其他 CompletableFuture 方法灵活组合，以实现更复杂的异步操作。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompositeThenRunExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture.supplyAsync(() -> "初始任务结果")
.thenApply(result -> result + " 经过转换")
.thenAccept(System.out::println)
.thenRun(() -> System.out.println("后置处理任务执行"))
.get();
}
}
在这段代码中，首先通过 supplyAsync 创建了一个异步任务并返回 “初始任务结果”。接着通过 thenApply 对结果进行转换，添加了 “ 经过转换”。然后通过 thenAccept 打印转换后的结果。最后通过 thenRun 执行后置处理任务，打印 “后置处理任务执行”。这种组合使用展示了 CompletableFuture 强大的链式调用能力，可以按照顺序执行一系列异步操作和后置处理。
自定义执行器与 thenRun
在默认情况下，thenRun 注册的任务会提交到 ForkJoinPool.commonPool() 中执行。但我们也可以通过使用带执行器参数的 thenRun 方法来指定自定义的执行器。
import java.util.concurrent.*;
public class CustomExecutorThenRunExample {
public static void main(String[] args) {
ExecutorService customExecutor = Executors.newSingleThreadExecutor();
CompletableFuture.supplyAsync(() -> "任务结果", customExecutor)
.thenRun(() -> System.out.println("后置处理任务执行"), customExecutor);
customExecutor.shutdown();
}
}
在上述代码中，我们创建了一个单线程的自定义执行器 customExecutor。通过 supplyAsync 方法使用这个自定义执行器来执行异步任务，并且在 thenRun 方法中也指定了这个自定义执行器来执行后置处理任务。在程序结束时关闭自定义执行器。
thenRun 应用场景
- 资源清理：在异步任务完成后，可能需要关闭文件、释放数据库连接等资源。例如，在读取文件的异步任务完成后，关闭文件流。
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
public class ResourceCleanupWithThenRun {
public static void main(String[] args) {
CompletableFuture.supplyAsync(() -> {
BufferedReader reader = null;
try {
reader = new BufferedReader(new FileReader("example.txt"));
return reader.readLine();
} catch (IOException e) {
e.printStackTrace();
return null;
} finally {
if (reader != null) {
try {
reader.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}).thenRun(() -> {
System.out.println("文件读取任务完成，资源已清理");
});
}
}
在这个例子中，虽然在异步任务内部已经进行了文件流的关闭操作，但通过 thenRun 可以额外记录资源清理完成的信息，使代码的执行流程更加清晰。
- 日志记录：在异步任务完成后，记录任务的执行情况，包括任务是否成功、耗时等信息。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class LoggingWithThenRun {
public static void main(String[] args) throws ExecutionException, InterruptedException {
long startTime = System.currentTimeMillis();
CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "任务成功";
}).thenRun(() -> {
long endTime = System.currentTimeMillis();
System.out.println("任务完成，耗时: " + (endTime - startTime) + " 毫秒");
}).get();
}
}
在上述代码中，通过 thenRun 记录了任务的执行耗时，方便对异步任务的性能进行监控和分析。
- 触发其他异步任务：在某个异步任务完成后，触发另一个轻量级的异步任务。例如，在订单处理完成后，异步发送通知。
import java.util.concurrent.CompletableFuture;
public class TriggerAnotherTaskWithThenRun {
public static void main(String[] args) {
CompletableFuture.supplyAsync(() -> {
// 模拟订单处理
System.out.println("订单处理中...");
return "订单处理完成";
}).thenRun(() -> {
CompletableFuture.runAsync(() -> {
System.out.println("发送订单处理完成通知...");
});
});
}
}
在订单处理异步任务完成后，通过 thenRun 触发了一个新的异步任务来发送通知，这种方式可以有效地解耦不同的异步操作。
性能考虑
虽然 thenRun 非常方便，但在高并发场景下，需要考虑性能问题。由于默认情况下 thenRun 注册的任务会提交到 ForkJoinPool.commonPool() 中执行，如果这个公共线程池中的任务过多，可能会导致任务执行的延迟。
一种优化方式是使用自定义的线程池，并根据实际的业务需求调整线程池的大小。例如，如果后置处理任务是 CPU 密集型的，那么线程池大小可以设置为 CPU 核心数；如果是 I/O 密集型的，可以适当增大线程池大小。
另外，尽量减少后置处理任务中的耗时操作，因为 thenRun 主要用于轻量级的任务处理。如果后置处理任务非常耗时，可能需要考虑将其拆分成更小的任务或者使用其他更适合的异步处理方式，以避免影响整个异步操作链的性能。
通过合理地使用 thenRun 方法，并结合性能优化措施，我们可以在 Java 异步编程中有效地进行轻量级任务的后置处理，提高程序的并发性能和代码的可读性。在实际项目中，根据不同的业务场景灵活运用 thenRun 与其他 CompletableFuture 方法，能够构建出高效、健壮的异步应用程序。无论是简单的资源清理、日志记录，还是复杂的任务链组合，thenRun 都为我们提供了一种简洁而强大的异步任务后置处理手段。同时，通过对异常处理、执行器选择等方面的深入理解和优化，我们可以进一步提升应用程序在高并发环境下的稳定性和性能表现。