Java CompletableFuture 中 怎么实现流水线处理
在 java 中，completablefuture 的流水线处理主要通过链式调用 thenapply()、thencompose() 和 thenaccept() 来实现。核心做法是将每个处理步骤封装为一个独立的异步任务，并通过前一个任务的完成来触发下一个任务，从而形成一条自动流转的处理链。这种方式避免了手动管理线程池和回调嵌套，使代码逻辑线性化且易于维护。
同步转换与异步组合的区别是什么
同步转换使用 thenApply()，它会在当前阶段完成后立即执行映射函数，适用于轻量级计算；异步组合使用 thenCompose()，它接收一个返回 CompletableFuture 的函数，用于串联多个异步操作，适用于 I/O 密集型任务或需要独立线程调度的场景。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class PipelineDemo {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(4);
// 模拟原始数据输入
String input = "Hello";
// 构建流水线：输入 -> 处理1 -> 处理2 -> 输出
CompletableFuture<String> result = CompletableFuture.supplyAsync(() -> {
System.out.println("Step 0: Input received");
return input;
}, executor)
.thenApply(s -> {
System.out.println("Step 1: Transforming (Sync)");
return s.toUpperCase();
})
.thenCompose(s -> {
System.out.println("Step 2: Async Processing (IO Simulation)");
return CompletableFuture.supplyAsync(() -> {
try { Thread.sleep(100); } catch (InterruptedException e) {}
return s + " World";
}, executor);
})
.thenAccept(finalResult -> {
System.out.println("Final Result: " + finalResult);
});
// 阻塞等待结果
result.join();
executor.shutdown();
}
}运行上述代码，控制台将按顺序输出 Step 0、Step 1、Step 2 以及最终结果。这证明了 thenApply 是同步执行的（在当前线程或上一阶段线程），而 thenCompose 内部触发的异步任务会在新线程中执行。
立即学习“Java免费学习笔记（深入）”；
如何处理流水线中的异常中断
当流水线中某个节点抛出异常时，后续的正常步骤会被跳过，但可以通过 exceptionally() 或 handle() 进行捕获和处理。推荐使用 exceptionally() 提供默认值或降级逻辑，确保流水线不会因单个错误而静默失败，同时保持主流程的简洁性。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class ErrorHandlingDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 模拟一个会失败的步骤
CompletableFuture<string> faultyStage = CompletableFuture.supplyAsync(() -> {
throw new RuntimeException("Processing failed!");
});
// 添加异常处理分支
CompletableFuture<string> safePipeline = faultyStage
.thenApply(s -> s.toUpperCase()) // 如果上面失败，这里不会执行
.exceptionally(ex -> {
System.out.println("Caught exception: " + ex.getMessage());
return "Default Value"; // 提供默认返回值，恢复流水线
})
.thenApply(s -> s +</string></string> 
