1. 从AQS到FutureTask为什么值得亲手造一遍轮子刚入行那会儿我对FutureTask的理解一直停留在“线程池提交任务后返回的那个东西调get()就能拿结果”。直到有一次排查线上问题发现某个接口在并发量上来之后线程池里的任务明明已经跑完了但调用方还是卡在get()上迟迟不返回最后定位到是任务内部抛了异常但没人处理get()重新抛出时把整个调用链给堵死了。那次之后我才意识到FutureTask远不是“拿结果”这么简单它背后是一整套基于AQSAbstractQueuedSynchronizer的同步状态管理机制。这篇文章的核心目标很明确用AQS亲手实现一个简化版的FutureTask重点把get()方法的阻塞与唤醒逻辑讲透。为什么选这个角度因为大部分讲AQS的文章都在讲ReentrantLock但ReentrantLock是独占锁的典型而FutureTask用的是共享模式的同步器两者的状态流转、节点唤醒策略完全不同。把FutureTask的get()用AQS实现一遍你对AQS的理解会从“知道有这个东西”直接跳到“能自己设计一个同步组件”。适合谁来读如果你已经能看懂ReentrantLock的加锁解锁流程知道AQS里有state、有CLH队列、有Node节点但对共享模式怎么工作、get()怎么阻塞、任务完成后怎么唤醒等待线程这些细节还模模糊糊那这篇就是写给你的。我会从设计思路开始拆然后逐行分析核心代码最后把实操中容易踩的坑和排查技巧一并交代清楚。全程不贴大段源码让你自己悟而是把每个关键决策背后的“为什么”讲明白。2. 整体设计思路为什么FutureTask的同步器要用共享模式2.1 先搞清楚FutureTask到底要解决什么问题FutureTask本质上是一个可取消的异步计算载体。它同时实现了Runnable和Future接口这意味着它既能被线程执行run()又能对外提供获取结果、取消任务、查询状态的能力get()、cancel()、isDone()。这三个能力里get()是最核心也最复杂的因为它涉及一个线程等待另一个线程完成计算这个跨线程协作场景。跨线程协作在Java里有两种经典实现路径一种是基于Object的wait/notify另一种就是基于AQS的同步队列。FutureTask选了后者原因很直接——wait/notify需要自己维护等待队列、自己处理中断、自己保证状态可见性代码又臭又长还容易出bug。AQS已经把这些脏活累活都封装好了你只需要定义好state的含义和tryAcquireShared/tryReleaseShared的逻辑就行。2.2 共享模式 vs 独占模式选错的代价AQS有两种模式独占Exclusive和共享Shared。ReentrantLock用独占同一时刻只有一个线程能持有锁FutureTask用共享因为多个线程可以同时等待同一个任务的完成结果。这个区别非常关键选错了整个设计就崩了。我举个具体场景你就明白了。假设你提交了一个计算量很大的任务到线程池然后有三个线程都调用了同一个FutureTask的get()。如果是独占模式第一个线程拿到“锁”之后开始等待第二个、第三个线程会被挡在门外它们甚至没法进入等待队列的正确位置。而共享模式下这三个线程都能以共享节点的身份进入AQS的等待队列任务完成后统一被唤醒各自去拿结果。这就是为什么FutureTask内部用的是tryAcquireShared和tryReleaseShared而不是tryAcquire和tryRelease。2.3 状态机的设计state到底代表什么AQS的state是一个volatile int具体含义由使用者定义。在FutureTask里state承载了任务的完整生命周期状态值常量名含义0NEW任务新建尚未开始执行1COMPLETING任务正在完成中中间态2NORMAL任务正常完成结果已设置3EXCEPTIONAL任务抛出异常结束4CANCELLED任务被取消5INTERRUPTING正在中断执行线程中间态6INTERRUPTED已中断执行线程这里有个设计细节值得注意COMPLETING和INTERRUPTING是两个中间态存在时间极短但它们的存在是为了保证状态转换的原子性。比如从NEW到NORMAL必须先经过COMPLETING先把结果赋值到outcome字段再用CAS把state从COMPLETING改成NORMAL。这样做的目的是确保任何线程看到state NORMAL时outcome一定已经写好了不会读到null。注意中间态的设计是AQS同步组件里非常经典的手法。你自己设计同步器时如果状态转换涉及多个字段的写入一定要考虑是否需要中间态来保证可见性顺序。2.4 等待队列的节点复用共享节点的传播机制AQS的等待队列是一个双向链表每个节点代表一个等待线程。在共享模式下当一个节点被唤醒并成功获取到同步状态后它会继续唤醒它的后继节点这就是所谓的“传播”propagation。FutureTask的get()正是利用了这个机制任务完成后releaseShared会唤醒队列头部的等待节点该节点获取到“完成”状态后又会去唤醒下一个等待节点直到所有等待线程都被唤醒。这个传播过程在doReleaseShared方法里实现是AQS共享模式最精妙也最难理解的部分。我在第一次读的时候卡了很久后来画了好几遍队列变化图才理清楚。后面实操部分我会用具体的队列快照来演示这个过程。3. 核心细节解析get()方法的阻塞与唤醒全流程3.1 get()方法的入口逻辑get()方法有两个版本无参的get()会一直阻塞直到任务完成带超时的get(timeout, unit)最多阻塞指定时间。两者的核心逻辑类似区别在于超时版本在入队前会先检查一次状态如果没完成就进入doAcquireSharedNanos在自旋过程中每次被唤醒都会检查是否超时。先看无参版本的简化逻辑public V get() throws InterruptedException, ExecutionException { int s state; if (s COMPLETING) { s awaitDone(false, 0L); } return report(s); }这里有几个关键点。第一s COMPLETING这个判断覆盖了NEW和COMPLETING两个状态也就是说任务还没开始、正在执行、正在完成中这三种情况都需要等待。第二awaitDone是真正干活的它负责把当前线程包装成AQS节点入队、阻塞、被唤醒后重新检查状态。第三report(s)根据最终状态决定是返回结果还是抛出异常。3.2 awaitDone入队与阻塞的核心awaitDone是整个get()方法里最复杂的部分它把AQS的入队、阻塞、中断处理、超时处理全部揉在了一起。我把它拆成几个阶段来看第一阶段快速检查。进入awaitDone后先检查state如果已经完成就直接返回避免不必要的入队。这个优化很重要因为很多情况下任务在你调get()之前就已经跑完了。第二阶段入队。如果任务没完成就把当前线程包装成一个共享模式的Node通过CAS插入到等待队列的尾部。这里用的是addWaiter(Node.SHARED)注意节点类型是SHARED而不是EXCLUSIVE。第三阶段自旋阻塞。入队后进入一个for(;;)循环每次循环做几件事检查当前节点的前驱是不是头节点如果是就尝试tryAcquireShared检查是否被中断检查是否超时如果都不满足就通过LockSupport.park阻塞当前线程。第四阶段清理。被唤醒并成功获取到状态后把当前节点设为头节点断开与前驱的连接返回最终状态。3.3 tryAcquireShared判断任务是否完成tryAcquireShared在FutureTask里的实现非常简洁protected int tryAcquireShared(int arg) { return (state NORMAL || state EXCEPTIONAL || state CANCELLED || state INTERRUPTED) ? 1 : -1; }返回1表示获取成功返回-1表示需要继续等待。注意这里判断的是“任务是否已经进入终态”而不是“任务是否正常完成”。因为即使任务抛了异常或者被取消了等待线程也应该被唤醒去处理这些情况而不是一直傻等。实操心得很多人在自己实现同步器时容易把tryAcquireShared的返回值语义搞混。记住一个原则——返回值大于等于0表示获取成功小于0表示失败需要入队等待。具体返回1还是其他正数不重要重要的是符号。3.4 tryReleaseShared任务完成时的状态发布tryReleaseShared在任务完成时被调用负责把state从中间态改成终态并返回true触发AQS的唤醒流程protected boolean tryReleaseShared(int arg) { // 由set()或setException()调用state已经在之前被CAS改成了中间态 // 这里只需要把中间态改成终态 if (state COMPLETING) { state NORMAL; return true; } // 其他终态直接返回true return true; }实际上在完整的FutureTask实现里状态转换是在set()和setException()方法里完成的tryReleaseShared更多是配合AQS的releaseShared来触发唤醒。这个设计的好处是把“状态变更”和“唤醒等待者”两个动作解耦了。3.5 中断处理等待线程被中断时怎么办get()方法声明了throws InterruptedException这意味着等待线程被中断时应该抛出异常并退出等待。在awaitDone的自旋循环里每次循环都会检查Thread.interrupted()如果发现中断标志被设置就把当前节点从队列中移除然后抛出InterruptedException。这里有个容易忽略的细节移除节点后需要调用unparkSuccessor来唤醒后继节点否则如果被移除的节点后面还有等待线程它们可能永远等不到唤醒。这个逻辑在AQS的cancelAcquire方法里有完整实现自己写的时候很容易漏掉。4. 实操过程从零实现一个简化版FutureTask4.1 定义状态常量与核心字段先把骨架搭起来。我们需要一个继承自AQS的内部同步器类以及任务状态、结果、执行线程这几个核心字段public class SimpleFutureTaskV implements Runnable { private final Sync sync new Sync(); private CallableV callable; private Object outcome; private volatile Thread runner; // 状态常量 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; private static final int INTERRUPTING 5; private static final int INTERRUPTED 6; }outcome字段用Object类型是为了同时容纳正常结果和异常对象。runner记录当前执行任务的线程用于中断操作。sync是AQS同步器的实例所有状态管理都委托给它。4.2 实现Sync同步器Sync类继承AbstractQueuedSynchronizer重写tryAcquireShared和tryReleaseSharedprivate final class Sync extends AbstractQueuedSynchronizer { protected int tryAcquireShared(int arg) { return isDone() ? 1 : -1; } protected boolean tryReleaseShared(int arg) { return true; } boolean isDone() { int s getState(); return s ! NEW s ! COMPLETING; } void releaseShared() { // 触发AQS的共享模式释放 super.releaseShared(0); } }注意isDone()的判断条件只要不是NEW和COMPLETING就认为任务已经结束。这个判断在tryAcquireShared和get()里都会用到抽成一个方法避免重复。4.3 实现run()方法任务执行与状态发布run()方法是任务执行的入口它需要处理几种情况任务已经被取消、任务已经被其他线程执行、正常执行并设置结果、执行过程中抛异常public void run() { // 如果状态不是NEW说明已经被执行或取消 if (sync.getState() ! NEW) { return; } // CAS设置runner防止并发执行 if (!sync.compareAndSetState(NEW, COMPLETING)) { return; } runner Thread.currentThread(); try { CallableV c callable; if (c ! null) { V result c.call(); set(result); } } catch (Throwable ex) { setException(ex); } finally { runner null; } }这里我用compareAndSetState(NEW, COMPLETING)来保证只有一个线程能执行任务同时把状态推进到中间态。这个CAS操作是整个并发安全的关键。4.4 set与setException结果发布与唤醒set()方法负责把正常结果写入outcome然后把状态从COMPLETING改成NORMAL最后调用releaseShared唤醒所有等待线程private void set(V v) { outcome v; // 状态从COMPLETING改为NORMAL sync.setState(NORMAL); sync.releaseShared(); } private void setException(Throwable t) { outcome t; sync.setState(EXCEPTIONAL); sync.releaseShared(); }注意这里我简化了状态转换直接用setState而不是CAS。在完整实现里应该用CAS保证原子性但因为我们前面已经用CAS把状态从NEW改成了COMPLETING所以这里直接设置是安全的。4.5 get()方法等待与结果获取最后是核心的get()方法public V get() throws InterruptedException, ExecutionException { int s sync.getState(); if (s COMPLETING) { s awaitDone(false, 0L); } return report(s); } private int awaitDone(boolean timed, long nanos) throws InterruptedException { long deadline timed ? System.nanoTime() nanos : 0L; Node node null; for (;;) { int s sync.getState(); if (s COMPLETING) { // 任务已完成清理节点 if (node ! null) { node.next null; } return s; } if (Thread.interrupted()) { // 被中断移除节点并抛出异常 removeWaiter(node); throw new InterruptedException(); } if (node null) { node new Node(Thread.currentThread(), Node.SHARED); } else if (timed nanos 0) { // 超时 removeWaiter(node); return s; } // 入队并阻塞 if (node.waiter ! null) { LockSupport.park(this); } } }这段代码是简化版实际AQS的awaitDone要复杂得多涉及节点的正确入队、前驱节点的检查、shouldParkAfterFailedAcquire的调用等。但核心逻辑就是循环检查状态没完成就阻塞被唤醒后继续检查。4.6 report方法根据状态返回结果或抛异常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); }report方法根据最终状态决定行为NORMAL返回结果CANCELLED及以上抛取消异常EXCEPTIONAL抛执行异常。这个设计把状态判断和结果处理集中在一处逻辑清晰。5. 常见问题与排查技巧实录5.1 get()一直阻塞不返回这是最常见的问题。排查思路按以下顺序来排查点可能原因解决方法任务是否真的执行了线程池拒绝策略、任务被取消检查线程池状态和任务提交日志状态是否卡在COMPLETINGset()方法抛异常导致状态未推进在set()里加try-finally保证状态推进等待队列是否死锁节点入队逻辑有bug打印队列快照检查节点连接关系是否被中断但未处理中断标志被清除检查awaitDone的中断处理逻辑我遇到过一次状态卡在COMPLETING的情况原因是set()方法里写outcome时抛了OutOfMemoryError导致后面的setState(NORMAL)没执行。后来在set()外面包了一层try-finally保证无论outcome赋值是否成功状态都要推进到终态。5.2 多个线程同时get()只有一个被唤醒这是共享模式传播机制没生效的典型症状。检查tryReleaseShared的返回值必须返回true才能触发AQS的doReleaseShared。另外检查tryAcquireShared的返回值如果返回0而不是正数AQS会认为获取失败不会继续传播唤醒。实操心得AQS共享模式的传播逻辑是“一个节点获取成功后会尝试唤醒下一个节点”。如果你的实现里只有第一个等待线程被唤醒大概率是tryAcquireShared返回值不对或者节点的waitStatus没有正确设置成PROPAGATE。5.3 任务完成后get()抛出的异常不对report方法里的状态判断顺序很重要。s CANCELLED这个判断必须放在EXCEPTIONAL之前因为CANCELLED、INTERRUPTING、INTERRUPTED的值都大于EXCEPTIONAL。如果顺序写反了取消的任务会抛出ExecutionException而不是CancellationException。5.4 中断等待线程后队列出现断裂在awaitDone里处理中断时移除节点后必须调用unparkSuccessor唤醒后继节点。我最初的实现里只做了node.next null结果队列中间断了一截后面的等待线程永远等不到唤醒。正确的做法是调用AQS提供的cancelAcquire方法它会处理好节点移除和唤醒后继的逻辑。5.5 超时get()返回后任务还在执行这是正常行为。get(timeout)超时返回后任务并不会被取消它可能还在后台继续执行。如果你希望超时后取消任务需要在get返回后手动调用cancel(true)。这个设计是合理的因为超时只是调用方不想等了不代表任务本身需要被终止。6. 从AQS实现FutureTask中学到的设计经验6.1 状态机的中间态不是多余的COMPLETING和INTERRUPTING这两个中间态乍看之下好像可以省略直接CAS从NEW到NORMAL不行吗不行。因为outcome字段的写入和state的变更不是原子的如果先改state再写outcome其他线程可能看到state NORMAL但outcome还是null。中间态的作用就是在状态变更和字段写入之间建立一个顺序保证先把outcome写好再把state从中间态改成终态这样任何看到终态的线程一定能看到完整的outcome。6.2 共享模式的传播是链式的AQS共享模式的唤醒不是“一次性唤醒所有等待线程”而是“唤醒第一个第一个再唤醒第二个依次传播”。这个链式传播的设计是为了避免惊群效应同时保证每个被唤醒的线程都有机会去竞争同步状态。理解这一点你就能明白为什么doReleaseShared里有一个for(;;)循环以及为什么PROPAGATE状态是必要的。6.3 中断处理要保证队列完整性等待线程被中断时不能简单地把它从队列里删掉就完事。因为AQS的等待队列是一个双向链表删除中间节点需要同时更新前驱的next和后继的prev还要保证后继节点能被正确唤醒。这就是为什么AQS提供了cancelAcquire这个专门的方法来处理取消操作而不是让使用者自己操作节点指针。6.4 超时检查要放在自旋循环里带超时的get不能只在入队前检查一次超时因为线程可能在park之后被虚假唤醒或者在等待过程中时间已经过了。正确的做法是在自旋循环的每一轮都检查是否超时一旦超时就移除节点并返回。这个细节在实现时很容易漏掉导致超时时间不准确。6.5 自己实现同步组件的通用套路把这次实现FutureTask的经验抽象一下自己设计一个基于AQS的同步组件大致遵循这个套路定义state的含义明确每个状态值代表什么状态之间的转换路径是什么。确定共享还是独占多个线程能否同时获取同步状态能就用共享不能就用独占。实现tryAcquire/tryRelease这是同步器的核心逻辑决定什么时候能获取、什么时候能释放。处理中断和超时如果对外方法声明了InterruptedException必须在自旋循环里处理中断。保证状态可见性涉及多字段写入时考虑用中间态或volatile保证顺序。这套路适用于绝大多数基于AQS的同步组件包括信号量、倒计时门闩、读写锁等。把FutureTask吃透了再去看这些组件的源码你会发现它们的骨架惊人地相似。最后分享一个我调试AQS同步器时常用的小技巧在关键路径上打印state和队列节点的waitStatus用jstack抓线程栈看哪些线程在park状态。AQS的问题十有八九出在状态转换和队列操作上把这两个东西盯紧了大部分bug都能快速定位。