星星博客 »  > 

JAVASE-3 多线程体系5-FutureTask源码分析

引言

在研究线程池源码的时候,遇到了FutureTask,对这个类较为感兴趣。且线程池与FutureTask交互的过程不可避免地需要与Runnable与Callable打交道,这个过程中还涉及了适配器设计模式。这部分内容对提高代码设计能力有借鉴意义,也有助于理解线程池的实现原理,有必要总结一下。
本打算在线程池内容中一起介绍,但FutureTask这部分怎么排版都与线程池的其他主题内容显得格格不入。因此先阻塞写线程池blog的进程,另外开启一个进程,先总结FutureTask。之后,在线程池blog一文中直接引用本文,这样就不会影响线程池的整体路线。

1.Runnabel接口与Callable接口

先看一下定义这两个接口的代码

public interface Callable<V> {
    V call() throws Exception;
}
public interface Runnable {
    void run();
}

从定义可以看出:
1.Callable的核心是call()方法,而Runnable的核心是run()方法。
2.call()是可以返回值以及抛出异常的,run()没有返回值也不能抛出异常(子类不能抛出比父类范围大的异常)
另外,这两个方法都没有形参,作用就是用来封装一段代表程序逻辑的代码块(以下用任务来表示)。

2.Future类

在jdk1.4及以前版本中,线程只能通过实现Runnabel接口或继承Thread类来实现。Runnable的run()没有返回值,需要依靠共享变量来实现线程之间的通讯,使用较为不便,且无法抛出异常。为了弥补这些缺点,在jdk1.5中引入了Callable和Future类,使线程具有返回值的能力。
在很多文章里,Future被定义为“可能还没被执行完成的结果”。

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeOut, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

Future类中比较常用的是get()get(long timeOut, TimeUnit unit)
get()如果逻辑已经执行完成,直接返回结果;否则,阻塞调用get()的线程, 直到逻辑执行完成;
get(long timeOut, TimeUnit unit) 最多等待timeOut时间,如果还没有返回,直接抛出TimeoutException异常。
另外,还提供了3个抽象方法:cancel(mayInterruptIfRunning)用来取消任务;isDone()用来查询任务是否完成;isCancelled()查询任务是否取消,将在介绍FutureTask源码时详细进行介绍。

3.FutureTask类

3.1 FutureTask类继承体系

研究一个对象时,一般先跳出来看一眼全局,再钻进去研究细节。不妨先从类的继承体系的角度来研究一下FutureTask。
在这里插入图片描述
RunnabelFuture整合了Runnabel和Future,并没有引入新的方法;
FutureTask是RunnabelFuture的实现类,实现了Runnabel和Future的所有接口。
从命名也可以看出FutureTask是一个具有Future能力的Runnable,本质上是一个任务。

3.2 FutureTask构造函数

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;
}
public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;
}

两个构造函数分别接受Callable和Runnable类型的任务:
如果是Callable 类型,直接赋值给FutureTask的callable 属性;如果是Runnable 类型,需要适配为Callable类型,再进行赋值。
适配的实现是通过引入了一个中间层,实现Callable接口,并在call()方法中直接调用run()
Runnable -> Callable的适配过程传入了一个result对象,但是Runnable 没有处理返回结果的能力,所以result变量本质上没什么作用,仅仅起适配作用。
这里还有一点需要注意的是:新创建的FutureTask,state的属性值都被设置成了NEW状态。

3.3 FutureTask属性

// 待执行的任务
private Callable<V> callable; 
// 返回的结果,也可能代表异常类
private Object outcome; 
// FutureTask的状态
private volatile int state;
// 执行callable这个任务线程
private volatile Thread runner; 
// 阻塞在获取callable执行结果上的线程
private volatile WaitNode waiters;

callableoutcome表示任务以及任务的返回结果。
runner指向一个线程,用来运行callable任务。

在介绍waiters之前,先介绍一下WaitNode 类:

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

以单链表方式构成一个等待队列,存放的是因为获取callable任务运行结果而阻塞的线程。
在这里插入图片描述
需要注意FutureTask中涉及两类线程:

(1) 运行callable的线程,就是上面的runner实例变量;
(2) 获取callable运行结果而阻塞的线程(调用future.get()),可以有多个,存放在等待队列waiters中。因为waiters指向的是链表的顶部节点,所以叫队列不太准确,应该是栈(本质上是个Treiber栈)

3.4 FutureTask状态

研究JUC源码的过程中,发现大部分类都维持一个state属性,实现逻辑也围绕 查询state、比较state、设置state,修改时基本都是使用cas方式;FutureTask也不例外。

    /**
     * Possible state transitions:
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
     */
private volatile int state;
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;

从代码注释也可以看出状态切换有4条路线:
在这里插入图片描述
FutureTask的任务创建之初处于NEW状态,call()方法执行完后,由NEW转为COMPLETING
进行结果值的设置:如果正常返回,进入NORMAL状态;如果遇到异常,进入EXCEPTIONAL状态;
处于NEW状态的任务,如果被取消了,调用了cancel()会走下面两条路线:根据传入cancel()的参数来确定,下一节会详细介绍。
另外,由于中间态只是用来处理结果的,所以时间极其短暂

3.5 FutureTask方法

run()

既然FutureTask本质是一个Runnable, 代码解析首先从run()开始。

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

上面的代码比较复杂,先看一个简单版本(省略了校验、异常处理),大致了解主体路线。

public void run() {
	Callable<V> c = callable;
	V result;
	boolean ran;
	try {
		result = c.call();
		ran = true;
	} catch (Throwable ex) {
		result = null;
		ran = false;
		setException(ex);
	}
	if (ran) {
		set(result);
	}
}

比较直观:在run()方法中直接调用callable实例对象的call()方法,如果正常执行ran为true,会执行set(result)设置返回结果;如果抛出异常,执行setException(ex)设置返回异常。
再看一下省略的判断:

    if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
        return;

要求FutureTask必须处于NEW状态下,即表示callable尚未被调用;要求运行这个run()方法时,runner属性是空,并用runner记录当前线程。异常处理逻辑比较清楚,且不是本文讨论的重点,此处省略。
callable的call()方法运行完成后,会调用set()或者setException()来处理结果。

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
        finishCompletion();
    }
}

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

set()方法中:
1.将state设置为COMPLETING状态;
2.将结果赋值给outcome;
3.将state设置为NORMAL状态;
4.调用finishCompletion()方法;

setException()方法中:
1.先将state设置为COMPLETING状态;
2.将异常结果赋值给outcome;
3.将state设置为EXCEPTIONAL状态;
4.调用finishCompletion()方法;
这里涉及到FutureTask的四个状态和两条状态切换路线:
在这里插入图片描述
结合代码可以看出COMPLETING处于一个极其低的时间段;切处于COMPLETING时表示call()方法已经执行完成,任务已被执行,处于处理结果的过程。
然后再看看finishCompletion()方法:

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, 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;
                q = next;
            }
            break;
        }
    }

    done();

    callable = null;        // to reduce footprint
}

由于done()是提供的钩子方法—留给自类扩展实现特定后处理功能,在FutureTask里为空,此处省略。
简化为:

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, 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;
                q = next;
            }
            break;
        }
    }
}

WaitNode q; (q = waiters) != null;WaitNode next = q.next;可以看出是一个遍历的过程;
内部有一个自旋逻辑,一般看自旋逻辑代码,先看退出点(可以退出自旋逻辑):

 if (next == null)
        break;

结合UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)LockSupport.unpark(t) 可以知道:这段代码是为了清空等待队列并唤醒这些线程。
等待执行结果的线程因为调用future.get()等阻塞了,当callable执行完获得结果后,需要主动唤醒这些阻塞的线程。
介绍完run(),我们再看看上文提到的Future对外提供的其他api。

get() 和get(long timeout, TimeUnit unit)

以下是get()源码:

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

只有一处判断逻辑:如果state处于NEW或者COMPLETING,调用awaitDone方法(继续等一下)。否则调用report方法获取结果。这个过程涉及了两个方法:awaitDone方法和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内部逻辑比较简单,根据state的完成状态来返回结果:
如果stateNORMALoutcome表示的是call()正常执行返回的结果,直接放回outcome;
如果stateCANCELLED, 表示任务被取消或被中断,直接返回CancellationException异常;
如果stateEXCEPTIONAL,表示任务执行遇到异常,此时outcome表示的是异常类,使用outcome构造一个ExecutionException异常返回。
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) {
			Thread.yield();
		} else if (q == null) {
			q = new WaitNode();
		} else if (!queued) {
			queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q);
		} else if (timed) {
			nanos = deadline - System.nanoTime();
			if (nanos <= 0L) {
				removeWaiter(q);
				return state;
			}
			LockSupport.parkNanos(this, nanos);
		} else {
			LockSupport.park(this);
		}
	}
}

主体实现都包含在自旋块中:

if (Thread.interrupted()) {
	removeWaiter(q);
	throw new InterruptedException();
}

先判断该线程是否被中断过,如果已被中断,从等待队列中删除该线程所在节点,并直接抛出中断异常。因为future.get()是一个阻塞方法,响应线程中断。
后续逻辑完全围绕state的值展开,根据state的值,存在6个if分支:

int s = state;
if (s > COMPLETING) { //if_1
	if (q != null){
		q.thread = null;
	}
	return s;
} else if (s == COMPLETING) { //if_2
	Thread.yield();
} else if (q == null) { //if_3
	q = new WaitNode();
} else if (!queued) { //if_4
	queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q);
} else if (timed) { //if_5
	nanos = deadline - System.nanoTime();
	if (nanos <= 0L) {
		removeWaiter(q);
		return state;
	}
	LockSupport.parkNanos(this, nanos);
} else { //if_6
	LockSupport.park(this);
}

if_1 表示任务已经执行完成且结果以及准备好,直接返回state;
if_2 表示任务以及执行完成,但是正常处理返回结果,调用Thread.yeild()稍等一下;
如果if_1if_2都没有进入,表示任务还在执行中,即任务处于COMPLETING状态;
if_3 表示新建一个WaitNode节点来存放当前线程,准备放入等待队列;
if_4 能进if_4表示if_3false(q这个节点已经创建好了),但是还没有入队,此时将q压入等待队列栈顶;
if_5 get()方法不涉及(timed是false);
if_6 其他情况,线程调用LockSupport.park(this);阻塞——等待任务执行完成,被唤醒;

对于get(long timeout, TimeUnit unit)
awaitDone方法中:

else if (timed) {
    nanos = deadline - System.nanoTime();
    if (nanos <= 0L) {
        removeWaiter(q);
        return state;
    }
    LockSupport.parkNanos(this, nanos);
}

如果超时了,就直接返回state,没有超时就继续等待,等待nanos后苏醒。

    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);
    }

由于超时,awaitDone(true, unit.toNanos(timeout)))返回时,任务并没有完成,此时的stateNEW,所以直接抛出TimeoutException()。其他逻辑与get()完全一致,不再赘述。

cancel(boolean)

cancel方法有个boolean类型的传参,表示是否打断线程。

public boolean cancel(boolean mayInterruptIfRunning) {
	if (!(state == NEW &&
		  UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
			  mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) {
		return false;
	}
	try {
		if (mayInterruptIfRunning) {
			try {
				Thread t = runner;
				if (t != null) {
					t.interrupt();
				}
			} finally { // final state
				UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
			}
		}
	} finally {
		finishCompletion();
	}
	return true;
}

先看第一段判断:

	if (!(state == NEW &&
		  UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
			  mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) {
		return false;
	}

cancel只能作用在任务还处于NEW状态下。如果call()已经执行完成了,cancel()直接跳过。
当任务还处于New状态时,调用cancel()会根据参数mayInterruptIfRunning走不同的逻辑。

mayInterruptIfRunning = true;表示线程可以打断:
先将任务的状态设置为INTERRUPTING
然后中断线程:t.interrupt();
然后将任务的状态设置为INTERRUPTED;
唤醒所有阻塞在获取任务执行结果的线程;

mayInterruptIfRunning = false;表示线程不允许打断:
直接将任务设置为CANCELLED
唤醒所有阻塞在获取任务执行结果的线程;

这里涉及一下两条路线:其中INTERRUPTING时间段及其短暂。
在这里插入图片描述

isCancelled() 和 isDone()

public boolean isCancelled() {
    return state >= CANCELLED;
}

iscancelled()比较简单,判断如果state是CANCELLED,INTERRUPTING 或 INTERRUPTED时,返回true,即是否调用了cancelled()方法;否则返回false;

public boolean isDone() {
    return state != NEW;
}

isDone代表的意思是任务是否执行完成了,在之前介绍任务状态的时候,就知道:任务一旦不是NEW,就表示任务的call()方法已经执行完成了。

相关文章