Error message here!

Hide Error message here!

忘记密码?

Error message here!

请输入正确邮箱

Hide Error message here!

密码丢失?请输入您的电子邮件地址。您将收到一个重设密码链接。

Error message here!

返回登录

Close

异步任务 -- FutureTask

lucare 2019-01-24 21:03:00 阅读数:192 评论数:0 点赞数:0 收藏数:0

任务提交

之前在分析线程池的时候,提到过 AbstractExecutorService 的实现:

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}

对于 submit 提交的任务,不管是 Runnable 还是 Callable,最终都会统一为 FutureTask 并传给 execute 方法。

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

对于 Runnable 还会创建一个适配器 :

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 有下面几种状态:

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;

初次创建的时候构造器中赋值 state = NEW,后面状态可能有下面几种演化:

  • NEW -> COMPLETING -> NORMAL (正常完成的过程)
  • NEW -> COMPLETING -> EXCEPTIONAL (执行过程中遇到异常)
  • NEW -> CANCELLED (执行前被取消)
  • NEW -> INTERRUPTING -> INTERRUPTED (取消时被中断)

任务执行

当线程池执行任务的时候,最终都会执行 FutureTask 的 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 must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}

对于 Callable 直接执行其 call 方法。执行成功则调用 set 方法设置结果,如果遇到异常则调用 setException 设置异常:

protected void set(V v) {
// 首先 CAS 设置 state 为中间状态 COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
// 设置为正常状态
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
// 设置为异常状态
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}

这两个方法都是对全局变量 outcome 的赋值。当我们通过 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 {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
// 通过 for 循环来阻塞当前线程
for (;;) {
// 响应中断
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
// 任务已完成或者已抛出异常 直接返回
if (s > COMPLETING) {
// WaitNode已创建此时也没用了
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
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);
}
}

如果任务已完成或者等待任务直到完成后,调用 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);
}

如果 state == NORMAL,标识任务正常完成,返回实际结果。如果 state >= CANCELLED, 则返回 CancellationException,否则返回 ExecutionException,这样在线程池中执行的任务不管是异常还是正常返回了结果,都能被感知。

Treiber Stack

/**
* Simple linked list nodes to record waiting threads in a Treiber
* stack. See other classes such as Phaser and SynchronousQueue
* for more detailed explanation.
*/
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}

在 awaitDone 方法中 WaitNode q = null,第一次会创建一个 WaitNode,这时即使有多个线程在等待结果,都会创建各自的 WaitNode:

else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);

然后在for循环中会跳到第二个 else if,由于没有入队,这时会通过 CAS 将新建的 WaitNode 类型的 q 赋值给 waiters,这个时候同一时刻只有一个线程能赋值成功,后一个在失败后又经历一次循环,最终成功地将当前 WaitNode 插入到 waiters 的头部。

任务取消

FutureTask 有一个 cancel 方法,包含一个 boolean 类型的参数(在执行中的任务是否可以中断):

public boolean cancel(boolean mayInterruptIfRunning) {
// 如果任务不是刚创建或者是刚创建但是更改为指定状态失败则返回 false
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}

最终都会调用 finishCompletion() ,在 set 方法和 setException 方法中也调用了这个 finishCompletion 方法:

private void finishCompletion() {
// assert state > COMPLETING;
// 如果任务执行完或者存在异常的话 这个waiters已经为null了
for (WaitNode q; (q = waiters) != null;) {
// 首先不断尝试把 waiters 设置为 null,如果很多线程调用 task.cancel(),也只有一个能成功
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; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}

当在 finishCompletion 方法中唤醒线程后,被唤醒的线程在 awaitDone 方法中继续循环,发现状态已完成:

int s = state;
// 任务已完成或者已抛出异常 直接返回
if (s > COMPLETING) {
// WaitNode已创建此时也没用了
if (q != null)
q.thread = null;
return s;
}

接着调用 report 方法,发现状态为异常的话将包装成 ExecutionException((Throwable)x); 这个异常就是我们在使用 get 的时候需要捕获的异常。

最近比较忙,这块东西已经很久没有看了, FutureTask 感觉没有彻底弄明白,也没有一个好的结尾,现在这里标记下,后面继续更新。

版权声明
本文为[lucare]所创,转载请带上原文链接,感谢
https://www.cnblogs.com/lucare/p/10316808.html

编程之旅,人生之路,不止于编程,还有诗和远方。
阅代码原理,看框架知识,学企业实践;
赏诗词,读日记,踏人生之路,观世界之行;

支付宝红包,每日可领