FutureTask源码分析

Posted 2022-12-11 _houzhi

FutureTask是JDK1.6新增的内容，间接实现了Future, Runnable接口。前面已经介绍过了Future的get方法会阻塞在那，等待执行完成，获取结果，那么FutureTask是怎么实现这一功能的呢？

相关类简介

LockSupport

LockSupport是用来创建锁及其他同步类的基本线程阻塞元素，它的park和 unpark能够分别阻塞线程和解除线程阻塞。它提供了可以指定阻塞时长的park方法。park和unpark的基本接口为:

public static void park() 
    unsafe.park(false, 0L);

public static void unpark(Thread thread) 
    if (thread != null)
        unsafe.unpark(thread);

unpark需要指定对应的线程，而park是将当前线程阻塞。其实这里也提供阻塞线程新的方式，我们已经或者用忙等待，或者用中断睡眠的方式阻塞唤醒线程，这里可以通过LockSupport实现。FutureTask就是用这种方式实现的。下面看一下unsafe，它是一个Unsafe类

Unsafe

Java不能够直接访问操作系统底层，而是通过本地方法来访问。Unsafe提供了硬件级别的原子访问，主要提供一下功能：
1. 分配释放内存
2. 定位某个字段的内存位置
3. 挂起一个线程和恢复，更多的是通过LockSupport来访问。park和unpark
4. CAS操作，比较一个对象的某个位置的内存值是否与期望值一致，一致则更新对应值，此更新是不可中断的。主要方法是compareAndSwap*。

FutureTask源码

FutureTask主要是通过这两个类来实现的，调用get的时候如果异步线程还没有执行完，就是先park当前线程，如果超时则返回，这个时候如果异步线程还是没有执行完成，则返回失败。如果异步线程执行完成了，则会调用LockSupport的unpark方法，将主线程（与异步线程对应）解除阻塞，而主线程解除阻塞后就会获取执行的结果，get函数返回结果。下面看一下get方法的源码

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException 
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);

它实际实现wait的是awaitDone方法：

private int awaitDone(boolean timed, long nanos)
    throws InterruptedException 
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) 
        if (Thread.interrupted()) 
            removeWaiter(q);
            throw new InterruptedException();
        

        int s = state;
        if (s > COMPLETING) 
            if (q != null)
                q.thread = null;
            return s;
        
        else if (s == COMPLETING) // cannot time out yet, 已经完成了，但是状态还在COMPLETING，yield一下下次再判断。
            Thread.yield();
        else if (q == null)　//如果队列还是为创建一个等待节点
            q = new WaitNode();
        else if (!queued) //如果还没有加入waiters队列，则先将waiters赋值给q.next，然后判断waiters是否已经是q.next，如果是，则将q赋值给waiters(waitersOffset对应waiters在FutureTask中的偏移量)。关键在于此一系列操作能保证一致性。
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        else if (timed)  //如果是需要判断超时，则使用parkNanos
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) 
                removeWaiter(q);
                return state;
            
            LockSupport.parkNanos(this, nanos);
        
        else 
            LockSupport.park(this);
    

waitDone就是将当前线程加入等待队列(WaitNode有当前Thread的Thread变量)，然后用LockSupport将自己阻塞，等待超时或者被解除阻塞后，判断是否已经完成(state为>= COMPLETING)，如果未完成(state< COMPLETING)抛出超时异常，如果已完成则稍等或者直接返回结果。

下面看看异步线程如何唤醒当前线程，从run方法开始：

    public void run() 
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread())) //给runner赋值
            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)　//已经运行完则设置结果，前面一段是调用Callable运行需要做的操作
                    set(result);
            
         finally 
           ...
        
    

set方法：

protected void set(V v) 
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING))  //原子操作state
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();//通知线程
        
    

finishCompletion:

private void finishCompletion() 
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) 
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) //如果已经赋值给q ,则对waiter置null
            for (;;) 　//一个一个地unpark。
                Thread t = q.thread;
                if (t != null) 
                    q.thread = null;
                    LockSupport.unpark(t);
                
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            
            break;
        
    

    done(); //默认实现什么也不做

    callable = null;        // to reduce footprint

上面的代码就是遍历waiters,然后解除他们的阻塞。这样整个流程就算完了

总结

在整个FeatureTask中没有直接使用锁机制，而是通过LockSupport来阻塞线程，唤醒线程。对于多线程访问FeatureTask的waiters,state，都是采用Unsafe来操作，避免使用锁（毕竟锁其实很耗时），改为直接原子操作对应的变量。FeatureTask是一个非常好的Unsafe和LockSupport例子。