尧图网页设计 
从零开始用启发式教学带你彻底搞懂非阻塞延迟这个利器。每个 API 先讲清功能再上代码注释逐行拆解。第一章我们遇到了什么问题1.1 一个真实场景假设你在写一个订单服务。调用第三方支付接口时偶尔会失败。你的第一反应是什么失败了就重试啊。好那重试之间要不要等一等毕竟对方服务可能正在抖动立刻重试大概率还是失败。等个 1 秒再试。很合理。那你会怎么写public String callPayment() { for (int i 0; i 3; i) { try { return doCallPayment(); // 调用支付接口 } catch (Exception e) { Thread.sleep(1000); // 等 1 秒再重试 } } throw new RuntimeException(支付调用失败); }这段代码有什么问题1.2 问题出在哪Thread.sleep(1000)做了什么它让当前线程冻结 1 秒。思考如果这段代码运行在线程池里——比如一个 Web 服务器处理请求的线程池——会发生什么线程池共 4 个线程 T1: [处理请求A] → 支付失败 → sleep(1s) → 重试 → 成功 T2: [处理请求B] → 支付失败 → sleep(1s) → 重试 → 成功 T3: [处理请求C] → 支付失败 → sleep(1s) → 重试 → 成功 T4: [处理请求D] → 支付失败 → sleep(1s) → 重试 → 成功 问题4 个线程全在 sleep新的请求 E 来了没人处理核心矛盾线程在sleep期间什么都没干但占着位置不放。就像 4 个厨师都在等锅热但灶台被他们占着别的菜做不了。1.3 我们想要什么我们想要的其实是等 1 秒然后继续做——但等待期间线程应该被释放回去干别的活。理想情况 T1: [处理请求A] → 支付失败 → 释放线程→ (1秒后回来) → 重试 → 成功 → 释放 T2: [处理请求B] → 支付失败 → 释放线程→ (1秒后回来) → 重试 → 成功 → 释放 ↑ 等待期间 T1、T2 可以去处理请求 E、F、G...怎么实现释放线程1 秒后自动回来这就是delayedExecutor要解决的问题。第二章认识 delayedExecutor2.1 它是什么CompletableFuture.delayedExecutor是 JDK 9 引入的一个静态方法。public static Executor delayedExecutor(long delay, TimeUnit unit)一句话它返回一个特殊的Executor——你往这个 Executor 提交的任务不会立即执行而是等指定时间后才执行。2.2 先搞懂 Executor 是什么在 Java 中Executor是一个接口只有一个方法public interface Executor { void execute(Runnable command); // 执行一个任务 }常见的 Executor 实现实现类行为ThreadPoolExecutor提交后立即交给线程池执行ForkJoinPool.commonPool()提交后立即交给公共池执行delayedExecutor(1, SECONDS)提交后等 1 秒再执行关键理解delayedExecutor返回的也是一个Executor只是它的execute方法多了一个延迟行为。2.3 它内部怎么工作你的代码 JDK 内部 ───────── ────────── ┌──────────────────────────────┐ │ Delayer守护线程只有 1 个 │ │ 职责专门负责计时 │ └──────────────────────────────┘ │ delayedExecutor(2, SECONDS) │ .execute(myTask) ──────────▶ 告诉 Delayer2 秒后执行 myTask │ 你的线程池继续干别的事 Delayer 开始计时... 2 秒到了 → 把 myTask 丢到线程池执行要点等待期间你的线程池完全不受影响——线程该干嘛干嘛计时由 JDK 内部唯一的 1 个守护线程负责——它不是你的线程池里的线程零额外线程占用指的是不占你的用户线程池2.4 它解决什么问题不适合解决什么问题适合的场景场景说明非阻塞重试退避失败后等 N 秒重试等待期间不占线程延迟任务调度5 秒后发一条通知、10 秒后检查状态限流/冷却两次操作之间强制间隔超时补偿超时后延迟执行清理逻辑不适合的场景场景为什么不适合更好的选择精确定时任务每天 8 点执行delayedExecutor只做相对延迟不做绝对时间ScheduledExecutorService.scheduleAtFixedRate周期性重复任务它只延迟一次不自动重复ScheduledExecutorService需要取消/修改延迟一旦提交无法取消ScheduledFuture第三章从最简单的例子开始3.1 Hello delayedExecutor先写一个最简单的程序感受延迟执行的效果。import java.util.concurrent.*; /** * 最基础的 delayedExecutor 使用演示。 * * 目标提交一个任务让它 2 秒后才执行。 */ public class HelloDelayedExecutor { public static void main(String[] args) throws Exception { // 记录当前时间用于计算耗时 long start System.currentTimeMillis(); System.out.println([ elapsed(start) ms] 准备提交延迟任务); // ─── 核心代码创建延迟执行器 ─── // delayedExecutor(2, SECONDS) 返回一个 Executor // 这个 Executor 的特点是提交的任务会延迟 2 秒才真正执行 Executor delayed CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS); // ─── 提交任务 ─── // runAsync 把一个 Runnable 任务提交到指定的 Executor 上执行 // 因为我们的 Executor 是 delayedExecutor所以任务会延迟 2 秒 CompletableFuture.runAsync(() - { // 这个 lambda 就是我们要执行的任务 // 它会在 2 秒后才真正运行 System.out.println([ elapsed(start) ms] 任务执行了 线程名 Thread.currentThread().getName()); }, delayed); // ↑ 注意第二个参数指定在哪个 Executor 上执行 // 这里传入的是 delayedExecutor所以会延迟 2 秒 // ─── 验证主线程没有被阻塞 ─── // runAsync 是异步的——它立即返回不会等待任务完成 // 所以这行代码会立刻执行不会等 2 秒 System.out.println([ elapsed(start) ms] 主线程继续执行没有被阻塞); // 等待程序退出否则主线程结束JVM 退出延迟任务来不及执行 Thread.sleep(4000); System.out.println([ elapsed(start) ms] 程序结束); } /** * 计算从 start 到现在经过了多少毫秒 */ static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 准备提交延迟任务 [0ms] 主线程继续执行没有被阻塞 ← 立即执行没有等 2 秒 [2001ms] 任务执行了 线程名ForkJoinPool.commonPool-worker-1 [4001ms] 程序结束观察与思考任务在 2 秒后才执行——延迟生效主线程没有等待——runAsync是异步的执行线程是ForkJoinPool.commonPool-worker-1——不是主线程3.2 理解执行线程问题任务在哪个线程上执行delayedExecutor内部有两步计时由 JDK 内部的 Delayer 守护线程负责执行计时到期后任务被丢到ForkJoinPool.commonPool()执行所以任务实际运行在commonPool的线程上而不是 Delayer 线程上。如果你想让任务在你自己的线程池上执行呢import java.util.concurrent.*; /** * 演示让延迟任务在自定义线程池上执行。 */ public class DelayedWithCustomPool { public static void main(String[] args) throws Exception { // 创建我们自己的线程池方便观察线程名 ExecutorService myPool Executors.newFixedThreadPool(2, r - { Thread t new Thread(r); t.setName(my-pool- t.getId()); // 自定义线程名 return t; }); long start System.currentTimeMillis(); // 方案先用 delayedExecutor 延迟再用 supplyAsync 指定线程池 // // 执行流程 // 1. delayedExecutor 等 1 秒 // 2. 1 秒后把任务丢到 myPool 执行 CompletableFuture.supplyAsync(() - { System.out.println([ elapsed(start) ms] 任务执行线程 Thread.currentThread().getName()); return 结果; }, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS, myPool)); // ↑ 第三个参数指定执行线程池 // 延迟到期后任务在这个线程池上执行 Thread.sleep(3000); myPool.shutdown(); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[1001ms] 任务执行线程my-pool-14关键发现delayedExecutor有一个三参数重载版本// 两参数版延迟后在 commonPool 执行 delayedExecutor(1, TimeUnit.SECONDS) // 三参数版延迟后在指定 Executor 执行 delayedExecutor(1, TimeUnit.SECONDS, myPool)第四章核心场景——非阻塞重试4.1 先用阻塞方式实现重试对比基线import java.util.concurrent.*; import java.util.concurrent.atomic.*; /** * 阻塞式重试Thread.sleep 方式。 * * 问题sleep 期间线程被冻结无法服务其他任务。 */ public class BlockingRetry { // 模拟计数器前 2 次调用失败第 3 次成功 static final AtomicInteger counter new AtomicInteger(0); public static void main(String[] args) throws Exception { // 只有 1 个线程的线程池——问题更明显 ExecutorService pool Executors.newFixedThreadPool(1); long start System.currentTimeMillis(); // 提交 2 个任务 pool.submit(() - retryTask(任务A, start)); pool.submit(() - retryTask(任务B, start)); // ↑ 任务B 必须等任务A 完全结束包括 sleep才能开始 pool.shutdown(); pool.awaitTermination(30, TimeUnit.SECONDS); } static void retryTask(String name, long start) { for (int attempt 1; attempt 3; attempt) { counter.incrementAndGet(); System.out.println([ elapsed(start) ms] name 第 attempt 次尝试线程 Thread.currentThread().getName() ); // 模拟前 4 次调用失败第 5 次成功 if (counter.get() 5) { System.out.println( → 失败等待 1 秒后重试...); try { // ★ 问题所在线程在 sleep 期间完全冻结 ★ Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } else { System.out.println( → 成功); return; } } } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 任务A 第1次尝试线程pool-1-thread-1 → 失败等待 1 秒后重试... ↑ 线程在 sleep任务B 进不来 [1001ms] 任务A 第2次尝试线程pool-1-thread-1 → 失败等待 1 秒后重试... [2001ms] 任务A 第3次尝试线程pool-1-thread-1 → 成功 ↑ 任务A 终于结束任务B 可以开始了 [3001ms] 任务B 第1次尝试线程pool-1-thread-1 → 失败等待 1 秒后重试... [4001ms] 任务B 第2次尝试线程pool-1-thread-1 → 成功问题任务 B 等了 3 秒才开始因为线程池只有 1 个线程被任务 A 的sleep占满了。4.2 用 delayedExecutor 实现非阻塞重试现在用delayedExecutor来解决这个问题。核心思路失败 → 不 sleep → 线程立即释放 → 用 delayedExecutor 设一个1秒后的闹钟 → 闹钟响了 → 再提交一次任务到线程池import java.util.concurrent.*; import java.util.concurrent.atomic.*; /** * 非阻塞重试delayedExecutor 方式。 * * 对比上一节同样的场景1 线程池 2 个任务 重试 * 观察任务 B 是否还需要等待。 */ public class NonBlockingRetry { static final AtomicInteger counter new AtomicInteger(0); public static void main(String[] args) throws Exception { ExecutorService pool Executors.newFixedThreadPool(1); long start System.currentTimeMillis(); // 提交 2 个任务——使用非阻塞重试 CompletableFuture.runAsync(() - nonBlockingRetry(任务A, pool, start), pool); CompletableFuture.runAsync(() - nonBlockingRetry(任务B, pool, start), pool); Thread.sleep(10000); pool.shutdown(); } /** * 非阻塞重试方法。 * * param name 任务名 * param pool 线程池重试时要把任务提交回同一个池 * param start 起始时间用于打印耗时 */ static void nonBlockingRetry(String name, ExecutorService pool, long start) { int count counter.incrementAndGet(); System.out.println([ elapsed(start) ms] name 尝试线程 Thread.currentThread().getName() ); if (count 5) { // 失败但不 sleep——线程立即释放 System.out.println( → 失败1 秒后重试线程已释放...); // ★ 核心代码 ★ // // runAsync(() - {}, ...) 中的 () - {} 是空操作。 // 我们不需要它做事——我们只需要 delayedExecutor 的延迟效果。 // // 整个表达式的含义 // 等 1 秒后执行 whenComplete 里的代码 // // 等 1 秒期间 // - 当前线程已经释放可以去处理其他任务 // - JDK 内部的 Delayer 守护线程在计时 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .whenComplete((v, ex) - { // 1 秒后到达这里——重新尝试 // 把任务提交回线程池 pool.submit(() - nonBlockingRetry(name, pool, start)); }); } else { System.out.println( → 成功); } } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 任务A 尝试线程pool-1-thread-1 → 失败1 秒后重试线程已释放... ↑ 线程释放了任务B 可以进来了 [0ms] 任务B 尝试线程pool-1-thread-1 → 失败1 秒后重试线程已释放... ↑ 两个任务的闹钟同时在计时 [1001ms] 任务A 尝试线程pool-1-thread-1 → 失败1 秒后重试线程已释放... [2001ms] 任务A 尝试线程pool-1-thread-1 → 成功 [2001ms] 任务B 尝试线程pool-1-thread-1 → 成功对比阻塞版 任务A 开始0ms 任务B 开始3001ms ← 等了 3 秒 全部完成5001ms 非阻塞版 任务A 开始0ms 任务B 开始0ms ← 立即开始 全部完成3001ms ← 快了将近 2 秒4.3 更优雅的实现Promise 递归模式上面的实现有个问题每次重试都重新调用nonBlockingRetry方法逻辑分散。更好的方式是用Promise 模式——一个方法返回 Future内部递归链接重试。先理解 Promise 模式import java.util.concurrent.*; /** * Promise 模式入门。 * * 核心思想创建一个 CompletableFuture承诺 * 稍后在某个时机完成它complete 或 completeExceptionally。 * 调用方拿到这个 Future用 whenComplete/get 等待结果。 */ public class PromiseDemo { public static void main(String[] args) throws Exception { long start System.currentTimeMillis(); // 创建一个承诺——稍后会被完成 CompletableFutureString promise new CompletableFuture(); // 2 秒后完成这个承诺 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS)) .whenComplete((v, ex) - { // 2 秒后到达这里 System.out.println([ elapsed(start) ms] 延迟到期完成 promise); promise.complete(Hello from delayedExecutor!); // ↑ 完成 promise调用方拿到结果 }); // 调用方等待 promise 完成 // get() 会阻塞当前线程直到 promise 被 complete String result promise.get(); System.out.println([ elapsed(start) ms] 拿到结果 result); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[2001ms] 延迟到期完成 promise [2001ms] 拿到结果Hello from delayedExecutor!现在用 Promise 模式重写非阻塞重试import java.util.concurrent.*; import java.util.concurrent.atomic.*; import java.util.function.*; /** * 用 Promise delayedExecutor 实现优雅的非阻塞重试。 * * 这个 retryAsync 方法可以直接作为工具类使用。 */ public class RetryWithPromise { static final AtomicInteger counter new AtomicInteger(0); public static void main(String[] args) throws Exception { ExecutorService pool Executors.newFixedThreadPool(2); long start System.currentTimeMillis(); // 使用 retryAsync 工具方法 CompletableFutureString result retryAsync( () - { // 要重试的操作 int c counter.incrementAndGet(); System.out.println([ elapsed(start) ms] 尝试 # c 线程 Thread.currentThread().getName() ); if (c 3) { throw new RuntimeException(失败); // 模拟失败 } return 第 c 次成功; }, 3, // 最大重试次数 500, // 退避延迟毫秒 pool // 线程池 ); // 等待最终结果 System.out.println(最终结果 result.get()); pool.shutdown(); } /** * 非阻塞重试工具方法。 * * param action 要执行的操作可能抛异常 * param maxRetries 最大重试次数 * param backoffMs 退避延迟毫秒 * param executor 线程池 * return 最终结果的 Future */ static T CompletableFutureT retryAsync( CallableT action, int maxRetries, long backoffMs, ExecutorService executor) { // 创建承诺——调用方持有这个 Future 等待最终结果 CompletableFutureT promise new CompletableFuture(); // 启动第一次尝试然后递归链接重试 attemptAndRetry(action, maxRetries, backoffMs, executor, promise, 0); return promise; } /** * 执行一次尝试失败时用 delayedExecutor 延迟后递归重试。 * * param action 要执行的操作 * param maxRetries 最大重试次数 * param backoffMs 退避延迟 * param executor 线程池 * param promise 最终承诺 * param attempt 当前是第几次尝试从 0 开始 */ static T void attemptAndRetry( CallableT action, int maxRetries, long backoffMs, ExecutorService executor, CompletableFutureT promise, int attempt) { // supplyAsync把任务提交到线程池异步执行 // 返回一个 Future任务完成时 Future 自动完成异常时异常完成 CompletableFuture.supplyAsync(() - { try { return action.call(); // 执行用户操作 } catch (Exception e) { // 把受检异常包装为 CompletionException让 supplyAsync 的 Future 异常完成 throw new CompletionException(e); } }, executor) // whenComplete无论成功还是异常回调都执行 // 参数result 成功时的值异常时为 nullex 异常成功时为 null .whenComplete((result, ex) - { if (ex null) { // ─── 成功路径 ─── // 任务成功了完成 promise调用方拿到结果 promise.complete(result); return; } // ─── 异常路径 ─── if (attempt maxRetries) { // 重试次数耗尽传播异常 promise.completeExceptionally(ex); return; } // 还有重试机会——用 delayedExecutor 延迟后重试 System.out.println( 第 (attempt 1) 次重试延迟 backoffMs ms); // ★ 核心delayedExecutor 延迟 → whenComplete 触发 → 递归重试 ★ CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(backoffMs, TimeUnit.MILLISECONDS)) .whenComplete((v, delayEx) - { // 延迟到期递归调用自身进行下一次尝试 attemptAndRetry(action, maxRetries, backoffMs, executor, promise, attempt 1); }); }); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 尝试 #1线程pool-1-thread-1 第 1 次重试延迟 500ms [501ms] 尝试 #2线程pool-1-thread-1 第 2 次重试延迟 500ms [1001ms] 尝试 #3线程pool-1-thread-1 最终结果第 3 次成功执行流程图supplyAsync(尝试#1) → 失败 ↓ whenComplete: 失败还有重试机会 ↓ runAsync(() - {}, delayedExecutor(500ms)) ← 设置 500ms 闹钟 ↓ 线程释放 ... 500ms 后 ... ↓ Delayer: 叮 whenComplete: 延迟到期 ↓ supplyAsync(尝试#2) → 失败 ↓ whenComplete: 失败还有重试机会 ↓ runAsync(() - {}, delayedExecutor(500ms)) ← 再设一个闹钟 ↓ 线程释放 ... 500ms 后 ... ↓ Delayer: 叮 supplyAsync(尝试#3) → 成功 ↓ promise.complete(第 3 次成功)第五章进阶用法5.1 指数退避实际生产中重试延迟通常逐次增加指数退避避免对故障服务造成压力。import java.util.concurrent.*; /** * 指数退避重试每次重试延迟翻倍。 * * 第 1 次重试100ms * 第 2 次重试200ms * 第 3 次重试400ms * 第 4 次重试800ms * ... */ public class ExponentialBackoff { public static void main(String[] args) throws Exception { ExecutorService pool Executors.newFixedThreadPool(2); long start System.currentTimeMillis(); retryWithExponentialBackoff( () - { int c Counter.next(); System.out.println([ elapsed(start) ms] 尝试 # c); if (c 4) throw new RuntimeException(失败); return 成功; }, 5, // 最大重试 5 次 100, // 初始延迟 100ms pool ).get(); pool.shutdown(); } static T CompletableFutureT retryWithExponentialBackoff( CallableT action, int maxRetries, long initialDelayMs, ExecutorService executor) { return doRetry(action, maxRetries, initialDelayMs, executor, new CompletableFuture(), 0); } static T CompletableFutureT doRetry( CallableT action, int maxRetries, long currentDelayMs, ExecutorService executor, CompletableFutureT promise, int attempt) { CompletableFuture.supplyAsync(() - { try { return action.call(); } catch (Exception e) { throw new CompletionException(e); } }, executor).whenComplete((result, ex) - { if (ex null) { promise.complete(result); return; } if (attempt maxRetries) { promise.completeExceptionally(ex); return; } // ★ 指数退避当前延迟 初始延迟 × 2^attempt ★ // attempt0: 100 × 2^0 100ms // attempt1: 100 × 2^1 200ms // attempt2: 100 × 2^2 400ms long delay currentDelayMs; System.out.println( 重试 # (attempt 1) 延迟 delay ms); CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS)) .whenComplete((v, e) - // ★ 下次延迟翻倍 ★ doRetry(action, maxRetries, currentDelayMs * 2, executor, promise, attempt 1)); }); return promise; } // 简单的全局计数器 static class Counter { static int n 0; static int next() { return n; } } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 尝试 #1 重试 #1延迟 100ms [101ms] 尝试 #2 重试 #2延迟 200ms [302ms] 尝试 #3 重试 #3延迟 400ms [703ms] 尝试 #4 最终结果成功5.2 与其他 CompletableFuture 操作组合delayedExecutor可以和其他 CF 操作自由组合import java.util.concurrent.*; /** * delayedExecutor 与其他 CF 操作的组合用法。 */ public class ComposingDelayed { public static void main(String[] args) throws Exception { long start System.currentTimeMillis(); // ─── 场景 1延迟后转换结果 ─── // 等 1 秒然后计算 42 的 2 倍 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenApply(v - 42 * 2) // 延迟后转换 .whenComplete((result, ex) - System.out.println([ elapsed(start) ms] 场景1结果 result)); // 输出[1001ms] 场景1结果84 // ─── 场景 2延迟后执行异步操作 ─── // 等 1 秒然后发起一个异步查询 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenCompose(v - CompletableFuture.supplyAsync(() - 查询结果)) .whenComplete((result, ex) - System.out.println([ elapsed(start) ms] 场景2结果 result)); // 输出[1001ms] 场景2结果查询结果 // ─── 场景 3多个延迟任务并行 ─── // 1 秒后做 A2 秒后做 B等两个都完成 CompletableFutureString taskA CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenApply(v - A完成); CompletableFutureString taskB CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS)) .thenApply(v - B完成); CompletableFuture.allOf(taskA, taskB) .thenApply(v - taskA.join() taskB.join()) .whenComplete((result, ex) - System.out.println([ elapsed(start) ms] 场景3结果 result)); // 输出[2001ms] 场景3结果A完成 B完成 Thread.sleep(4000); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }第六章常见陷阱陷阱 1忘记把任务提交到 delayedExecutor// ❌ 错误只是创建了 delayedExecutor什么都没做 CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS); // 这行代码没有任何效果 // ✅ 正确必须配合 runAsync / supplyAsync 使用 CompletableFuture.runAsync(() - doSomething(), CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));陷阱 2用 supplyAsync 但 lambda 没有返回值// ❌ 编译错误supplyAsync 要求 Supplier 有返回值 CompletableFuture.supplyAsync(() - { System.out.println(执行); // 没有 return编译报错 }, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)); // ✅ 正确无返回值用 runAsync CompletableFuture.runAsync(() - { System.out.println(执行); }, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));陷阱 3以为 delayedExecutor 会中断已有任务// ❌ 误解以为 1 秒后会中断正在运行的 supplyAsync 任务 CompletableFuture.supplyAsync(() - { Thread.sleep(10000); // 任务要跑 10 秒 return done; }, pool); // delayedExecutor 跟上面这个任务没有任何关系 // 它只负责延迟后执行新提交的任务 // ✅ 正确理解delayedExecutor 只是一个定时器 // 它不取消、不中断、不影响任何已经在执行的任务陷阱 4在延迟回调里阻塞等待// ❌ 错误延迟后又去 get() 阻塞——白用了非阻塞延迟 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenApply(v - someOtherFuture.get()); // 阻塞 // ✅ 正确继续用异步操作 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenCompose(v - someOtherFuture); // 非阻塞陷阱 5忽略返回的 Future// ❌ 问题提交了延迟任务但没有持有 Future 引用 // 如果主线程退出延迟任务可能来不及执行 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS)); // 主线程立即结束 → JVM 退出 → 延迟任务没机会执行 // ✅ 正确持有 Future 引用确保程序不会提前退出 CompletableFuture? delayed CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS)); delayed.get(); // 等待延迟任务完成 // 或者delayed.join();总结一句话记忆delayedExecutor(delay, unit) 一个定时闹钟告诉它X 秒后做某事然后你就可以走开干别的了。等待期间你的线程池零占用。使用模板// 基本用法延迟后执行操作 CompletableFuture.runAsync(() - { // 这里是要做的事 }, CompletableFuture.delayedExecutor(延迟时间, 时间单位)); // 指定线程池延迟后在自定义线程池上执行 CompletableFuture.supplyAsync(() - { return 结果; }, CompletableFuture.delayedExecutor(延迟时间, 时间单位, 自定义线程池)); // 链式操作延迟后继续做其他事 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenCompose(v - 下一个异步操作) .whenComplete((result, ex) - 处理结果);适用场景速查场景用法非阻塞重试退避失败 →delayedExecutor→ 重新提交任务延迟通知5 秒后发消息 →runAsync(() - send(), delayedExecutor(5, SECONDS))延迟健康检查启动 30 秒后检查 →supplyAsync(() - check(), delayedExecutor(30, SECONDS))操作冷却间隔两次调用间强制间隔 → 链式delayedExecutor超时后清理orTimeout异常 →whenComplete→delayedExecutor→ 延迟清理 从零开始，带你彻底搞懂“非阻塞延迟“这个利器
从零开始用启发式教学带你彻底搞懂非阻塞延迟这个利器。每个 API 先讲清功能再上代码注释逐行拆解。第一章我们遇到了什么问题1.1 一个真实场景假设你在写一个订单服务。调用第三方支付接口时偶尔会失败。你的第一反应是什么失败了就重试啊。好那重试之间要不要等一等毕竟对方服务可能正在抖动立刻重试大概率还是失败。等个 1 秒再试。很合理。那你会怎么写public String callPayment() { for (int i 0; i 3; i) { try { return doCallPayment(); // 调用支付接口 } catch (Exception e) { Thread.sleep(1000); // 等 1 秒再重试 } } throw new RuntimeException(支付调用失败); }这段代码有什么问题1.2 问题出在哪Thread.sleep(1000)做了什么它让当前线程冻结 1 秒。思考如果这段代码运行在线程池里——比如一个 Web 服务器处理请求的线程池——会发生什么线程池共 4 个线程 T1: [处理请求A] → 支付失败 → sleep(1s) → 重试 → 成功 T2: [处理请求B] → 支付失败 → sleep(1s) → 重试 → 成功 T3: [处理请求C] → 支付失败 → sleep(1s) → 重试 → 成功 T4: [处理请求D] → 支付失败 → sleep(1s) → 重试 → 成功 问题4 个线程全在 sleep新的请求 E 来了没人处理核心矛盾线程在sleep期间什么都没干但占着位置不放。就像 4 个厨师都在等锅热但灶台被他们占着别的菜做不了。1.3 我们想要什么我们想要的其实是等 1 秒然后继续做——但等待期间线程应该被释放回去干别的活。理想情况 T1: [处理请求A] → 支付失败 → 释放线程→ (1秒后回来) → 重试 → 成功 → 释放 T2: [处理请求B] → 支付失败 → 释放线程→ (1秒后回来) → 重试 → 成功 → 释放 ↑ 等待期间 T1、T2 可以去处理请求 E、F、G...怎么实现释放线程1 秒后自动回来这就是delayedExecutor要解决的问题。第二章认识 delayedExecutor2.1 它是什么CompletableFuture.delayedExecutor是 JDK 9 引入的一个静态方法。public static Executor delayedExecutor(long delay, TimeUnit unit)一句话它返回一个特殊的Executor——你往这个 Executor 提交的任务不会立即执行而是等指定时间后才执行。2.2 先搞懂 Executor 是什么在 Java 中Executor是一个接口只有一个方法public interface Executor { void execute(Runnable command); // 执行一个任务 }常见的 Executor 实现实现类行为ThreadPoolExecutor提交后立即交给线程池执行ForkJoinPool.commonPool()提交后立即交给公共池执行delayedExecutor(1, SECONDS)提交后等 1 秒再执行关键理解delayedExecutor返回的也是一个Executor只是它的execute方法多了一个延迟行为。2.3 它内部怎么工作你的代码 JDK 内部 ───────── ────────── ┌──────────────────────────────┐ │ Delayer守护线程只有 1 个 │ │ 职责专门负责计时 │ └──────────────────────────────┘ │ delayedExecutor(2, SECONDS) │ .execute(myTask) ──────────▶ 告诉 Delayer2 秒后执行 myTask │ 你的线程池继续干别的事 Delayer 开始计时... 2 秒到了 → 把 myTask 丢到线程池执行要点等待期间你的线程池完全不受影响——线程该干嘛干嘛计时由 JDK 内部唯一的 1 个守护线程负责——它不是你的线程池里的线程零额外线程占用指的是不占你的用户线程池2.4 它解决什么问题不适合解决什么问题适合的场景场景说明非阻塞重试退避失败后等 N 秒重试等待期间不占线程延迟任务调度5 秒后发一条通知、10 秒后检查状态限流/冷却两次操作之间强制间隔超时补偿超时后延迟执行清理逻辑不适合的场景场景为什么不适合更好的选择精确定时任务每天 8 点执行delayedExecutor只做相对延迟不做绝对时间ScheduledExecutorService.scheduleAtFixedRate周期性重复任务它只延迟一次不自动重复ScheduledExecutorService需要取消/修改延迟一旦提交无法取消ScheduledFuture第三章从最简单的例子开始3.1 Hello delayedExecutor先写一个最简单的程序感受延迟执行的效果。import java.util.concurrent.*; /** * 最基础的 delayedExecutor 使用演示。 * * 目标提交一个任务让它 2 秒后才执行。 */ public class HelloDelayedExecutor { public static void main(String[] args) throws Exception { // 记录当前时间用于计算耗时 long start System.currentTimeMillis(); System.out.println([ elapsed(start) ms] 准备提交延迟任务); // ─── 核心代码创建延迟执行器 ─── // delayedExecutor(2, SECONDS) 返回一个 Executor // 这个 Executor 的特点是提交的任务会延迟 2 秒才真正执行 Executor delayed CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS); // ─── 提交任务 ─── // runAsync 把一个 Runnable 任务提交到指定的 Executor 上执行 // 因为我们的 Executor 是 delayedExecutor所以任务会延迟 2 秒 CompletableFuture.runAsync(() - { // 这个 lambda 就是我们要执行的任务 // 它会在 2 秒后才真正运行 System.out.println([ elapsed(start) ms] 任务执行了 线程名 Thread.currentThread().getName()); }, delayed); // ↑ 注意第二个参数指定在哪个 Executor 上执行 // 这里传入的是 delayedExecutor所以会延迟 2 秒 // ─── 验证主线程没有被阻塞 ─── // runAsync 是异步的——它立即返回不会等待任务完成 // 所以这行代码会立刻执行不会等 2 秒 System.out.println([ elapsed(start) ms] 主线程继续执行没有被阻塞); // 等待程序退出否则主线程结束JVM 退出延迟任务来不及执行 Thread.sleep(4000); System.out.println([ elapsed(start) ms] 程序结束); } /** * 计算从 start 到现在经过了多少毫秒 */ static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 准备提交延迟任务 [0ms] 主线程继续执行没有被阻塞 ← 立即执行没有等 2 秒 [2001ms] 任务执行了 线程名ForkJoinPool.commonPool-worker-1 [4001ms] 程序结束观察与思考任务在 2 秒后才执行——延迟生效主线程没有等待——runAsync是异步的执行线程是ForkJoinPool.commonPool-worker-1——不是主线程3.2 理解执行线程问题任务在哪个线程上执行delayedExecutor内部有两步计时由 JDK 内部的 Delayer 守护线程负责执行计时到期后任务被丢到ForkJoinPool.commonPool()执行所以任务实际运行在commonPool的线程上而不是 Delayer 线程上。如果你想让任务在你自己的线程池上执行呢import java.util.concurrent.*; /** * 演示让延迟任务在自定义线程池上执行。 */ public class DelayedWithCustomPool { public static void main(String[] args) throws Exception { // 创建我们自己的线程池方便观察线程名 ExecutorService myPool Executors.newFixedThreadPool(2, r - { Thread t new Thread(r); t.setName(my-pool- t.getId()); // 自定义线程名 return t; }); long start System.currentTimeMillis(); // 方案先用 delayedExecutor 延迟再用 supplyAsync 指定线程池 // // 执行流程 // 1. delayedExecutor 等 1 秒 // 2. 1 秒后把任务丢到 myPool 执行 CompletableFuture.supplyAsync(() - { System.out.println([ elapsed(start) ms] 任务执行线程 Thread.currentThread().getName()); return 结果; }, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS, myPool)); // ↑ 第三个参数指定执行线程池 // 延迟到期后任务在这个线程池上执行 Thread.sleep(3000); myPool.shutdown(); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[1001ms] 任务执行线程my-pool-14关键发现delayedExecutor有一个三参数重载版本// 两参数版延迟后在 commonPool 执行 delayedExecutor(1, TimeUnit.SECONDS) // 三参数版延迟后在指定 Executor 执行 delayedExecutor(1, TimeUnit.SECONDS, myPool)第四章核心场景——非阻塞重试4.1 先用阻塞方式实现重试对比基线import java.util.concurrent.*; import java.util.concurrent.atomic.*; /** * 阻塞式重试Thread.sleep 方式。 * * 问题sleep 期间线程被冻结无法服务其他任务。 */ public class BlockingRetry { // 模拟计数器前 2 次调用失败第 3 次成功 static final AtomicInteger counter new AtomicInteger(0); public static void main(String[] args) throws Exception { // 只有 1 个线程的线程池——问题更明显 ExecutorService pool Executors.newFixedThreadPool(1); long start System.currentTimeMillis(); // 提交 2 个任务 pool.submit(() - retryTask(任务A, start)); pool.submit(() - retryTask(任务B, start)); // ↑ 任务B 必须等任务A 完全结束包括 sleep才能开始 pool.shutdown(); pool.awaitTermination(30, TimeUnit.SECONDS); } static void retryTask(String name, long start) { for (int attempt 1; attempt 3; attempt) { counter.incrementAndGet(); System.out.println([ elapsed(start) ms] name 第 attempt 次尝试线程 Thread.currentThread().getName() ); // 模拟前 4 次调用失败第 5 次成功 if (counter.get() 5) { System.out.println( → 失败等待 1 秒后重试...); try { // ★ 问题所在线程在 sleep 期间完全冻结 ★ Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return; } } else { System.out.println( → 成功); return; } } } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 任务A 第1次尝试线程pool-1-thread-1 → 失败等待 1 秒后重试... ↑ 线程在 sleep任务B 进不来 [1001ms] 任务A 第2次尝试线程pool-1-thread-1 → 失败等待 1 秒后重试... [2001ms] 任务A 第3次尝试线程pool-1-thread-1 → 成功 ↑ 任务A 终于结束任务B 可以开始了 [3001ms] 任务B 第1次尝试线程pool-1-thread-1 → 失败等待 1 秒后重试... [4001ms] 任务B 第2次尝试线程pool-1-thread-1 → 成功问题任务 B 等了 3 秒才开始因为线程池只有 1 个线程被任务 A 的sleep占满了。4.2 用 delayedExecutor 实现非阻塞重试现在用delayedExecutor来解决这个问题。核心思路失败 → 不 sleep → 线程立即释放 → 用 delayedExecutor 设一个1秒后的闹钟 → 闹钟响了 → 再提交一次任务到线程池import java.util.concurrent.*; import java.util.concurrent.atomic.*; /** * 非阻塞重试delayedExecutor 方式。 * * 对比上一节同样的场景1 线程池 2 个任务 重试 * 观察任务 B 是否还需要等待。 */ public class NonBlockingRetry { static final AtomicInteger counter new AtomicInteger(0); public static void main(String[] args) throws Exception { ExecutorService pool Executors.newFixedThreadPool(1); long start System.currentTimeMillis(); // 提交 2 个任务——使用非阻塞重试 CompletableFuture.runAsync(() - nonBlockingRetry(任务A, pool, start), pool); CompletableFuture.runAsync(() - nonBlockingRetry(任务B, pool, start), pool); Thread.sleep(10000); pool.shutdown(); } /** * 非阻塞重试方法。 * * param name 任务名 * param pool 线程池重试时要把任务提交回同一个池 * param start 起始时间用于打印耗时 */ static void nonBlockingRetry(String name, ExecutorService pool, long start) { int count counter.incrementAndGet(); System.out.println([ elapsed(start) ms] name 尝试线程 Thread.currentThread().getName() ); if (count 5) { // 失败但不 sleep——线程立即释放 System.out.println( → 失败1 秒后重试线程已释放...); // ★ 核心代码 ★ // // runAsync(() - {}, ...) 中的 () - {} 是空操作。 // 我们不需要它做事——我们只需要 delayedExecutor 的延迟效果。 // // 整个表达式的含义 // 等 1 秒后执行 whenComplete 里的代码 // // 等 1 秒期间 // - 当前线程已经释放可以去处理其他任务 // - JDK 内部的 Delayer 守护线程在计时 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .whenComplete((v, ex) - { // 1 秒后到达这里——重新尝试 // 把任务提交回线程池 pool.submit(() - nonBlockingRetry(name, pool, start)); }); } else { System.out.println( → 成功); } } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 任务A 尝试线程pool-1-thread-1 → 失败1 秒后重试线程已释放... ↑ 线程释放了任务B 可以进来了 [0ms] 任务B 尝试线程pool-1-thread-1 → 失败1 秒后重试线程已释放... ↑ 两个任务的闹钟同时在计时 [1001ms] 任务A 尝试线程pool-1-thread-1 → 失败1 秒后重试线程已释放... [2001ms] 任务A 尝试线程pool-1-thread-1 → 成功 [2001ms] 任务B 尝试线程pool-1-thread-1 → 成功对比阻塞版 任务A 开始0ms 任务B 开始3001ms ← 等了 3 秒 全部完成5001ms 非阻塞版 任务A 开始0ms 任务B 开始0ms ← 立即开始 全部完成3001ms ← 快了将近 2 秒4.3 更优雅的实现Promise 递归模式上面的实现有个问题每次重试都重新调用nonBlockingRetry方法逻辑分散。更好的方式是用Promise 模式——一个方法返回 Future内部递归链接重试。先理解 Promise 模式import java.util.concurrent.*; /** * Promise 模式入门。 * * 核心思想创建一个 CompletableFuture承诺 * 稍后在某个时机完成它complete 或 completeExceptionally。 * 调用方拿到这个 Future用 whenComplete/get 等待结果。 */ public class PromiseDemo { public static void main(String[] args) throws Exception { long start System.currentTimeMillis(); // 创建一个承诺——稍后会被完成 CompletableFutureString promise new CompletableFuture(); // 2 秒后完成这个承诺 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS)) .whenComplete((v, ex) - { // 2 秒后到达这里 System.out.println([ elapsed(start) ms] 延迟到期完成 promise); promise.complete(Hello from delayedExecutor!); // ↑ 完成 promise调用方拿到结果 }); // 调用方等待 promise 完成 // get() 会阻塞当前线程直到 promise 被 complete String result promise.get(); System.out.println([ elapsed(start) ms] 拿到结果 result); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[2001ms] 延迟到期完成 promise [2001ms] 拿到结果Hello from delayedExecutor!现在用 Promise 模式重写非阻塞重试import java.util.concurrent.*; import java.util.concurrent.atomic.*; import java.util.function.*; /** * 用 Promise delayedExecutor 实现优雅的非阻塞重试。 * * 这个 retryAsync 方法可以直接作为工具类使用。 */ public class RetryWithPromise { static final AtomicInteger counter new AtomicInteger(0); public static void main(String[] args) throws Exception { ExecutorService pool Executors.newFixedThreadPool(2); long start System.currentTimeMillis(); // 使用 retryAsync 工具方法 CompletableFutureString result retryAsync( () - { // 要重试的操作 int c counter.incrementAndGet(); System.out.println([ elapsed(start) ms] 尝试 # c 线程 Thread.currentThread().getName() ); if (c 3) { throw new RuntimeException(失败); // 模拟失败 } return 第 c 次成功; }, 3, // 最大重试次数 500, // 退避延迟毫秒 pool // 线程池 ); // 等待最终结果 System.out.println(最终结果 result.get()); pool.shutdown(); } /** * 非阻塞重试工具方法。 * * param action 要执行的操作可能抛异常 * param maxRetries 最大重试次数 * param backoffMs 退避延迟毫秒 * param executor 线程池 * return 最终结果的 Future */ static T CompletableFutureT retryAsync( CallableT action, int maxRetries, long backoffMs, ExecutorService executor) { // 创建承诺——调用方持有这个 Future 等待最终结果 CompletableFutureT promise new CompletableFuture(); // 启动第一次尝试然后递归链接重试 attemptAndRetry(action, maxRetries, backoffMs, executor, promise, 0); return promise; } /** * 执行一次尝试失败时用 delayedExecutor 延迟后递归重试。 * * param action 要执行的操作 * param maxRetries 最大重试次数 * param backoffMs 退避延迟 * param executor 线程池 * param promise 最终承诺 * param attempt 当前是第几次尝试从 0 开始 */ static T void attemptAndRetry( CallableT action, int maxRetries, long backoffMs, ExecutorService executor, CompletableFutureT promise, int attempt) { // supplyAsync把任务提交到线程池异步执行 // 返回一个 Future任务完成时 Future 自动完成异常时异常完成 CompletableFuture.supplyAsync(() - { try { return action.call(); // 执行用户操作 } catch (Exception e) { // 把受检异常包装为 CompletionException让 supplyAsync 的 Future 异常完成 throw new CompletionException(e); } }, executor) // whenComplete无论成功还是异常回调都执行 // 参数result 成功时的值异常时为 nullex 异常成功时为 null .whenComplete((result, ex) - { if (ex null) { // ─── 成功路径 ─── // 任务成功了完成 promise调用方拿到结果 promise.complete(result); return; } // ─── 异常路径 ─── if (attempt maxRetries) { // 重试次数耗尽传播异常 promise.completeExceptionally(ex); return; } // 还有重试机会——用 delayedExecutor 延迟后重试 System.out.println( 第 (attempt 1) 次重试延迟 backoffMs ms); // ★ 核心delayedExecutor 延迟 → whenComplete 触发 → 递归重试 ★ CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(backoffMs, TimeUnit.MILLISECONDS)) .whenComplete((v, delayEx) - { // 延迟到期递归调用自身进行下一次尝试 attemptAndRetry(action, maxRetries, backoffMs, executor, promise, attempt 1); }); }); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 尝试 #1线程pool-1-thread-1 第 1 次重试延迟 500ms [501ms] 尝试 #2线程pool-1-thread-1 第 2 次重试延迟 500ms [1001ms] 尝试 #3线程pool-1-thread-1 最终结果第 3 次成功执行流程图supplyAsync(尝试#1) → 失败 ↓ whenComplete: 失败还有重试机会 ↓ runAsync(() - {}, delayedExecutor(500ms)) ← 设置 500ms 闹钟 ↓ 线程释放 ... 500ms 后 ... ↓ Delayer: 叮 whenComplete: 延迟到期 ↓ supplyAsync(尝试#2) → 失败 ↓ whenComplete: 失败还有重试机会 ↓ runAsync(() - {}, delayedExecutor(500ms)) ← 再设一个闹钟 ↓ 线程释放 ... 500ms 后 ... ↓ Delayer: 叮 supplyAsync(尝试#3) → 成功 ↓ promise.complete(第 3 次成功)第五章进阶用法5.1 指数退避实际生产中重试延迟通常逐次增加指数退避避免对故障服务造成压力。import java.util.concurrent.*; /** * 指数退避重试每次重试延迟翻倍。 * * 第 1 次重试100ms * 第 2 次重试200ms * 第 3 次重试400ms * 第 4 次重试800ms * ... */ public class ExponentialBackoff { public static void main(String[] args) throws Exception { ExecutorService pool Executors.newFixedThreadPool(2); long start System.currentTimeMillis(); retryWithExponentialBackoff( () - { int c Counter.next(); System.out.println([ elapsed(start) ms] 尝试 # c); if (c 4) throw new RuntimeException(失败); return 成功; }, 5, // 最大重试 5 次 100, // 初始延迟 100ms pool ).get(); pool.shutdown(); } static T CompletableFutureT retryWithExponentialBackoff( CallableT action, int maxRetries, long initialDelayMs, ExecutorService executor) { return doRetry(action, maxRetries, initialDelayMs, executor, new CompletableFuture(), 0); } static T CompletableFutureT doRetry( CallableT action, int maxRetries, long currentDelayMs, ExecutorService executor, CompletableFutureT promise, int attempt) { CompletableFuture.supplyAsync(() - { try { return action.call(); } catch (Exception e) { throw new CompletionException(e); } }, executor).whenComplete((result, ex) - { if (ex null) { promise.complete(result); return; } if (attempt maxRetries) { promise.completeExceptionally(ex); return; } // ★ 指数退避当前延迟 初始延迟 × 2^attempt ★ // attempt0: 100 × 2^0 100ms // attempt1: 100 × 2^1 200ms // attempt2: 100 × 2^2 400ms long delay currentDelayMs; System.out.println( 重试 # (attempt 1) 延迟 delay ms); CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(delay, TimeUnit.MILLISECONDS)) .whenComplete((v, e) - // ★ 下次延迟翻倍 ★ doRetry(action, maxRetries, currentDelayMs * 2, executor, promise, attempt 1)); }); return promise; } // 简单的全局计数器 static class Counter { static int n 0; static int next() { return n; } } static long elapsed(long start) { return System.currentTimeMillis() - start; } }运行结果[0ms] 尝试 #1 重试 #1延迟 100ms [101ms] 尝试 #2 重试 #2延迟 200ms [302ms] 尝试 #3 重试 #3延迟 400ms [703ms] 尝试 #4 最终结果成功5.2 与其他 CompletableFuture 操作组合delayedExecutor可以和其他 CF 操作自由组合import java.util.concurrent.*; /** * delayedExecutor 与其他 CF 操作的组合用法。 */ public class ComposingDelayed { public static void main(String[] args) throws Exception { long start System.currentTimeMillis(); // ─── 场景 1延迟后转换结果 ─── // 等 1 秒然后计算 42 的 2 倍 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenApply(v - 42 * 2) // 延迟后转换 .whenComplete((result, ex) - System.out.println([ elapsed(start) ms] 场景1结果 result)); // 输出[1001ms] 场景1结果84 // ─── 场景 2延迟后执行异步操作 ─── // 等 1 秒然后发起一个异步查询 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenCompose(v - CompletableFuture.supplyAsync(() - 查询结果)) .whenComplete((result, ex) - System.out.println([ elapsed(start) ms] 场景2结果 result)); // 输出[1001ms] 场景2结果查询结果 // ─── 场景 3多个延迟任务并行 ─── // 1 秒后做 A2 秒后做 B等两个都完成 CompletableFutureString taskA CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenApply(v - A完成); CompletableFutureString taskB CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS)) .thenApply(v - B完成); CompletableFuture.allOf(taskA, taskB) .thenApply(v - taskA.join() taskB.join()) .whenComplete((result, ex) - System.out.println([ elapsed(start) ms] 场景3结果 result)); // 输出[2001ms] 场景3结果A完成 B完成 Thread.sleep(4000); } static long elapsed(long start) { return System.currentTimeMillis() - start; } }第六章常见陷阱陷阱 1忘记把任务提交到 delayedExecutor// ❌ 错误只是创建了 delayedExecutor什么都没做 CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS); // 这行代码没有任何效果 // ✅ 正确必须配合 runAsync / supplyAsync 使用 CompletableFuture.runAsync(() - doSomething(), CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));陷阱 2用 supplyAsync 但 lambda 没有返回值// ❌ 编译错误supplyAsync 要求 Supplier 有返回值 CompletableFuture.supplyAsync(() - { System.out.println(执行); // 没有 return编译报错 }, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)); // ✅ 正确无返回值用 runAsync CompletableFuture.runAsync(() - { System.out.println(执行); }, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));陷阱 3以为 delayedExecutor 会中断已有任务// ❌ 误解以为 1 秒后会中断正在运行的 supplyAsync 任务 CompletableFuture.supplyAsync(() - { Thread.sleep(10000); // 任务要跑 10 秒 return done; }, pool); // delayedExecutor 跟上面这个任务没有任何关系 // 它只负责延迟后执行新提交的任务 // ✅ 正确理解delayedExecutor 只是一个定时器 // 它不取消、不中断、不影响任何已经在执行的任务陷阱 4在延迟回调里阻塞等待// ❌ 错误延迟后又去 get() 阻塞——白用了非阻塞延迟 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenApply(v - someOtherFuture.get()); // 阻塞 // ✅ 正确继续用异步操作 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenCompose(v - someOtherFuture); // 非阻塞陷阱 5忽略返回的 Future// ❌ 问题提交了延迟任务但没有持有 Future 引用 // 如果主线程退出延迟任务可能来不及执行 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS)); // 主线程立即结束 → JVM 退出 → 延迟任务没机会执行 // ✅ 正确持有 Future 引用确保程序不会提前退出 CompletableFuture? delayed CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(5, TimeUnit.SECONDS)); delayed.get(); // 等待延迟任务完成 // 或者delayed.join();总结一句话记忆delayedExecutor(delay, unit) 一个定时闹钟告诉它X 秒后做某事然后你就可以走开干别的了。等待期间你的线程池零占用。使用模板// 基本用法延迟后执行操作 CompletableFuture.runAsync(() - { // 这里是要做的事 }, CompletableFuture.delayedExecutor(延迟时间, 时间单位)); // 指定线程池延迟后在自定义线程池上执行 CompletableFuture.supplyAsync(() - { return 结果; }, CompletableFuture.delayedExecutor(延迟时间, 时间单位, 自定义线程池)); // 链式操作延迟后继续做其他事 CompletableFuture.runAsync(() - {}, CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS)) .thenCompose(v - 下一个异步操作) .whenComplete((result, ex) - 处理结果);适用场景速查场景用法非阻塞重试退避失败 →delayedExecutor→ 重新提交任务延迟通知5 秒后发消息 →runAsync(() - send(), delayedExecutor(5, SECONDS))延迟健康检查启动 30 秒后检查 →supplyAsync(() - check(), delayedExecutor(30, SECONDS))操作冷却间隔两次调用间强制间隔 → 链式delayedExecutor超时后清理orTimeout异常 →whenComplete→delayedExecutor→ 延迟清理