CompletableFuture 与异步编排

// 传统 Future 方式
ExecutorService executor = Executors.newFixedThreadPool(10);
Future<String> future = executor.submit(() -> {
    // 异步任务
    Thread.sleep(1000);
    return "结果";
});

// 问题1:阻塞获取结果
String result = future.get();  // 阻塞等待

// 问题2:无法编排后续操作
// future.get() 后才能继续处理,不能链式调用

// 问题3:异常处理不友好
try {
    String result = future.get();
} catch (ExecutionException e) {
    // 异常被包装在 ExecutionException 中
}

// 回调地狱示例
serviceA.getData(new Callback<String>() {
    @Override
    public void onSuccess(String resultA) {
        serviceB.process(resultA, new Callback<String>() {
            @Override
            public void onSuccess(String resultB) {
                serviceC.save(resultB, new Callback<String>() {
                    @Override
                    public void onSuccess(String resultC) {
                        // 层层嵌套,代码难以维护
                    }
                    
                    @Override
                    public void onError(Exception e) {
                        // 异常处理混乱
                    }
                });
            }
            
            @Override
            public void onError(Exception e) {
                // 异常处理混乱
            }
        });
    }
    
    @Override
    public void onError(Exception e) {
        // 异常处理混乱
    }
});

// CompletableFuture 方式
CompletableFuture<String> result = CompletableFuture
    .supplyAsync(() -> serviceA.getData())
    .thenApply(resultA -> serviceB.process(resultA))
    .thenApply(resultB -> serviceC.save(resultB))
    .exceptionally(ex -> "默认值");

// 清晰、可读、易维护

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SupplyAsyncExample {
    public static void main(String[] args) {
        // 使用默认线程池(ForkJoinPool.commonPool())
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行异步任务,线程: " + Thread.currentThread().getName());
            return "结果1";
        });
        
        // 使用自定义线程池(推荐)
        ExecutorService executor = Executors.newFixedThreadPool(10);
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行异步任务,线程: " + Thread.currentThread().getName());
            return "结果2";
        }, executor);
        
        // 获取结果
        System.out.println(future1.join());  // "结果1"
        System.out.println(future2.join());  // "结果2"
        
        executor.shutdown();
    }
}

// 无返回值的异步任务
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    System.out.println("执行无返回值的异步任务");
});

// 创建已完成的 Future
CompletableFuture<String> completed = CompletableFuture.completedFuture("已完成");

// 创建未完成的 Future
CompletableFuture<String> future = new CompletableFuture<>();

// 手动完成
future.complete("手动设置结果");

// 手动完成异常
future.completeExceptionally(new RuntimeException("异常"));

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "hello")
    .thenApply(result -> result + " world")  // 同步转换
    .thenApply(String::toUpperCase);         // 继续转换

System.out.println(future.join());  // "HELLO WORLD"

ExecutorService executor = Executors.newFixedThreadPool(10);

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    System.out.println("步骤1,线程: " + Thread.currentThread().getName());
    return "hello";
}, executor)
    .thenApplyAsync(result -> {
        System.out.println("步骤2,线程: " + Thread.currentThread().getName());
        return result + " world";
    }, executor)
    .thenApplyAsync(result -> {
        System.out.println("步骤3,线程: " + Thread.currentThread().getName());
        return result.toUpperCase();
    }, executor);

System.out.println(future.join());

executor.shutdown();

// thenCompose:扁平化嵌套的 CompletableFuture
CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "userId-123")
    .thenCompose(userId -> getUserInfo(userId))  // 返回 CompletableFuture<User>
    .thenCompose(user -> getPermissions(user.getId()))
    .thenApply(permissions -> "权限: " + permissions);

// 辅助方法
private CompletableFuture<User> getUserInfo(String userId) {
    return CompletableFuture.supplyAsync(() -> {
        // 模拟查询用户信息
        return new User(userId, "张三");
    });
}

private CompletableFuture<String> getPermissions(String userId) {
    return CompletableFuture.supplyAsync(() -> {
        // 模拟查询权限
        return "admin,user";
    });
}

// thenApply:同步转换,返回 T -> R
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "hello")
    .thenApply(result -> result + " world");

// thenCompose:异步链式,返回 T -> CompletableFuture<R>
CompletableFuture<User> future2 = CompletableFuture.supplyAsync(() -> "userId")
    .thenCompose(userId -> getUserInfo(userId));  // 返回 CompletableFuture<User>

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThenCombineExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        // 任务1:查询商品信息
        CompletableFuture<Product> productFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品信息");
            sleep(1000);
            return new Product("iPhone 15", 5999);
        }, executor);
        
        // 任务2:查询库存
        CompletableFuture<Integer> stockFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询库存");
            sleep(800);
            return 100;
        }, executor);
        
        // 合并两个任务结果
        CompletableFuture<ProductDetail> detailFuture = productFuture.thenCombine(
            stockFuture,
            (product, stock) -> new ProductDetail(product, stock)
        );
        
        ProductDetail detail = detailFuture.join();
        System.out.println(detail);
        
        executor.shutdown();
    }
    
    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class Product {
    private String name;
    private int price;
    
    public Product(String name, int price) {
        this.name = name;
        this.price = price;
    }
    
    public String getName() { return name; }
    public int getPrice() { return price; }
}

class ProductDetail {
    private Product product;
    private int stock;
    
    public ProductDetail(Product product, int stock) {
        this.product = product;
        this.stock = stock;
    }
    
    @Override
    public String toString() {
        return String.format("商品: %s, 价格: %d, 库存: %d", 
            product.getName(), product.getPrice(), stock);
    }
}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AllOfExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        List<CompletableFuture<String>> futures = new ArrayList<>();
        
        // 提交多个任务
        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            futures.add(CompletableFuture.supplyAsync(() -> {
                System.out.println("任务 " + taskId + " 开始执行");
                sleep(1000);
                return "结果-" + taskId;
            }, executor));
        }
        
        // 等待所有任务完成
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
        
        // 获取所有结果
        CompletableFuture<List<String>> resultsFuture = allFutures.thenApply(v -> {
            List<String> results = new ArrayList<>();
            for (CompletableFuture<String> future : futures) {
                results.add(future.join());
            }
            return results;
        });
        
        List<String> results = resultsFuture.join();
        System.out.println("所有任务完成: " + results);
        
        executor.shutdown();
    }
    
    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AnyOfExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        // 从多个数据源查询,谁快用谁
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            sleep(1500);
            return "数据源1";
        }, executor);
        
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            return "数据源2";
        }, executor);
        
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            sleep(2000);
            return "数据源3";
        }, executor);
        
        // 任意一个完成就返回
        CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(
            future1, future2, future3
        );
        
        String result = (String) anyFuture.join();
        System.out.println("最快返回: " + result);  // "数据源2"
        
        executor.shutdown();
    }
    
    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("模拟异常");
    }
    return "正常结果";
}).exceptionally(ex -> {
    System.out.println("发生异常: " + ex.getMessage());
    return "默认值";  // 异常兜底
});

System.out.println(future.join());

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("模拟异常");
    }
    return "正常结果";
}).handle((result, ex) -> {
    if (ex != null) {
        System.out.println("发生异常: " + ex.getMessage());
        return "异常处理后的默认值";
    }
    return result + " - 处理后";
});

System.out.println(future.join());

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("模拟异常");
    }
    return "正常结果";
}).whenComplete((result, ex) -> {
    if (ex != null) {
        System.out.println("任务失败: " + ex.getMessage());
    } else {
        System.out.println("任务成功: " + result);
    }
});

// whenComplete 不影响最终结果
System.out.println(future.join());

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ProductDetailAggregator {
    private ExecutorService executor = Executors.newFixedThreadPool(20);
    
    public ProductDetailVO getProductDetail(Long productId, Long userId) {
        // 并发查询多个下游服务
        
        // 1. 查询商品基础信息(必须成功)
        CompletableFuture<ProductDTO> productFuture = CompletableFuture.supplyAsync(
            () -> productService.getProduct(productId), executor
        );
        
        // 2. 查询库存(必须成功)
        CompletableFuture<Integer> stockFuture = CompletableFuture.supplyAsync(
            () -> inventoryService.getStock(productId), executor
        );
        
        // 3. 查询营销活动(可降级)
        CompletableFuture<ActivityDTO> activityFuture = CompletableFuture.supplyAsync(
            () -> marketingService.getActivity(productId), executor
        ).exceptionally(ex -> ActivityDTO.empty());  // 失败时返回空对象
        
        // 4. 查询用户推荐(可降级)
        CompletableFuture<List<ProductDTO>> recommendFuture = CompletableFuture.supplyAsync(
            () -> recommendService.getRecommendations(userId), executor
        ).exceptionally(ex -> Collections.emptyList());  // 失败时返回空列表
        
        // 等待所有查询完成并聚合结果
        CompletableFuture<ProductDetailVO> resultFuture = CompletableFuture
            .allOf(productFuture, stockFuture, activityFuture, recommendFuture)
            .thenApply(v -> {
                ProductDetailVO detail = new ProductDetailVO();
                detail.setProduct(productFuture.join());
                detail.setStock(stockFuture.join());
                detail.setActivity(activityFuture.join());
                detail.setRecommendations(recommendFuture.join());
                return detail;
            })
            .orTimeout(3, TimeUnit.SECONDS)  // 整体超时3秒
            .exceptionally(ex -> {
                log.error("查询商品详情失败", ex);
                return ProductDetailVO.empty();  // 返回空对象
            });
        
        return resultFuture.join();
    }
    
    // 其他方法省略...
}

public class OrderService {
    private ExecutorService executor = Executors.newFixedThreadPool(20);
    
    public OrderCreateResult createOrder(OrderCreateRequest request) {
        // 步骤1:校验用户(异步)
        CompletableFuture<UserDTO> userFuture = CompletableFuture.supplyAsync(
            () -> userService.checkUser(request.getUserId()), executor
        );
        
        // 步骤2:扣减库存(依赖步骤1,异步)
        CompletableFuture<Void> stockFuture = userFuture.thenAcceptAsync(
            user -> inventoryService.deductStock(request.getProductId(), request.getQuantity()),
            executor
        );
        
        // 步骤3:生成订单(依赖步骤1和2,异步)
        CompletableFuture<Order> orderFuture = userFuture.thenCombineAsync(
            stockFuture,
            (user, v) -> orderService.createOrder(request),
            executor
        );
        
        // 步骤4:发送通知(依赖步骤3,异步,可降级)
        CompletableFuture<Void> notifyFuture = orderFuture.thenAcceptAsync(
            order -> {
                try {
                    notificationService.sendOrderNotification(order);
                } catch (Exception e) {
                    log.warn("发送通知失败", e);  // 失败不影响主流程
                }
            },
            executor
        );
        
        // 步骤5:返回结果(依赖步骤3和4)
        CompletableFuture<OrderCreateResult> resultFuture = orderFuture
            .thenCombineAsync(
                notifyFuture,
                (order, v) -> new OrderCreateResult(order.getId(), "订单创建成功"),
                executor
            )
            .orTimeout(5, TimeUnit.SECONDS)
            .exceptionally(ex -> {
                log.error("订单创建失败", ex);
                throw new BusinessException("订单创建失败: " + ex.getMessage());
            });
        
        return resultFuture.join();
    }
}

public class BatchProcessor {
    private ExecutorService executor = Executors.newFixedThreadPool(50);
    
    public List<TaskResult> processBatch(List<TaskRequest> requests) {
        // 并行处理多个任务
        List<CompletableFuture<TaskResult>> futures = requests.stream()
            .map(request -> CompletableFuture.supplyAsync(
                () -> processTask(request), executor
            ))
            .collect(Collectors.toList());
        
        // 等待所有任务完成
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );
        
        // 汇总结果
        CompletableFuture<List<TaskResult>> resultsFuture = allFutures.thenApply(v ->
            futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
        );
        
        return resultsFuture.join();
    }
    
    private TaskResult processTask(TaskRequest request) {
        // 处理单个任务
        return new TaskResult(request.getId(), "成功");
    }
}

public class TimeoutExample {
    private ExecutorService executor = Executors.newFixedThreadPool(10);
    
    public String getDataWithTimeout() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟耗时操作
            sleep(3000);
            return "正常结果";
        }, executor)
        .orTimeout(2, TimeUnit.SECONDS)  // 超时2秒
        .exceptionally(ex -> {
            if (ex instanceof TimeoutException) {
                log.warn("操作超时,返回默认值");
                return "默认值";  // 超时降级
            }
            log.error("操作异常", ex);
            return "异常默认值";
        });
        
        return future.join();
    }
    
    // Java 8 方式(使用 CompletableFuture.get)
    public String getDataWithTimeoutJava8() {
        try {
            return CompletableFuture.supplyAsync(() -> {
                sleep(3000);
                return "正常结果";
            }, executor).get(2, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            return "默认值";  // 超时降级
        } catch (Exception e) {
            return "异常默认值";
        }
    }
}

import java.util.concurrent.*;

public class ThreadPoolConfig {
    // CPU 密集型任务线程池
    public static final ExecutorService CPU_INTENSIVE = new ThreadPoolExecutor(
        Runtime.getRuntime().availableProcessors(),
        Runtime.getRuntime().availableProcessors() * 2,
        60L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(100),
        new ThreadFactoryBuilder().setNameFormat("cpu-pool-%d").build(),
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
    
    // IO 密集型任务线程池
    public static final ExecutorService IO_INTENSIVE = new ThreadPoolExecutor(
        Runtime.getRuntime().availableProcessors() * 2,
        Runtime.getRuntime().availableProcessors() * 4,
        60L, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(200),
        new ThreadFactoryBuilder().setNameFormat("io-pool-%d").build(),
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
    
    // 快速响应任务线程池
    public static final ExecutorService FAST_RESPONSE = new ThreadPoolExecutor(
        20,
        50,
        60L, TimeUnit.SECONDS,
        new SynchronousQueue<>(),
        new ThreadFactoryBuilder().setNameFormat("fast-pool-%d").build(),
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
}

// × 错误示例:使用默认线程池
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 使用 ForkJoinPool.commonPool(),共享线程池
    return "结果";
});

// 正确示例:使用自定义线程池
ExecutorService executor = ThreadPoolConfig.IO_INTENSIVE;
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 使用自定义线程池,隔离资源
    return "结果";
}, executor);

// × 错误认识:CompletableFuture 自己管理线程
CompletableFuture.supplyAsync(() -> {
    // 实际上仍使用线程池(默认或自定义)
    return "结果";
});

// 正确理解:CompletableFuture 需要合理配置线程池
ExecutorService executor = new ThreadPoolExecutor(...);
CompletableFuture.supplyAsync(() -> "结果", executor);

// × 错误示例:每个步骤都 join()
String result1 = future1.join();  // 阻塞
String result2 = future2.join();  // 阻塞
String result3 = future3.join();  // 阻塞
// 完全退化成同步执行

// 正确示例:链式编排,最后统一 join()
String result = CompletableFuture
    .supplyAsync(() -> serviceA.getData())
    .thenApply(data -> serviceB.process(data))
    .thenApply(data -> serviceC.save(data))
    .join();  // 只在最后阻塞一次

// × 错误示例:异常被静默处理
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("异常");
}).exceptionally(ex -> null);  // 异常被吞掉,返回 null

// 正确示例:记录异常日志
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("异常");
}).exceptionally(ex -> {
    log.error("异步任务失败", ex);  // 记录异常
    return "默认值";
});

// × 错误示例:阻塞 IO 和 CPU 任务共用线程池
ExecutorService executor = Executors.newFixedThreadPool(10);

CompletableFuture<String> ioTask = CompletableFuture.supplyAsync(() -> {
    return httpClient.get(url);  // IO 阻塞
}, executor);

CompletableFuture<Integer> cpuTask = CompletableFuture.supplyAsync(() -> {
    return calculate(data);  // CPU 密集
}, executor);

// IO 阻塞会拖慢 CPU 任务

// 正确示例:线程池隔离
CompletableFuture<String> ioTask = CompletableFuture.supplyAsync(() -> {
    return httpClient.get(url);  // IO 阻塞
}, ThreadPoolConfig.IO_INTENSIVE);

CompletableFuture<Integer> cpuTask = CompletableFuture.supplyAsync(() -> {
    return calculate(data);  // CPU 密集
}, ThreadPoolConfig.CPU_INTENSIVE);

// × 错误示例:过度使用异步,代码难以理解
CompletableFuture
    .supplyAsync(() -> step1())
    .thenApplyAsync(step2)
    .thenComposeAsync(step3)
    .thenCombineAsync(future4, combine)
    .thenAcceptAsync(step5)
    .thenRunAsync(step6);

// 正确示例:简单场景用简单方案
// 如果不需要异步,直接同步执行更清晰
String result = step1();
result = step2(result);
result = step3(result);

// Java 21：虚拟线程 + 同步代码替代 thenApplyAsync 链
var executor = Executors.newVirtualThreadPerTaskExecutor();
Future<String> f = executor.submit(() -> {
    User u = fetchUser(id);          // 阻塞但不占 OS 线程
    Order o = fetchOrder(u.id);      // 顺序执行，可读性高
    return buildResponse(u, o);
});