FutureTask 源码级深度解析

灬沙师弟

发布于 2025-11-12 13:40:18

2260

一、为什么需要 FutureTask？

接口	特点
Runnable	无返回、不能抛受检异常
Callable	有返回、可抛异常，但无法直接提交给 Thread
Future	提供get/cancel/isDone，但创建需线程池

FutureTask = Runnable + Future + Callable 的粘合剂 它同时实现 Runnable 与 Future<V>，可以被线程池或裸线程执行，也能被调用方拿到结果，是高并发“单线程查库”场景的最轻量利器。

二、类图与核心成员

public class FutureTask<V> implements RunnableFuture<V> {
    private volatile int state;           // 任务状态
    private Callable<V> callable;         // 真正业务
    private Object outcome;               // 结果或异常
    private volatile Thread runner;       // 当前执行线程
    private volatile WaitNode waiters;    // Treiber 栈，等待线程
}

三、7 种状态机（状态压缩在一个 int）

状态	值	含义
NEW	0	新建
COMPLETING	1	结果已计算，正在赋值
NORMAL	2	正常结束
EXCEPTIONAL	3	异常结束
CANCELLED	4	被取消
INTERRUPTING	5	中断中
INTERRUPTED	6	已中断

状态流转图 NEW → COMPLETING → NORMAL NEW → COMPLETING → EXCEPTIONAL NEW → CANCELLED NEW → INTERRUPTING → INTERRUPTED

四、run() 方法——任务真正入口

public void run() {
    if (state != NEW || !RUNNER.compareAndSet(this, null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();          // 业务逻辑
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);           // 异常分支
            }
            if (ran) set(result);           // 正常分支
        }
    } finally {
        runner = null;
        int s = state;
        if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s);
    }
}

要点

通过 CAS 保证仅一个线程能进入执行。
结果赋值前先 COMPLETING，再写 outcome，二次 volatile 写保证可见性。
异常同样包装进 outcome，get() 时抛出 ExecutionException。

五、get() 与等待队列——Treiber 栈无锁实现

private volatile WaitNode waiters;   // 栈顶

static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
}

等待逻辑

状态 ≤ COMPLETING 时，线程包装成 WaitNode 压栈；
调用 LockSupport.park(this) 阻塞；
finishCompletion() 时弹栈并 unpark 唤醒。

优点 无锁、单生产者（runner）多消费者（get 线程），极高并发下仍保持弹性。

六、cancel(boolean mayInterruptIfRunning)

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW && STATE.compareAndSet(this, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    if (mayInterruptIfRunning) {
        Thread t = runner;
        if (t != null) t.interrupt();
    }
    STATE.setRelease(this, INTERRUPTED);
    finishCompletion();
    return true;
}

语义

任务未启动：直接标记为 CANCELLED；
已启动且参数为 true：中断执行线程；
任何取消都会唤醒所有等待 get() 的线程，抛出 CancellationException。

七、高并发缓存实战：FutureTask 解决“缓存击穿”

场景：热点 key 过期，大量线程同时穿透到 DB。

方案：单线程加载 + 其它线程阻塞等待结果

private final ConcurrentHashMap<Integer, FutureTask<BigDecimal>> cache =
        new ConcurrentHashMap<>();

public BigDecimal getPrice(Integer productId) throws Exception {
    while (true) {
        FutureTask<BigDecimal> f = cache.get(productId);
        if (f == null) {                                    // 1. 第一次检查
            Callable<BigDecimal> eval = () -> fetchFromDB(productId);
            FutureTask<BigDecimal> ft = new FutureTask<>(eval);
            f = cache.putIfAbsent(productId, ft);           // 2. CAS 放入
            if (f == null) {                                // 3. 抢占成功
                f = ft;
                ft.run();                                   // 4. 当前线程执行
            }
        }
        try {
            return f.get();                                 // 5. 阻塞等待结果
        } catch (CancellationException ignore) {
            cache.remove(productId, f);                     // 6. 被其它线程取消，重试
        }
    }
}

亮点

putIfAbsent 保证仅一个线程负责查库；
f.get() 让其余线程阻塞等待，不会击穿；
异常或取消时自动清理，防止脏数据。

八、与 CompletableFuture 的对比

维度	FutureTask	CompletableFuture
组合能力	❌ 不支持回调	✅ thenApply/Compose/Combine
手动完成	❌ 仅内部 set	✅ complete/obtrude
并发性能	极高，无锁	略低，依赖 CAS + Treiber
使用场景	轻量一次性任务	流式、事件驱动、复杂编排

结论： 如果只是“单线程加载 + 阻塞等待”，FutureTask 是最轻量、零依赖的方案；需要流水线式组合，再考虑 CompletableFuture。

九、性能压测数据

测试场景：32 线程，各自对 1000 个 key 调用 getPrice()，DB 模拟 200 ms 延迟。

实现	平均 RT (ms)	DB 查询次数	说明
无锁直接查	200	32 000	缓存击穿
synchronized 块	200 ~ 9 000	1 000	串行化，首线程慢则全体阻塞
FutureTask 版	200 ~ 210	1 000	单线程加载，其余等待