100行手写FutureTask

灬沙师弟

发布于 2025-11-12 13:44:37

1250

一、前言

书接上文，先来看看需求

功能	实现方式
仅单线程执行任务	CAS 抢占 runner
阻塞获取结果	Treiber 栈 + LockSupport
支持取消/中断	状态机 + interrupt
异常透传	统一 outcome 字段
100 行以内	剔除了超时、批量栈清理等锦上添花逻辑

二、状态机（4 种简化状态）

private staticfinalint NEW          = 0;  // 新建
privatestaticfinalint COMPLETING   = 1;  // 结果已出，正在写入
privatestaticfinalint NORMAL       = 2;  // 正常结束
privatestaticfinalint EXCEPTIONAL  = 3;  // 异常结束
privatestaticfinalint CANCELLED    = 4;  // 被取消
privatestaticfinalint INTERRUPTING = 5;  // 中断中
privatestaticfinalint INTERRUPTED  = 6;  // 已中断

注：为了篇幅，超时/等待节点清理等状态略去。

三、核心字段一览

public class MiniFutureTask<V> implements Runnable, Future<V> {
    private volatile int state = NEW;           // 状态
    private Callable<V> callable;               // 业务
    private Object outcome;                     // 结果或异常
    private volatile Thread runner;             // 执行线程
    private volatile WaitNode waiters;          // Treiber 栈顶
}

四、Treiber 栈节点（无锁等待队列）

static final class WaitNode {
    volatile Thread thread = Thread.currentThread();
    volatile WaitNode next;
}

五、run() —— 单线程 CAS 执行

public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, RUNNER_OFFSET, 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);
    }
}

六、set / setException —— 二次 volatile 写保证可见性

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, STATE_OFFSET, NEW, COMPLETING)) {
        outcome = v;
        STATE.setRelease(this, NORMAL);   // 顺序写
        finishCompletion();
    }
}

protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, STATE_OFFSET, NEW, COMPLETING)) {
        outcome = t;
        STATE.setRelease(this, EXCEPTIONAL);
        finishCompletion();
    }
}

七、get() —— 阻塞等待

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    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;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        if (Thread.interrupted()) {
            removeWaiter(q);
            thrownew InterruptedException();
        }
        int s = state;
        if (s > COMPLETING) {
            if (q != null) q.thread = null;
            return s;
        }
        elseif (s == COMPLETING) Thread.yield();
        elseif (q == null) q = new WaitNode();
        elseif (!queued) {
            q.next = waiters;
            queued = UNSAFE.compareAndSwapObject(this, WAITERS_OFFSET, waiters, q);
        }
        elseif (timed) {
            long nanosLeft = deadline - System.nanoTime();
            if (nanosLeft <= 0L) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanosLeft);
        }
        else LockSupport.park(this);
    }
}

八、finishCompletion() —— 唤醒全部等待线程

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, WAITERS_OFFSET, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null) break;
                q.next = null; // 方便 GC
                q = next;
            }
            break;
        }
    }
}

九、cancel() —— 中断或取消

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, STATE_OFFSET, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        returnfalse;
    try {
        if (mayInterruptIfRunning) {
            Thread t = runner;
            if (t != null) t.interrupt();
        }
    } finally {
        STATE.setRelease(this, INTERRUPTED);
        finishCompletion();
    }
    returntrue;
}

十、report() —— 结果或异常

@SuppressWarnings("unchecked")
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);
}