02.FutureTask源码
作者：互联网
FutureTask源码
看前须知
- 本文章所有内容是老师源码特训班的内容有兴趣的可以了解一波：https://space.bilibili.com/457326371?from=search&seid=882812707426049189
- 本章需要一些前置知识：线程池的基本使用，FutureTask的基本使用，CAS，Unsafe，volatile，线程中断
- 上述前置知识很多都可以在老师的公开课中找到
- 基于JDK8版本进行源码分析
重要属性说明
// 表示当前任务状态
private volatile int state;
// state的一个常量：当前任务尚未执行
private static final int NEW = 0;
// state的一个常量：当前任务正在结束，尚未完全结束，一种临界状态
private static final int COMPLETING = 1;
// state的一个常量：当前任务正常结束
private static final int NORMAL = 2;
// state的一个常量：当前任务执行过程中出现了异常，内部封装的Callable的run方法向上抛出了异常
private static final int EXCEPTIONAL = 3;
// state的一个常量：当前任务呗取消
private static final int CANCELLED = 4;
// state的一个常量：当前任务中断中
private static final int INTERRUPTING = 5;
// state的一个常量：当前任务已中断
private static final int INTERRUPTED = 6;
// 线程池的submit方法提交的Runnable或Callable对象，Runnable是通过适配器模式变成Callable类型的
private Callable<V> callable;
/*
* 1. 正常情况：任务正常执行结束，保存执行结果，Callable的返回值
* 2. 异常情况：Callable向上抛出异常，保存异常
*/
private Object outcome;
// 当前任务被线程执行期间，保存当前执行任务的线程引用
private volatile Thread runner;
// 头插队列，当有多个线程等待结果的时候，进入这个队列挂起
private volatile WaitNode waiters;
静态内部类WaitNode
1.源代码
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
2.属性说明
- thread： 当前节点对应的线程
- next： 下一个节点
非核心方法解析
构造方法解析
1.源代码
public FutureTask(Callable<V> callable) {
// 判空判断
if (callable == null)
throw new NullPointerException();
// 初始化成员对象
this.callable = callable;
// 初始化任务状态为NEW
this.state = NEW;
}
2.方法作用
初始化属性callable与state
set方法解析
1.源代码
protected void set(V v) {
// 使用CAS设置状态从NEW->COMPLETING，如果失败说明任务被cancel了
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 设置outcome为callable返回的值
outcome = v;
// 设置状态从COMPLETING->NORMAL
UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
finishCompletion();
}
}
2.方法作用
- 设置状态NEW->COMPLETING
- 设置outcome为callable返回的值
- 设置状态COMPLETING->NORMAL
setException方法解析
1.源代码
protected void setException(Throwable t) {
// 使用CAS设置状态NEW->COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 让outcome保存异常
outcome = t;
// 设置状态为EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
finishCompletion();
}
}
2.方法作用
- 设置状态NEW->COMPLETING
- 设置outcome为异常
- 设置状态COMPLETING->EXCEPTIONAL
核心方法run解析
1.源代码
public void run() {
/*
* 1. state != NEW——条件成立说明当前任务已经被执行过了
* 2. !UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())——使用CAS设置runner属性条件成立说明有竞争
*
*/
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
// 引用属性callable
Callable<V> c = callable;
/*
* 1. c != null——基本都成立，主要为了防止空指针
* 2. state == NEW——为了确保这个任务没有被 cancel掉
*/
if (c != null && state == NEW) {
// 结果引用
V result;
/*
* 1. true——表示callable.run执行成功
* 2. false——表示callable.run 抛出了异常执行失败
*/
boolean ran;
try {
// 调用程序员实现的callable代码
result = c.call();
// 设置执行成功，因为如果出现了异常走不到这里
ran = true;
} catch (Throwable ex) {
// 返回结果是null，因为执行出现了异常
result = null;
// 执行失败
ran = false;
setException(ex);
}
// 如果执行成功
if (ran)
// 设置结果到outcome属性
set(result);
}
} finally {
// 逻辑上类似于释放锁，也是清空执行当前任务的线程
runner = null;
//
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
2.方法作用
线程池执行当前任务的入口方法
3.变量说明
- c： 引用属性callable
- result： 保存任务执行完的返回值或异常
- ran： 任务是否正常执行完毕
- s： 引用属性state
核心方法get解析
1.源代码
public V get() throws InterruptedException, ExecutionException {
// 引用属性state
int s = state;
// 如果任务还没执行完或还未执行
if (s <= COMPLETING)
// 调用当前方法的线程去阻塞
s = awaitDone(false, 0L);
return report(s);
}
2.方法作用
阻塞获取任务执行结果
3.变量说明
- s： 引用属性state
核心方法awaitDone解析
1.源代码
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 如果不带超时功能那么就是0
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 当前线程封装的WaitNode对象
WaitNode q = null;
// 是否入队
boolean queued = false;
// 自选
for (;;) {
// 检查中断状态，如果中断过
if (Thread.interrupted()) {
// 从队列中移除当前节点
removeWaiter(q);
// 抛出中断异常
throw new InterruptedException();
}
// 引用属性state
int s = state;
// 条件成立：说明当前任务已经有结果了
if (s > COMPLETING) {
// 条件成立：说明已经为当前线程创建过node了，此时需要将Node的thread指向空
if (q != null)
q.thread = null;
// 返回状态
return s;
}
// 条件成立：说明任务快完成了，调用yield方法让出一些CPU
else if (s == COMPLETING)
Thread.yield();
// 条件成立：说明是第一次进入自选
else if (q == null)
// 创建WaitNode
q = new WaitNode();
// 条件成立：第二次自选会来到这里，表示当前线程创建了WaitNode对象但是还没有入队
else if (!queued)
// 当前节点的next指向waiter，然后使用CAS修改waiters为当前节点，总结来说就是入队
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);
}
}
2.方法作用
阻塞调用线程
3.方法入参说明
- timed： 是否要支持超时
- nanos： 挂起毫秒数
4.方法变量说明
- deadline： 用来计算挂起毫秒数
- q： 当前线程封装的WaitNode对象
- queued： 是否入队
- s： 引用属性state
核心方法removeWaiter解析
1.源代码
private void removeWaiter(WaitNode node) {
// 判断当前节点是不是null，如果不是说明当前节点根本没入队不用出对了
if (node != null) {
// 属性为null
node.thread = null;
// 外部循环的命名
retry:
// 自选
for (;;) {
/*
* 作用：遍历队列
* 1. pred——上一个节点
* 2. q——遍历时的节点，从头结点开始
*/
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
// 获取当前结点的下一个节点
s = q.next;
// 如果不是要被出队的节点
if (q.thread != null)
// 保存当前节点为pred
pred = q;
// 如果出队的节点不是头结点
else if (pred != null) {
// 出队节点的上一个节点的next指向s
pred.next = s;
if (pred.thread == null)
continue retry;
}
// 如果出队的节点是头结点，使用CAS让waiters指向头结点的下一个节点
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break;
}
}
}
2.方法作用
出队操作，如果有其他线程因为任务完成或抛出异常导致p.thread=null，也会帮助它们出队
3.方法入参说明
- node： 要出对的节点
4.方法变量说明
- pred： 遍历队列时当前节点的上一个节点
- q： 遍历队列时的当前节点
- s： 当前结点的下一个节点
核心方法report解析
1.源代码
private V report(int s) throws ExecutionException {
// 引用属性outcome
Object x = outcome;
// 条件成立：当前任务正常执行结束
if (s == NORMAL)
// 返回结果
return (V)x;
// 条件成立：任务被取消
if (s >= CANCELLED)
throw new CancellationException();
// 条件成立：任务执行有bug
throw new ExecutionException((Throwable)x);
}
2.方法作用
根据状态返回结果
3.方法入参说明
- s： 任务状态
4.方法变量说明
- x： 引用属性outcome
核心方法finishCompletion解析
1.源代码
private void finishCompletion() {
// q引用属性waiters
for (WaitNode q; (q = waiters) != null;) {
// 使用CAS设置 waiters为null
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
// 自选
for (;;) {
// 拿到遍历的当前节点的thread属性
Thread t = q.thread;
// 条件成立：说明当前线程不是null
if (t != null) {
// 唤醒
q.thread = null;
LockSupport.unpark(t);
}
// 获得当前节点的下一个节点
WaitNode next = q.next;
// 条件成立：当前节点是最后一个节点
if (next == null)
break;
// help GC
q.next = null;
// 推进遍历
q = next;
}
break;
}
}
// 扩展方法，里面啥都没有
done();
// help GC
callable = null;
}
2.方法作用
走到这里说明任务已经结束，这个方法负责唤醒等待结果所有线程，并清空队列
3.方法变量说明
- q： 引用遍历头插队列的每一个节点，刚开始是头节点
核心方法cancel解析
1.源代码
public boolean cancel(boolean mayInterruptIfRunning) {
/*
* 如果状态不是NEW或者CAS State失败都会直接返回
* 1. state == NEW——如果状态是NEW
* 2. UNSAFE.compareAndSwapInt(this, stateOffset, NEW,mayInterruptIfRunning ? INTERRUPTING : CANCELLED)——根据入参把state改为INTERRUPTING或CANCELLED
*/
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try {
// 如果是true说明需要给运行中的任务发一个中断信号
if (mayInterruptIfRunning) {
try {
Thread t = runner;
// 条件成立:说明当前任务中在运行
if (t != null)
// 给runner线程一个中断
t.interrupt();
} finally {
// 修改状态为INTERRUPTED
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
// 唤醒所有get方法阻塞的线程
finishCompletion();
}
return true;
}
2.方法作用
取消任务或中断任务 并唤醒所有get方法阻塞的线程
3.方法入参说明
- mayInterruptIfRunning： 如果是true会把正在运行的任务发送一个中断信号
4.方法变量说明
- t： 引用属性runner
标签：02,当前任务,state,源码,线程,FutureTask,NEW,null,节点 来源： https://blog.csdn.net/weixin_43891622/article/details/119347916