⚡ 异步编程与多线程实战

// 1. supplyAsync — 有返回值（最常用） CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> { return "查询库存：充足"; }); // 2. runAsync — 无返回值（像 void 方法） CompletableFuture<Void> cf2 = CompletableFuture.runAsync(() -> { System.out.println("发送邮件通知..."); }); // 3. completedFuture — 直接包装一个已有值 CompletableFuture<String> cf3 = CompletableFuture.completedFuture("已有数据"); // 4. supplyAsync + 自定义线程池（生产推荐！） ExecutorService bizPool = Executors.newFixedThreadPool(8); CompletableFuture<String> cf4 = CompletableFuture.supplyAsync(() -> { return "查询库存：充足"; }, bizPool); // ← 指定自己的线程池，不用默认的 ForkJoinPool

// 完整示例：模拟电商下单前的数据查询链 CompletableFuture<String> result = CompletableFuture // 第1步：查库存（supplyAsync = 有返回值的异步任务） .supplyAsync(() -> { sleep(200); // 模拟耗时200ms return "库存=100"; }) // 第2步：拿到上一步的结果，继续处理（thenApply = 同步转换） .thenApply(stock -> { System.out.println("收到: " + stock); // 收到: 库存=100 return stock + ", 可以下单"; }) // 第3步：最终消费结果（thenAccept = 只消费，不返回） .thenAccept(finalResult -> { System.out.println("最终结果: " + finalResult); // 输出: 最终结果: 库存=100, 可以下单 }) // 第4步：异常处理（exceptionally = 出错了怎么办） .exceptionally(ex -> { System.err.println("出错了: " + ex.getMessage()); return null; }); result.join(); // 等待整个链完成

// thenApply：同步转换（像 map） CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "库存=100") .thenApply(stock -> stock + ", 可以下单"); // 返回类型：CompletableFuture<String> // thenCompose：异步流转（像 flatMap） CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "库存=100") .thenCompose(stock -> // ← 注意这里返回的是 CompletableFuture，不是 String checkCoupon(stock) // 这个方法返回 CompletableFuture<String> ); // thenCompose 自动"展平"，不会变成 CompletableFuture<CompletableFuture<String>> // 对比：如果 thenApply 返回 CompletableFuture 会怎样？ CompletableFuture<CompletableFuture<String>> nested = // ❌ 嵌套了！ CompletableFuture.supplyAsync(() -> "库存=100") .thenApply(stock -> checkCoupon(stock)); // 变成了 CompletableFuture<CompletableFuture<String>> 嵌套！

// ① exceptionally — 出错了给个默认值（像 catch + 返回默认值） CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> { if (true) throw new RuntimeException("库存查询失败"); return "库存=100"; }).exceptionally(ex -> { System.out.println("异常: " + ex.getMessage()); return "库存查询失败，请稍后重试"; // 降级返回 }); System.out.println(cf1.join()); // 输出: 异常: java.lang.RuntimeException: 库存查询失败 // 输出: 库存查询失败，请稍后重试 // ② handle — 无论成功失败都走这里（像 finally + 可以改结果） CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "库存=100") .handle((result, ex) -> { if (ex != null) { return "出错了: " + ex.getMessage(); } return result + ", 处理完成"; }); System.out.println(cf2.join()); // 输出: 库存=100, 处理完成 // ③ whenComplete — 只记录，不改结果（像 finally + 不改返回值） CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> "库存=100") .whenComplete((result, ex) -> { if (ex != null) { System.out.println("记录日志: " + ex.getMessage()); } else { System.out.println("记录日志: 成功，结果=" + result); } // 注意：这里不能改返回值！只能做记录 }); System.out.println(cf3.join()); // 输出: 记录日志: 成功，结果=库存=100 // 输出: 库存=100

// 电商下单：3个查询并行 → 汇总 → 下单 public OrderResult createOrder(String userId, String productId) { // 3个查询同时发起（并行） CompletableFuture<String> stockFuture = CompletableFuture.supplyAsync(() -> checkStock(productId), bizPool); // 200ms CompletableFuture<String> couponFuture = CompletableFuture.supplyAsync(() -> checkCoupon(userId), bizPool); // 150ms CompletableFuture<String> addressFuture = CompletableFuture.supplyAsync(() -> checkAddress(userId), bizPool); // 100ms // 等待全部完成，汇总结果 CompletableFuture<OrderResult> orderFuture = CompletableFuture .allOf(stockFuture, couponFuture, addressFuture) .thenApply(v -> { // allOf 完成后，3个 Future 都已经有结果了 String stock = stockFuture.join(); // 不阻塞，立即返回 String coupon = couponFuture.join(); // 不阻塞，立即返回 String address = addressFuture.join(); // 不阻塞，立即返回 return new OrderResult(stock, coupon, address); }); return orderFuture.join(); // 等待最终结果 } // 输出结果： // OrderResult{stock='库存=100', coupon='满100减20', address='北京市朝阳区xx号'} // 总耗时：约200ms（取最慢的），而不是450ms（串行累加）

// orTimeout — 超时自动失败（Java 9+） CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> { sleep(5000); // 模拟慢查询5秒 return "结果"; }).orTimeout(2, TimeUnit.SECONDS); // 2秒超时 // 如果2秒内没完成 → 自动抛出 TimeoutException try { cf.join(); } catch (CompletionException e) { System.out.println("超时了: " + e.getCause()); // 输出: 超时了: java.util.concurrent.TimeoutException } // completeOnTimeout — 超时给默认值（更友好） CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> { sleep(5000); return "结果"; }).completeOnTimeout("查询超时，请稍后重试", 2, TimeUnit.SECONDS); System.out.println(cf2.join()); // 输出: 查询超时，请稍后重试

// 第1步：启动类上加 @EnableAsync @SpringBootApplication @EnableAsync public class Application { ... } // 第2步：方法上加 @Async @Service public class EmailService { @Async public void sendEmail(String to, String content) { sleep(1000); // 模拟发邮件耗时 System.out.println("邮件已发送给: " + to); } } // 第3步：调用（像普通方法一样调，但实际是异步的） @RestController public class OrderController { @Autowired private EmailService emailService; @PostMapping("/order") public String createOrder() { emailService.sendEmail("user@test.com", "下单成功"); return "下单成功"; // 立即返回，不用等邮件发完 } }

@Service public class OrderService { // ❌ 错误：同类内部调用，@Async 不生效！ public String createOrder() { sendEmail(); // ← 这是 this.sendEmail()，不走代理！同步执行！ return "下单成功"; } @Async public void sendEmail() { System.out.println("发送邮件..."); } } // ✅ 正确：把 @Async 方法放到另一个 Service 里 @Service public class EmailService { @Async public void sendEmail() { System.out.println("发送邮件..."); } } @Service public class OrderService { @Autowired private EmailService emailService; public String createOrder() { emailService.sendEmail(); // ✅ 通过代理调用，异步生效！ return "下单成功"; } }

@Service public class ReportService { @Async public void generateReport() { int i = 1 / 0; // ← 抛出 ArithmeticException System.out.println("这行不会执行"); } } // 调用方： reportService.generateReport(); System.out.println("调用完成"); // 正常打印，好像什么都没发生！ // ❌ 异常完全被吞了！没有任何日志！你都不知道出错了！ // 类比：寄快递没留电话，出了问题通知不到你

// 修复方式1：返回 CompletableFuture @Async public CompletableFuture<String> generateReport() { try { int i = 1 / 0; return CompletableFuture.completedFuture("成功"); } catch (Exception e) { return CompletableFuture.failedFuture(e); // ← 异常包装进 Future } } // 修复方式2：全局异常处理器 @Configuration public class AsyncConfig implements AsyncConfigurer { @Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return (ex, method, params) -> { System.err.println("异步任务异常! 方法=" + method.getName() + ", 参数=" + Arrays.toString(params) + ", 异常=" + ex.getMessage()); }; } } // 输出: 异步任务异常! 方法=generateReport, 参数=[], 异常=/ by zero

// @Async 不指定线程池时，默认用 SimpleAsyncTaskExecutor // 这个执行器：每来一个任务就创建一个新线程！没有上限！ // 高并发下 → 线程数爆增 → OOM // ✅ 正确：配置自定义线程池 @Configuration @EnableAsync public class AsyncConfig { @Bean("bizAsyncPool") public Executor bizAsyncPool() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(8); // 核心线程数 executor.setMaxPoolSize(20); // 最大线程数 executor.setQueueCapacity(100); // 队列容量 executor.setThreadNamePrefix("biz-async-"); // 线程名前缀（方便排查） executor.setRejectedExecutionHandler(new CallerRunsPolicy()); // 拒绝策略 executor.initialize(); return executor; } } // 使用时指定线程池 @Async("bizAsyncPool") public void sendEmail() { ... }

@Service public class OrderService { @Async @Transactional // ⚠️ 事务在异步线程中！ public void processOrder(Long orderId) { // 这个方法在异步线程中执行 // Spring 事务是基于 ThreadLocal 的 // 异步线程 ≠ 主线程 → 主线程的事务上下文丢了！ // // 情况1：异常回滚 — 只回滚异步线程的事务，主线程不知道 // 情况2：主线程已经提交了，异步线程还在跑 → 数据不一致 orderMapper.updateStatus(orderId, "PROCESSING"); } } // ✅ 正确做法：事务放在主线程，异步只做不需要事务的操作 @Transactional public void createOrder() { orderMapper.insert(order); // 主线程事务内操作 emailService.sendEmail(); // 异步发邮件（不需要事务） }

@Async public String getData() { // ❌ 返回 String，调用方拿到的永远是 null！ return "数据"; } // 调用方： String data = service.getData(); // data = null！因为实际在异步线程执行 // ✅ 正确：返回 CompletableFuture @Async public CompletableFuture<String> getData() { return CompletableFuture.completedFuture("数据"); } // 调用方： CompletableFuture<String> future = service.getData(); String data = future.join(); // ✅ 拿到 "数据"

// 看起来没问题？其实问题很大！ public class UnsafeCounter { private int count = 0; public void increment() { count++; // ← 这一行不是原子操作！ // 编译后实际上是3步： // 1. 读取 count 的值（比如 0） // 2. 加 1（得到 1） // 3. 写回 count（写入 1） // 两个线程同时读到 0 → 都加1 → 都写1 → 丢了1次！ } } // 测试：1000次 ++，结果应该=1000 UnsafeCounter counter = new UnsafeCounter(); IntStream.range(0, 1000).parallel().forEach(i -> counter.increment()); System.out.println("结果: " + counter.count); // 多次运行结果随机：973、988、991、967... // 几乎不可能得到 1000！

// 方式1：synchronized（最简单，但粒度粗） public synchronized void increment() { count++; } // 方式2：ReentrantLock（更灵活，可设超时） private final ReentrantLock lock = new ReentrantLock(); public void increment() { lock.lock(); try { count++; } finally { lock.unlock(); } } // 方式3：AtomicInteger（★推荐，CAS无锁，性能最好） private final AtomicInteger count = new AtomicInteger(0); public void increment() { count.incrementAndGet(); // ← CAS 无锁，极高并发下性能好 } // 方式4：LongAdder（超高并发下比 AtomicInteger 更快） private final LongAdder count = new LongAdder(); public void increment() { count.increment(); // ← 分段累加，超高并发下最快 }

// 转账：A给B转100，同时B给A转50 public void transfer(Account from, Account to, int amount) { synchronized (from) { // 先锁转出账户 System.out.println("锁定: " + from.name); synchronized (to) { // 再锁转入账户 from.debit(amount); to.credit(amount); } } } // 线程1：A→B → 先锁A，再锁B // 线程2：B→A → 先锁B，再锁A // // 如果时序恰好： // 线程1 锁住A → 等锁B（被线程2占着） // 线程2 锁住B → 等锁A（被线程1占着） // → 互相等待 → 死锁！

// 策略1：固定加锁顺序（最常用！） // 所有人都先锁A再锁B，就不会循环等待了 public void safeTransfer(Account from, Account to, int amount) { Account first = from.id < to.id ? from : to; // ← 按ID排序 Account second = from.id < to.id ? to : from; synchronized (first) { synchronized (second) { from.debit(amount); to.credit(amount); } } } // 策略2：超时放弃（tryLock with timeout） ReentrantLock lockA = new ReentrantLock(); ReentrantLock lockB = new ReentrantLock(); try { if (lockA.tryLock(100, TimeUnit.MILLISECONDS)) { try { if (lockB.tryLock(100, TimeUnit.MILLISECONDS)) { try { // 成功拿到两把锁，执行业务 } finally { lockB.unlock(); } } } finally { lockA.unlock(); } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } // 策略3-5：减小锁粒度 / 使用无锁设计(Atomic/CAS) / 死锁检测工具(Arthas)

// ❌ 没有 volatile：子线程可能永远看不到 flag 的修改 public class VisibilityDemo { private static boolean running = true; // ❌ 没有 volatile public static void main(String[] args) throws Exception { new Thread(() -> { int count = 0; while (running) { // ← 子线程可能把 running 缓存在自己的 CPU 核心里 count++; // 主线程改成 false，但子线程看不到！ } System.out.println("停止了，count=" + count); }).start(); Thread.sleep(1000); running = false; // 主线程改为 false // 但子线程可能永远不停！因为 running 的修改对子线程"不可见" } } // ✅ 加 volatile：保证所有线程都能看到最新值 private static volatile boolean running = true; // ✅ 加了 volatile

// 双重检查锁单例 — 为什么需要 volatile？ public class Singleton { private static volatile Singleton instance; // ← 必须加 volatile！ public static Singleton getInstance() { if (instance == null) { // 第1次检查 synchronized (Singleton.class) { if (instance == null) { // 第2次检查 instance = new Singleton(); // ← 这行可能被重排！ } } } return instance; } } // new Singleton() 实际分3步： // 1. 分配内存空间 // 2. 调用构造方法初始化 // 3. 将引用指向内存地址 // // 重排后可能变成 1→3→2： // 线程A 执行了1→3（引用非null了，但对象还没初始化） // 线程B 第1次检查 instance!=null → 直接返回！ // 但此时对象还没初始化完 → 用了半成品 → NPE！ // // volatile 禁止重排 → 保证 1→2→3 顺序执行

// ❌ 异常被静默吞掉 CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> { throw new RuntimeException("数据库连接失败！"); }); // 没有任何日志输出！异常就这么消失了！ // 你都不知道出错了！ // ✅ 修复：加上 exceptionally 或 handle CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> { throw new RuntimeException("数据库连接失败！"); }).exceptionally(ex -> { log.error("异步任务失败", ex); // ← 至少打印日志！ return "默认值"; // 降级返回 });

ExecutorService pool = Executors.newFixedThreadPool(4); Future<String> future = pool.submit(() -> { throw new RuntimeException("出错了！"); }); // ❌ 没有任何输出！异常被 Future 对象"吞"了 // 异常藏在 future 里面，只有你调用 future.get() 时才会抛出 // 如果你从来不调 future.get() → 异常就永远消失了 // 这在生产环境很常见：异步任务出错了，但没人看 Future

// ❌ 直接 new Thread，异常直接消失 new Thread(() -> { throw new RuntimeException("线程异常！"); }).start(); // 控制台可能有输出（JVM默认打印到stderr），但你的监控系统收不到 // ✅ 修复：设置 UncaughtExceptionHandler Thread t = new Thread(() -> { throw new RuntimeException("线程异常！"); }); t.setUncaughtExceptionHandler((thread, ex) -> { log.error("线程[{}]异常", thread.getName(), ex); // 发送告警、记录到监控系统等 }); t.start();

// 1. CompletableFuture：永远链式调用 exceptionally 或 handle cf.exceptionally(ex -> { log.error(...); return defaultVal; }); // 2. @Async：配置 AsyncUncaughtExceptionHandler（见P4） // 3. 线程池：设置 ThreadFactory + UncaughtExceptionHandler ThreadPoolExecutor pool = new ThreadPoolExecutor(...); pool.setThreadFactory(r -> { Thread t = new Thread(r); t.setUncaughtExceptionHandler((thread, ex) -> { log.error("线程[{}]未捕获异常", thread.getName(), ex); }); return t; }); // 4. 全局兜底：CompletableFuture 框架方法 // 在每个 CompletableFuture 链的末尾，加上 whenComplete 做日志 cf.whenComplete((result, ex) -> { if (ex != null) log.error("异步任务最终异常", ex); });

// ❌ 不指定线程池，所有 CompletableFuture 共享 ForkJoinPool.commonPool() // commonPool 的线程数 = CPU核数-1（比如8核机器只有7个线程） // 假设有两个业务都在用 CompletableFuture： // 业务A：导出报表（很慢，每个任务5秒） CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> { sleep(5000); // 慢任务，占住线程5秒 return "报表生成完成"; }); // ← 没传线程池，用的是 commonPool！ // 业务B：查库存（本来很快，50ms） CompletableFuture<String> fast = CompletableFuture.supplyAsync(() -> { return "库存=100"; }); // ← 也没传线程池，也在用 commonPool！ // 如果业务A同时跑了7个慢任务 → commonPool 全被占满！ // 业务B的快任务只能在队列里等 → 50ms的查询等了5秒！ // 看起来像是B很慢，实际是被A拖累的！ // ✅ 修复：每个业务用自己的线程池 ExecutorService reportPool = Executors.newFixedThreadPool(4); // 报表专用 ExecutorService bizPool = Executors.newFixedThreadPool(8); // 业务查询专用 CompletableFuture.supplyAsync(slowTask, reportPool); // ← 报表用自己的池子 CompletableFuture.supplyAsync(fastTask, bizPool); // ← 查询用自己的池子

// 线程池配置：核心8，最大20，队列100 ThreadPoolExecutor pool = new ThreadPoolExecutor( 8, 20, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy() // ← 拒绝策略 ); // CallerRunsPolicy 的行为：线程池满了 → 让提交任务的线程自己执行 // 听起来不错？但问题是： // // 如果是 HTTP 请求线程提交的异步任务 → // 线程池满了 → HTTP线程自己去跑任务（比如5秒的导出报表）→ // HTTP线程被阻塞5秒！→ 接口超时！→ 用户体验极差！ // // 更好的做法：用有界队列 + 自定义拒绝策略（记录日志+返回降级结果） // ✅ 推荐：自定义拒绝策略 new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { log.warn("线程池已满，任务被拒绝！队列大小={}, 活跃线程数={}", executor.getQueue().size(), executor.getActiveCount()); // 可以：返回降级结果 / 放入MQ延迟处理 / 发告警 } }

// ❌ SecurityContext 丢失 @GetMapping("/order") public String createOrder() { // 主线程中：SecurityContextHolder 能拿到当前用户 String currentUser = SecurityContextHolder.getContext() .getAuthentication().getName(); // ✅ 主线程能拿到 "张三" CompletableFuture.supplyAsync(() -> { // 异步线程中：SecurityContextHolder 是空的！ String user = SecurityContextHolder.getContext() .getAuthentication().getName(); // ❌ NPE！ return orderService.createOrder(user); }); return "ok"; } // ✅ 修复方式1：手动传递上下文 String currentUser = SecurityContextHolder.getContext() .getAuthentication().getName(); // 在主线程先取出来 CompletableFuture.supplyAsync(() -> { return orderService.createOrder(currentUser); // ← 直接传参 }); // ✅ 修复方式2：用 DelegatingSecurityContextExecutorService ExecutorService securedPool = new DelegatingSecurityContextExecutorService( Executors.newFixedThreadPool(8) ); // ← 自动把 SecurityContext 传到异步线程

// ❌ 主线程设了 traceId，但异步线程里没有 @GetMapping("/order") public String createOrder() { MDC.put("traceId", UUID.randomUUID().toString()); CompletableFuture.supplyAsync(() -> { log.info("处理订单"); // 日志输出：[traceId=] 处理订单 ← traceId 是空的！ // 在ELK里搜索不到这条日志！ return "ok"; }); return "ok"; } // ✅ 修复：手动传递 MDC Map<String, String> contextMap = MDC.getCopyOfContextMap(); // 在主线程复制 CompletableFuture.supplyAsync(() -> { if (contextMap != null) { MDC.setContextMap(contextMap); // ← 在异步线程恢复 } try { log.info("处理订单"); // [traceId=abc123] 处理订单 ✅ return "ok"; } finally { MDC.clear(); // ← 清理，防止线程复用时的上下文污染 } });

// ❌ 改造前：3个查询串行执行，总耗时 450ms @PostMapping("/order") public OrderVO createOrder(CreateOrderReq req) { String stock = stockService.check(req.getProductId()); // 200ms String coupon = couponService.check(req.getUserId()); // 150ms String address = addressService.get(req.getUserId()); // 100ms // 总耗时：200+150+100 = 450ms return buildOrderVO(stock, coupon, address); } // ✅ 改造后：3个查询并行，总耗时 200ms @PostMapping("/order") public OrderVO createOrder(CreateOrderReq req) { CompletableFuture<String> stockFuture = CompletableFuture .supplyAsync(() -> stockService.check(req.getProductId()), bizPool); CompletableFuture<String> couponFuture = CompletableFuture .supplyAsync(() -> couponService.check(req.getUserId()), bizPool); CompletableFuture<String> addressFuture = CompletableFuture .supplyAsync(() -> addressService.get(req.getUserId()), bizPool); // 等待全部完成 CompletableFuture.allOf(stockFuture, couponFuture, addressFuture).join(); return buildOrderVO( stockFuture.join(), couponFuture.join(), addressFuture.join() ); // 总耗时：max(200, 150, 100) = 200ms，提速 55%！ }

// ❌ 改造前：同步导出，大数据量直接超时 @GetMapping("/report/export") public void exportReport(HttpServletResponse response) { List<ReportData> data = reportService.queryAll(); // 查10万条，5秒 byte[] excel = excelService.generate(data); // 生成Excel，3秒 // 总耗时8秒，HTTP请求很可能超时！ response.getOutputStream().write(excel); } // ✅ 改造后：异步生成+轮询下载 // 第1步：提交导出任务，立即返回任务ID @PostMapping("/report/submit") public TaskVO submitExport() { String taskId = UUID.randomUUID().toString(); // 异步生成报表 CompletableFuture.runAsync(() -> { List<ReportData> data = reportService.queryAll(); byte[] excel = excelService.generate(data); cacheService.put(taskId, excel, 30, TimeUnit.MINUTES); // 缓存30分钟 taskService.updateStatus(taskId, "DONE"); }, reportPool).exceptionally(ex -> { taskService.updateStatus(taskId, "FAILED"); log.error("报表生成失败", ex); return null; }); return new TaskVO(taskId, "PROCESSING"); } // 第2步：前端轮询任务状态 @GetMapping("/report/status/{taskId}") public TaskVO getTaskStatus(String taskId) { String status = taskService.getStatus(taskId); return new TaskVO(taskId, status); } // 第3步：下载 @GetMapping("/report/download/{taskId}") public void download(String taskId, HttpServletResponse response) { byte[] excel = cacheService.get(taskId); response.getOutputStream().write(excel); }

// ❌ 改造前：同步逐条发邮件 public void sendNotification(List<String> users) { for (String user : users) { // 1000个用户 emailService.send(user, content); // 每封200ms } // 总耗时：1000 × 200ms = 200秒！完全不可接受 } // ✅ 改造后：CompletableFuture 并行 + 批次控制 public void sendNotification(List<String> users) { // 分批：每批50个用户 List<List<String>> batches = Lists.partition(users, 50); for (List<String> batch : batches) { List<CompletableFuture<Void>> futures = batch.stream() .map(user -> CompletableFuture.runAsync( () -> emailService.send(user, content), emailPool )) .collect(Collectors.toList()); // 等待当前批次全部完成，再处理下一批 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .join(); } // 总耗时：20批 × 200ms = 4秒（从200秒→4秒，提速50倍！） }

// ❌ 改造前：5个统计任务串行执行 @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点 public void dailyStats() { statsService.calcOrderStats(); // 30秒 statsService.calcUserStats(); // 20秒 statsService.calcProductStats(); // 15秒 statsService.calcRevenueStats(); // 25秒 statsService.calcTrafficStats(); // 10秒 // 总耗时：100秒 } // ✅ 改造后：5个任务并行执行 @Scheduled(cron = "0 0 2 * * ?") public void dailyStats() { CompletableFuture<?>[] tasks = new CompletableFuture[]{ CompletableFuture.runAsync(() -> statsService.calcOrderStats(), statsPool), CompletableFuture.runAsync(() -> statsService.calcUserStats(), statsPool), CompletableFuture.runAsync(() -> statsService.calcProductStats(), statsPool), CompletableFuture.runAsync(() -> statsService.calcRevenueStats(), statsPool), CompletableFuture.runAsync(() -> statsService.calcTrafficStats(), statsPool), }; CompletableFuture.allOf(tasks).join(); // 总耗时：max(30,20,15,25,10) = 30秒（从100秒→30秒，提速70%） // 别忘了异常处理！ for (CompletableFuture<?> task : tasks) { try { task.join(); } catch (CompletionException e) { log.error("统计任务失败", e.getCause()); } } }