首页 技术 正文
技术 2022年11月17日
0 收藏 996 点赞 3,207 浏览 7542 个字

本文主要介绍FutureTask中的核心方法,如果有错误,欢迎大家指出!

首先我们看一下在java中FutureTask的组织关系

我们看一下FutureTask中关键的成员变量以及其构造方法

//表示当前task状态
private volatile int state;
//当前任务尚未执行
private static final int NEW = 0;
//当前任务正在结束,稍微完全结束,一种临界状态
private static final int COMPLETING = 1;
//当前任务正常结束
private static final int NORMAL = 2;
//当前任务执行过程中发生了异常。 内部封装的 callable.run() 向上抛出异常了
private static final int EXCEPTIONAL = 3;
//当前任务被取消
private static final int CANCELLED = 4;
//当前任务中断中..
private static final int INTERRUPTING = 5;
//当前任务已中断
private static final int INTERRUPTED = 6;

//submit(runnable/callable)   runnable 使用 装饰者模式 伪装成 Callable了。
private Callable<V> callable;
//正常情况下:任务正常执行结束,outcome保存执行结果。 callable 返回值。
//非正常情况:callable向上抛出异常,outcome保存异常
private Object outcome;
//当前任务被线程执行期间,保存当前执行任务的线程对象引用。
private volatile Thread runner;
/** Treiber stack of waiting threads */
//因为会有很多线程去get当前任务的结果,所以 这里使用了一种数据结构 stack
private volatile WaitNode waiters;

/**
* 如果传进来的是一个Callable接口,callable就是程序员自己实现的业务类
*/
public FutureTask(Callable<V> callable) {
//如果传入的callable为null,抛出空指针异常
if (callable == null)
throw new NullPointerException();
this.callable = callable;
//设置当前任务状态为 NEW
this.state = NEW; // ensure visibility of callable
}
/**
* 如果传入的是一个Runnable接口
* 使用装饰者模式将runnable转换为了 callable接口,
* 外部线程通过get获取当前任务执行结果时,结果可能为 null 也可能为 传进来的值。
*/
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
/**
* 将一个Runnable接口封装为Callable接口,
*/
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}

FutureTask中的Run方法


public void run() {
//条件一:state != NEW 条件成立,说明当前task已经被执行过了 或者 被cancel 了,总之非NEW状态的任务,线程就不处理了。
//条件二:!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread())
// 条件成立:cas失败,当前任务被其它线程抢占了...
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
//执行到这里,当前task一定是 NEW 状态
// 而且 当前线程也抢占TASK成功(runner中存放着当前的线程对象)!
try {
//自己封装的callable或者runnable
Callable<V> c = callable;
//条件一:c != null 防止空指针异常
//条件二:state == NEW 防止外部线程 cancel掉当前任务。
if (c != null && state == NEW) {
V result;
//true 表示callable.run 代码块执行成功 未抛出异常
//false 表示callable.run 代码块执行失败 抛出异常
boolean ran;
try {
//调用程序员自己实现的callable 或者 装饰后的runnable
result = c.call();
//c.call未抛出任何异常,ran会设置为true 代码块执行成功
ran = true;
} catch (Throwable ex) {
//说明程序员自己写的逻辑块有bug了。
result = null;
ran = false;
//说明是从异常出来的,将异常设置在结果中
setException(ex);
}
//说明当前c.call正常执行结束了。
//set就是设置结果到outcome
if (ran)
set(result);
}
} finally {
//在执行过程中runner的值一定不为null
//防止并发调用run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
//看当前状态是否是中断状态,如果是处理中断操作
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}

//正常结束
protected void set(V v) {
//使用CAS方式设置当前任务状态为 完成中..
//有没有可能失败呢? 外部线程等不及了,直接在set执行CAS之前 将 task取消了。 很小概率事件。
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
//将结果赋值给 outcome之后,马上会将当前任务状态修改为 NORMAL 正常结束状态。
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}

//异常结束
protected void setException(Throwable t) {
//使用CAS方式设置当前任务状态为 完成中..
//有没有可能失败呢? 外部线程等不及了,直接在set执行CAS之前 将 task取消了。 很小概率事件。
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//将异常封装在结果中返回
outcome = t;
//将当前任务的状态 修改为 EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
//说明现在已经结束了
finishCompletion();
}
}

/**
* 结束以后要负责唤醒等待栈中真在等待获取结果的线程
*/
private void finishCompletion() {
// assert state > COMPLETING;
//q指向waiters 链表的头结点。
for (WaitNode q; (q = waiters) != null;) {
//使用cas设置 waiters 为 null 是因为怕
//外部线程使用 cancel 取消当前任务 也会触发finishCompletion方法。 小概率事件。
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
//获取当前node节点封装的 thread
Thread t = q.thread;
if (t != null) {
q.thread = null;
//唤醒当前节点对应的线程
LockSupport.unpark(t);
}
//next 当前节点的下一个节点
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
/**
* 处理有可能的取消中断
*/
private void handlePossibleCancellationInterrupt(int s) {
// It is possible for our interrupter to stall before getting a
// chance to interrupt us. Let's spin-wait patiently.
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt

// assert state == INTERRUPTED;

// We want to clear any interrupt we may have received from
// cancel(true). However, it is permissible to use interrupts
// as an independent mechanism for a task to communicate with
// its caller, and there is no way to clear only the
// cancellation interrupt.
//
// Thread.interrupted();
}
 

FutureTask中Get方法

public V get() throws InterruptedException, ExecutionException {
int s = state;
//条件成立:未执行、正在执行。 调用get的外部线程会被阻塞在get方法上。
//核心方法
// private static final int NEW = 0;
// private static final int COMPLETING = 1;

if (s <= COMPLETING)
s = awaitDone(false, 0L);

return report(s);
}

//同样的作用设置了超时时间
public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
//如果当前状态小于等于COMPLETING并且调用了awaitDone方法以后还是小于等于COMPLETING,说明当前是超时以后返回的
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}

/**
* 等待任务完成然后获取结果
*/
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
//timed等于false不带超时
//timed等于true带超时
final long deadline = timed ? System.nanoTime() + nanos : 0L;
//引用当前线程 封装成 WaitNode 对象
WaitNode q = null;
//表示当前线程waitNode对象 有没有压栈
boolean queued = false;
for (;;) {
//条件成立:说明当前线程唤醒 是被其它线程使用中断这种方式喊醒的。interrupted()
//返回true 后会将 Thread的中断标记重置回false.
if (Thread.interrupted()) {
//当前线程node出队
removeWaiter(q);
//get方法会得动中断异常
throw new InterruptedException();
}
//假设是被其他线程正常唤醒或者超时自动返回
int s = state;
//条件成立,说明当前任务已经有结果了
if (s > COMPLETING) {
//条件成立,说明已经为当前线程创建过node了
if (q != null)
q.thread = null;
//返回当前状态
return s;
}
//条件成立:说明当前任务接近完成状态...这里让当前线程再释放cpu ,进行下一次抢占cpu。
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//创建WaitNode对象
else if (q == null)
q = new WaitNode();
//条件成立:第二次自旋,当前线程已经创建 WaitNode对象了,但是node对象还未入栈
else if (!queued)
//将当前节点入栈
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
//第三次自旋,会到这里。
//如果是带有超时时间的,在超时时间里面变为 WAITING状态,相当于休眠了.
//除非有其它线程将你唤醒 或者 将当前线程 中断,或者到达超时时间。
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
//当前get操作的线程就会被park了。 线程状态会变为 WAITING状态,相当于休眠了..
//除非有其它线程将你唤醒 或者 将当前线程 中断。
LockSupport.park(this);
}
}
 
private void removeWaiter(WaitNode node) {
if (node != null) {
node.thread = null;
retry:
for (;;) { // restart on removeWaiter race
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
s = q.next;
if (q.thread != null)
pred = q;
else if (pred != null) {
pred.next = s;
if (pred.thread == null) // check for race
continue retry;
}
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break;
}
}
}

/**
* 返回结果或引发已完成任务的异常。
*/
private V report(int s) throws ExecutionException {
//正常情况下,outcome 保存的是callable运行结束的结果
//非正常,保存的是 callable 抛出的异常。
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);

FutureTask中Cancel方法

/**
* mayInterruptIfRunning为true,该task正在执行,这个task应该被中断;
* mayInterruptIfRunning为false,该task正在执行,这个task将会继续执行到完成。
*/
public boolean cancel(boolean mayInterruptIfRunning) {
//条件一:state == NEW 成立 表示当前任务处于运行中 或者 处于线程池 任务队列中..
//条件二:UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))
// 条件成立:说明修改状态成功,可以去执行下面逻辑了,否则 返回false 表示cancel失败。
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
//执行当前FutureTask 的线程,有可能现在是null,是null 的情况是: 当前任务在 队列中,还没有线程获取到它呢。。
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
相关推荐
python开发_常用的python模块及安装方法
adodb:我们领导推荐的数据库连接组件bsddb3:BerkeleyDB的连接组件Cheetah-1.0:我比较喜欢这个版本的cheeta…
日期:2022-11-24 点赞:878 阅读:9,103
Educational Codeforces Round 11 C. Hard Process 二分
C. Hard Process题目连接:http://www.codeforces.com/contest/660/problem/CDes…
日期:2022-11-24 点赞:807 阅读:5,579
下载Ubuntn 17.04 内核源代码
zengkefu@server1:/usr/src$ uname -aLinux server1 4.10.0-19-generic #21…
日期:2022-11-24 点赞:569 阅读:6,427
可用Active Desktop Calendar V7.86 注册码序列号
可用Active Desktop Calendar V7.86 注册码序列号Name: www.greendown.cn Code: &nb…
日期:2022-11-24 点赞:733 阅读:6,199
Android调用系统相机、自定义相机、处理大图片
Android调用系统相机和自定义相机实例本博文主要是介绍了android上使用相机进行拍照并显示的两种方式,并且由于涉及到要把拍到的照片显…
日期:2022-11-24 点赞:512 阅读:7,834
Struts的使用
一、Struts2的获取  Struts的官方网站为:http://struts.apache.org/  下载完Struts2的jar包,…
日期:2022-11-24 点赞:671 阅读:4,917