从零手写FutureTask：深入AQS共享模式实现get阻塞与唤醒

1. 从一个“卡死”的线程说起为什么要自己动手实现 FutureTask多线程编程里最让人头疼的场景之一就是主线程提交了一个任务之后不知道该什么时候去拿结果。拿早了任务还没跑完只能干等拿晚了又浪费了并行的时间窗口。FutureTask这个类就是为解决这个矛盾而生的——它既是Runnable又是Future既能被线程池执行又能让调用方在合适的时机去取结果。但很多人用FutureTask用了好几年对它的内部机制依然是一知半解。比如get()方法在任务没完成时到底是怎么阻塞的它是靠synchronized还是靠LockSupport任务完成后又是怎么唤醒等待线程的这些问题如果不去翻源码很难说清楚。我这段时间在梳理 AQSAbstractQueuedSynchronizer的应用场景发现FutureTask其实是一个被低估的 AQS 实战案例。大家平时聊 AQS张口就是ReentrantLock、Semaphore、CountDownLatch但FutureTask内部同样用 AQS 实现了一套精妙的同步机制而且它的用法和前面几个都不太一样——它用的是共享模式而不是独占模式。所以这篇文章我打算带着大家从零实现一个简化版的FutureTask重点拆解get()方法的阻塞与唤醒逻辑把 AQS 共享模式在其中的作用讲透。适合已经了解 AQS 基本结构state、CLH 队列、tryAcquire/tryRelease但还没深入看过FutureTask源码的读者。如果你对 AQS 还完全陌生建议先补一下ReentrantLock的加锁流程再来看这篇会顺畅很多。2. 先搞清楚 FutureTask 的设计骨架2.1 它到底承担了几个角色在动手写代码之前得先把FutureTask的职责理清楚。它同时实现了Runnable和Future两个接口这意味着它有两个身份作为 Runnable它可以被丢给线程或线程池去执行run()方法里会调用真正的业务逻辑Callable并把结果存起来。作为 Future它提供了get()、cancel()、isDone()等方法让调用方可以获取结果、取消任务、查询状态。这两个身份之间需要一个“状态桥梁”来协调——任务没跑完时get()要阻塞任务跑完了get()要立刻返回结果。这个桥梁就是state字段加上 AQS 的等待队列。2.2 状态流转是整个设计的核心FutureTask内部定义了一组状态常量用来描述任务的生命周期。简化后大致是这样的状态含义是否终态NEW任务刚创建还没开始跑否COMPLETING任务已跑完正在写入结果中间态否NORMAL任务正常完成结果已写入是EXCEPTIONAL任务抛异常异常已写入是CANCELLED任务被取消是INTERRUPTED任务被中断取消是状态只能从NEW出发经过COMPLETING到达NORMAL或EXCEPTIONAL或者直接从NEW跳到CANCELLED/INTERRUPTED。一旦进入终态就再也不会改变。注意COMPLETING这个中间态很容易被忽略。它的存在是为了保证“结果写入”和“状态变更”这两步操作的顺序性。如果先把状态改成NORMAL再写结果那么另一个线程可能在结果还没写入时就看到了NORMAL状态读到一个空结果。所以必须先设COMPLETING写完结果后再用 CAS 改成NORMAL。2.3 为什么用 AQS 共享模式而不是独占模式这是理解FutureTask的关键问题。ReentrantLock用的是独占模式同一时刻只有一个线程能持有锁。但FutureTask的场景不一样可能有多个线程同时调用get()等待同一个任务的结果。任务完成后这些线程应该全部被唤醒而不是只唤醒一个。这正是 AQS 共享模式SHARED的用武之地。共享模式下tryAcquireShared返回正数表示获取成功并且会传播唤醒后续节点。FutureTask内部就是通过一个Sync类继承 AQS重写tryAcquireShared和tryReleaseShared来实现的。简单说独占模式是“一把钥匙开一把锁”共享模式是“一个信号弹通知所有等待的人”。FutureTask需要的是后者。3. 手写一个简化版 FutureTask核心结构搭建3.1 定义状态与成员变量先把骨架搭起来。我们需要一个Callable来承载业务逻辑一个Object来存结果或异常一个volatile int来表示状态以及一个继承 AQS 的内部同步器。public class SimpleFutureTaskV implements Runnable, FutureV { private CallableV callable; private Object outcome; // 存结果或异常 private volatile int state; // 任务状态 private volatile Thread runner; // 执行任务的线程 private final Sync sync new Sync(); private static final int NEW 0; private static final int COMPLETING 1; private static final int NORMAL 2; private static final int EXCEPTIONAL 3; private static final int CANCELLED 4; public SimpleFutureTask(CallableV callable) { if (callable null) throw new NullPointerException(); this.callable callable; } }这里有几个细节值得说。outcome用Object而不是V是因为它既要存正常结果又要存异常对象用Object可以统一处理。runner记录当前执行任务的线程主要是为了支持cancel(true)时能中断它。state用volatile修饰保证多线程之间的可见性。3.2 内部 Sync 类的设计Sync继承AbstractQueuedSynchronizer只需要重写两个方法private final class Sync extends AbstractQueuedSynchronizer { // 共享模式获取state ! NEW 表示任务已完成可以获取 Override protected int tryAcquireShared(int arg) { return (getState() NORMAL || getState() EXCEPTIONAL || getState() CANCELLED) ? 1 : -1; } // 共享模式释放把 state 设为终态返回 true 触发唤醒 Override protected boolean tryReleaseShared(int arg) { setState(NORMAL); return true; } }tryAcquireShared返回1表示获取成功返回-1表示需要排队等待。tryReleaseShared返回true表示释放成功AQS 会负责唤醒队列中所有等待的共享节点。提示真实的FutureTask源码里tryAcquireShared的判断更复杂会区分NORMAL、EXCEPTIONAL、CANCELLED、INTERRUPTED等多种终态。这里为了简化只保留了最核心的几种。3.3 run 方法的实现run()方法是任务的入口它负责调用Callable、写入结果、更新状态、唤醒等待者Override public void run() { // 只有 NEW 状态才能执行防止重复执行 if (state ! NEW || !sync.compareAndSetState(NEW, COMPLETING)) { return; } runner Thread.currentThread(); try { V result callable.call(); setOutcome(result); sync.releaseShared(0); // 唤醒所有等待 get() 的线程 } catch (Throwable t) { setOutcome(t); sync.releaseShared(0); } finally { runner null; } } private void setOutcome(Object result) { outcome result; // 这里用 CAS 把 COMPLETING 改成 NORMAL // 简化处理实际源码用 UNSAFE 或 VarHandle state NORMAL; }这里有个容易踩的坑compareAndSetState(NEW, COMPLETING)这一步很关键。它保证了同一个任务不会被两个线程同时执行。如果两个线程同时拿到这个FutureTask并调用run()只有一个能成功把状态从NEW改成COMPLETING另一个直接返回。4. get 方法的阻塞与唤醒AQS 共享模式的实战4.1 get 方法的整体逻辑get()方法的核心就一句话如果任务没完成就排队等待如果完成了就返回结果或抛出异常。用 AQS 的语言来说就是调用acquireSharedInterruptibly(0)Override public V get() throws InterruptedException, ExecutionException { int s state; if (s COMPLETING) { // 任务未完成进入 AQS 队列等待 s sync.acquireSharedInterruptibly(0); } if (s CANCELLED) { throw new CancellationException(); } return report(s); }acquireSharedInterruptibly是 AQS 提供的模板方法它会先调用我们重写的tryAcquireShared如果返回负数就把当前线程包装成共享节点加入 CLH 队列然后park自己。当任务完成时releaseShared会唤醒队列中的节点被唤醒的线程再次尝试tryAcquireShared这次返回正数于是成功返回。4.2 为什么 get 方法要区分“快路径”和“慢路径”注意上面代码里的if (s COMPLETING)判断。这是一个快路径优化如果任务已经完成了state是NORMAL或EXCEPTIONAL根本不需要去碰 AQS 队列直接读结果就行。只有任务还没完成时才走 AQS 的排队逻辑。这个设计的意义在于get()方法在任务完成后被调用的频率可能非常高比如轮询场景如果每次都走 AQS 的acquire流程即使不阻塞也会有 CAS 和队列操作的开销。快路径用一个简单的volatile读就搞定了性能好很多。实操心得我在自己实现的时候一开始把if判断去掉了直接调acquireSharedInterruptibly。结果发现任务完成后调用get()依然会走一遍完整的 AQS 流程虽然不会阻塞但吞吐量明显下降。加上快路径判断后性能提升非常明显。这个细节在源码里很容易被忽略但它是真实存在的优化。4.3 report 方法结果与异常的分离处理get()最终返回结果时需要区分“正常结果”和“异常”两种情况private V report(int s) throws ExecutionException { Object x outcome; if (s NORMAL) { return (V) x; } if (s CANCELLED) { throw new CancellationException(); } throw new ExecutionException((Throwable) x); }这里用state来判断outcome里存的是结果还是异常。如果是NORMALoutcome就是正常结果如果是EXCEPTIONALoutcome就是异常对象需要包装成ExecutionException抛出。这种“用一个字段存两种东西靠状态区分”的做法在 JDK 源码里很常见好处是省内存坏处是可读性稍差。4.4 唤醒传播共享模式的精髓当run()方法执行完调用sync.releaseShared(0)时AQS 内部会做这几件事调用tryReleaseShared把state设为NORMAL返回true。检查 CLH 队列的head节点如果head.next是共享节点就唤醒它。被唤醒的节点尝试tryAcquireShared成功后继续唤醒它的后继节点这就是“传播”。这个传播机制保证了所有等待get()的线程都会被依次唤醒而不是只唤醒一个。如果用的是独占模式release只会唤醒一个节点其他线程还得继续等这显然不符合FutureTask的语义。我画一个简化的流程帮助理解用文字描述线程 A、B、C 都调用了get()任务未完成三个线程依次入队。任务执行完毕releaseShared被调用。AQS 唤醒队列中第一个节点线程 A。线程 A 的tryAcquireShared返回 1成功获取同时触发传播唤醒线程 B。线程 B 同样成功获取并唤醒线程 C。三个线程都拿到结果返回。整个过程不需要额外的notifyAllAQS 的共享模式天然支持这种“一对多”的唤醒语义。5. 实操中容易踩的坑与排查技巧5.1 状态判断用 还是 在get()方法里判断任务是否完成时我用的是s COMPLETING。为什么不是s NEW因为COMPLETING也是一个“未完成”的中间态——此时任务逻辑已经跑完但结果还没写入outcome。如果此时让get()返回可能读到旧值或空值。所以COMPLETING也必须走等待流程。这个细节很容易写错。我见过有人在实现时用s NEW判断结果在COMPLETING窗口期调用get()会直接返回null造成偶发的空指针问题。这种 bug 极难复现因为COMPLETING窗口非常短。5.2 异常处理的顺序问题run()方法里setOutcome和releaseShared的调用顺序不能反。必须先写结果再释放同步状态。如果先releaseShared等待线程被唤醒后可能读到还没写入的outcome拿到null。同理setOutcome内部也要先写outcome再改state。这就是为什么需要COMPLETING这个中间态——它保证了“写结果”和“改状态”之间的顺序性。5.3 中断响应的处理acquireSharedInterruptibly会在等待过程中响应中断。如果线程在get()时被中断会抛出InterruptedException。但要注意此时任务可能还在执行FutureTask本身不会因为get()被中断而取消任务。这是符合Future接口语义的——get()的中断只影响等待行为不影响任务执行。如果需要“中断等待的同时取消任务”得自己额外调用cancel(true)。这一点在实际使用中经常被混淆值得注意。5.4 常见问题速查表问题现象可能原因排查方向get() 一直阻塞不返回run() 没被调用或 state 没更新检查任务是否提交到线程池检查 run() 里的 CAS 是否成功get() 返回 nullCOMPLETING 状态被误判为完成检查状态判断是否用了 COMPLETING多个 get() 只有一个被唤醒用了独占模式而非共享模式检查 tryAcquireShared 的返回值确认是共享模式任务被执行两次run() 缺少 CAS 状态检查检查 compareAndSetState(NEW, COMPLETING) 是否存在get() 抛出的异常类型不对report 方法状态判断有误检查 NORMAL/EXCEPTIONAL/CANCELLED 的分支避坑技巧调试 AQS 相关问题时可以在tryAcquireShared和tryReleaseShared里加日志打印当前线程名和 state 值。AQS 的问题往往出在状态判断上把状态流转打出来一眼就能看出问题。6. 从简化版到真实源码还差哪些细节6.1 状态变更的原子性保障我上面的简化版里setOutcome直接用了state NORMAL赋值。但真实源码里这一步是用 CAS 完成的通过UNSAFE.compareAndSwapInt或VarHandle。为什么要用 CAS因为COMPLETING到NORMAL的转换虽然只有一个线程会执行run()里已经保证了但cancel()方法可能并发地把状态改成CANCELLED。用 CAS 可以保证只有一个操作成功避免状态被覆盖。6.2 cancel 方法的实现要点cancel(boolean mayInterruptIfRunning)的逻辑比get()复杂一些Override public boolean cancel(boolean mayInterruptIfRunning) { if (!(state NEW sync.compareAndSetState(NEW, CANCELLED))) { return false; } if (mayInterruptIfRunning) { Thread t runner; if (t ! null) { t.interrupt(); } } sync.releaseShared(0); // 唤醒等待的 get() 线程 return true; }关键点取消成功后也要调用releaseShared否则等待get()的线程会一直阻塞。这是一个很容易遗漏的地方——很多人只记得任务完成时唤醒忘了取消时也要唤醒。6.3 泛型与类型转换的处理真实源码里outcome的类型转换用了(V)强转编译器会给出 unchecked 警告。JDK 源码里用SuppressWarnings(unchecked)压制。自己实现时如果不想看到警告也可以这样做。但要注意强转的安全性是由state保证的——只有NORMAL状态下outcome才是V类型其他状态下可能是Throwable。6.4 与线程池的配合FutureTask通常和线程池配合使用。ThreadPoolExecutor.submit()内部就是把Callable包装成FutureTask然后调用execute()。理解这一点后就能明白为什么FutureTask要实现Runnable——因为线程池只接受Runnable。自己实现时可以写一个简单的测试public static void main(String[] args) throws Exception { SimpleFutureTaskString task new SimpleFutureTask(() - { Thread.sleep(2000); return done; }); new Thread(task).start(); System.out.println(等待结果...); String result task.get(); System.out.println(结果: result); }运行后应该看到“等待结果...”先打印2 秒后才打印“结果: done”。如果把get()换成get(1, TimeUnit.SECONDS)应该抛出TimeoutException。7. 几个值得深挖的延伸问题7.1 为什么不用 synchronized wait/notify有人可能会问用synchronized加wait/notifyAll也能实现类似效果为什么非要用 AQS原因有几个性能AQS 的 CLH 队列在竞争激烈时比wait/notifyAll更高效因为notifyAll会唤醒所有等待线程然后它们一起竞争锁造成“惊群效应”。AQS 的共享模式虽然也是唤醒所有但每个节点获取成功后直接返回不需要重新竞争锁。可中断性AQS 提供了acquireSharedInterruptibly天然支持中断响应。用wait/notify需要自己处理InterruptedException。超时支持AQS 提供了tryAcquireSharedNanos实现带超时的get()很方便。7.2 共享模式与独占模式的本质区别维度独占模式共享模式获取方法tryAcquiretryAcquireShared返回值booleanint负数失败非负数成功唤醒行为只唤醒后继一个节点唤醒后继节点并传播典型应用ReentrantLockCountDownLatch、Semaphore、FutureTask理解这个区别后再看FutureTask的Sync实现就会觉得非常自然——它需要的就是“任务完成后通知所有等待者”这正是共享模式的语义。7.3 超时 get 的实现思路带超时的get(long timeout, TimeUnit unit)用的是tryAcquireSharedNanos它在acquireSharedInterruptibly的基础上增加了超时判断。如果超时还没获取到就返回负数然后get()抛出TimeoutException。实现时需要注意超时后要把节点从队列中移除否则会造成内存泄漏。AQS 内部已经处理了这一点自己实现时如果不用 AQS就得手动清理。8. 我个人的一些实践体会自己动手实现一遍FutureTask之后我对 AQS 共享模式的理解上了一个台阶。以前看CountDownLatch的源码总觉得tryAcquireShared返回1和-1的语义有点绕现在明白了返回非负数表示“可以继续”返回负数表示“需要等待”。这个约定贯穿了所有共享模式的实现。另外有一个小技巧分享如果你在调试 AQS 相关代码时发现线程莫名其妙地卡住可以在park前后加日志或者用jstack看线程状态。AQS 的等待线程状态是WAITING (parking)如果看到大量这种状态的线程基本可以确定是某个release没被调用。还有一个容易忽略的点FutureTask的get()在任务被取消后调用会抛出CancellationException而不是ExecutionException。这个异常是RuntimeException的子类不需要显式捕获。很多人第一次遇到时会困惑为什么catch (ExecutionException e)没接住其实就是异常类型搞错了。最后说一个扩展方向如果你对 AQS 感兴趣可以试着把FutureTask改造成支持“多个任务依赖”的版本——比如任务 B 依赖任务 A 的结果A 完成后自动触发 B。这其实就是CompletableFuture的雏形。理解了FutureTask的同步机制再看CompletableFuture的Completion链式结构会容易很多。