线程阻塞(三)，FutureTask
2018-05-17 本文已影响67人
William_hi
今天介绍一下FutureTask的使用。FutureTask有两个使用场景：一个是保证线程阻塞；另外一个是FutureTask在高并发环境下确保任务只执行一次。
1、使用场景
1.1、线程阻塞
FutureTask实现了RunnableFuture<V>接口，RunnableFuture<V>接口继承了Runnable, Future<V>
public class FutureTask<V> implements RunnableFuture<V>{
...
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
也就是说，FutureTask既是Runnable对象，也是Future对象。对于Runnable大家都比较了解了，Future是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果、设置结果操作，get方法会阻塞，直到任务返回结果，也就是说，阻塞。
下面介绍一下FutureTask的使用，先看下构造方法
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
}
有两个构造方法，一个接收Callable对象，一个接收Runnable，大家可以看下第二个构造方法，接收到Runnable对象后，又封装成了Callable对象。那么Callable是什么呢？
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
对于Runnable都比较熟悉了，里面主要是有一个 void run() 方法，而Callable对应的有一个call() 方法，这两个类由什么区别呢？
(1)Callable执行的方法是call()，Runnable执行的方法是run()。其中Runnable可以提交给Thread来包装下，直接启动一个线程来执行，或者交给ExecuteService执行，而Callable则一般都是提交给ExecuteService来执行。
(2)Callable的任务执行后可返回值，而Runnable的任务是不能返回值得
(3)call方法可以抛出异常，run方法不可以
(4)运行Callable任务可以拿到一个Future对象，表示异步计算的结果，而且拿到的这个Future对象后，如果调用get()方法，当前线程会阻塞，直到此Callable的call方法执行完毕。
下面写一个简单的例子，由Thread执行
public static void main(String[] args) {
CountSum countSum = new CountSum();// 创建一个Callable对象
FutureTask<Integer> futureTask = new FutureTask<>(countSum);// 创建FutureTask对象
Thread thread = new Thread(futureTask);// 开启线程
thread.start();
try {
int sum = futureTask.get();
System.out.println("主线程获得求和结果：" + sum);
} catch (Exception e) {
e.printStackTrace();
futureTask.cancel(true);
}
}
static class CountSum implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println("开始求和");
int sum = 0;
for (int i = 0; i < 10000; i++) {
sum += i;
}
Thread.sleep(2000);
System.out.println("求和结束");
return sum;
}
}
执行结果
开始求和
求和结束
主线程获得求和结果：49995000
打印"开始求和"后等待了一会，才再打印出结果。
另外，Futuretask也可以使用ExecutorService执行
public static void main(String[] args) {
CountSum countSum = new CountSum();// 创建一个Callable对象
FutureTask<Integer> futureTask = new FutureTask<>(countSum);// 创建FutureTask对象
ExecutorService executorService = Executors.newCachedThreadPool();// 开启线程
executorService.execute(futureTask);
try {
int sum = futureTask.get();
System.out.println("主线程获得求和结果：" + sum);
} catch (Exception e) {
e.printStackTrace();
futureTask.cancel(true);
} finally {
executorService.shutdown();
}
}
结果和上面Thread执行是一样的。
1.2、在高并发环境下确保任务只执行一次。
网上有一个很典型的例子
private ConcurrentHashMap<String,FutureTask<Connection>> connectionPool = new ConcurrentHashMap<String, FutureTask<Connection>>();
public Connection getConnection(String key) throws Exception{
FutureTask<Connection>connectionTask=connectionPool.get(key);
if(connectionTask!=null){
return connectionTask.get();
}else{
Callable<Connection> callable = new Callable<Connection>(){
@Override
public Connection call() throws Exception {
return createConnection();
}
};
FutureTask<Connection> newTask = new FutureTask<Connection>(callable);
connectionTask = connectionPool.putIfAbsent(key, newTask);
if(connectionTask==null){
connectionTask = newTask;
connectionTast.run(); //保证执行的一定是当前futureTask
}
return connectionTask.get();
}
}
//创建Connection
private Connection createConnection(){
return null;
}
相对于直接创建Connection，这里使用了FutureTask，先put到map里，再通过get()方法创建连接，保证了连接只创建一次。只是，在高并发的情况下，有可能Callable和FutureTask会创建多次，这里只是保证了只有一个FutureTask调用了get()方法从而执行了创建连接的方法，感觉还是没有加锁安全，了解一下就可以。
2、源码解析
JDK1.6使用AQS构建，JDK1.7是在内部采用简单的Stack来保存等待线程。下面分析一下JDK1.7的版本。
FutureTask 类结构
public class FutureTask<V> implements RunnableFuture<V> {
/**
* 运行状态。
*
* 可能存在的状态转换
* 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; //任务正常完成，结果被set
private static final int EXCEPTIONAL = 3; //任务抛出异常
private static final int CANCELLED = 4; //任务已被取消
private static final int INTERRUPTING = 5; //线程中断状态被设置ture，但线程未响应中断
private static final int INTERRUPTED = 6; //线程已被中断
//将要执行的任务，运行完成后设置为null
private Callable<V> callable;
//用于get()返回的结果，也可能是用于get()方法抛出的异常
private Object outcome; // 本身没有volatile修饰, 依赖state的读写来保证可见性。
//执行callable的线程
private volatile Thread runner;
//存放等待线程的Treiber Stack。
private volatile WaitNode waiters;
....
看一下WaitNode代码
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
包含了当前线程对象，并有指向下一个WaitNode的指针，waiters就是由WaitNode组成的一个单向链表。
run 方法
public void run() {
//如果state不为null，尝试设置runner为当前线程，失败就退出。
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
//如果state不等于null，直接退出。
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必须在设置了state之后再置空，避免run方法出现并发问题。
runner = null;
// 这里还必须再读一次state，避免丢失中断。
int s = state;
if (s >= INTERRUPTING)
//处理可能发生的取消中断(cancel(true))。
handlePossibleCancellationInterrupt(s);
}
}
run方法正常执行完后后，会调用set方法
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
//唤醒Treiber Stack中所有等待线程。
finishCompletion();
}
}
- 首先把state的NEW状态修改成COMPLETING状态。
- 修改成功则把v值赋给outcome变量。然后再把state状态修改成NORMAL，表示现在可以获取返回值。
- 最后调用finishCompletion()方法，唤醒等待队列中的所有节点。
setException
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
和set差不多
- 首先把state的NEW状态修改成COMPLETING状态。
- 修改成功则把v值赋给outcome变量。然后再把state状态修改成EXCEPTIONAL，表示待返回的异常信息设置成功。
- 最后调用finishCompletion()方法，唤醒等待队列中的所有节点。
finishCompletion方法
private void finishCompletion() {
for (WaitNode q; (q = waiters) != null;) {
//尝试将waiters设置为null。
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
//然后将waiters中的等待线程全部唤醒。
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，减少内存占用
callable = null;
}
finishCompletion主要就是在任务执行完毕后，移除waiters栈，并将waiters中所有等待获取任务结果的线程唤醒，然后回调下done钩子方法。
handlePossibleCancellationInterrupt方法
/**
* 确保cancel(true)产生的中断发生在run或runAndReset方法过程中。
*/
private void handlePossibleCancellationInterrupt(int s) {
// 如果当前正在中断过程中，自旋等待一下，等中断完成。
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
// 这里的state状态一定是INTERRUPTED;
// 这里不能清除中断标记，因为没办法区分来自cancel(true)的中断。
// Thread.interrupted();
}
总结一下执run方法：
- 只有state为NEW的时候才执行任务(调用内部callable的run方法)。执行前会原子的设置执行线程(runner)，防止竞争。
- 如果任务执行成功，任务状态从NEW迁转为COMPLETING(原子)，设置执行结果，任务状态从COMPLETING迁转为NORMAL(LazySet)；如果任务执行过程中发生异常，任务状态从NEW迁转为COMPLETING(原子)，设置异常结果，任务状态从COMPLETING迁转为EXCEPTIONAL(LazySet)。
- 将Treiber Stack中等待当前任务执行结果的等待节点中的线程全部唤醒，同时删除这些等待节点，将整个Treiber Stack置空。
- 最后别忘了等一下可能发生的cancel(true)中引起的中断，让这些中断发生在执行任务过程中(别泄露出去)。
runAndReset方法(周期性任务的时候用到)
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} 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
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}
runAndReset与run方法的区别只是执行完毕后不设置结果、而且有返回值表示是否执行成功。
get方法
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L); //如果任务还没执行完毕，等待任务执行完毕。
return report(s); //如果任务执行完毕，获取执行结果。
}
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()) {
//如果当前线程被中断，移除等待节点q，然后抛出中断异常。
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
//如果任务已经执行完毕
if (q != null)
q.thread = null; //如果q不为null，将q中的thread置空。
return s; 返回任务状态。
}
else if (s == COMPLETING)
Thread.yield(); //如果当前正在完成过程中，出让CPU。
else if (q == null)
q = new WaitNode(); //创建一个等待节点。
else if (!queued)
//将q(包含当前线程的等待节点)入队。
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
//如果超时，移除等待节点q
removeWaiter(q);
//返回任务状态。
return state;
}
//超时的话，就阻塞给定时间。
LockSupport.parkNanos(this, nanos);
}
else
//没设置超时的话，就阻塞当前线程。
LockSupport.park(this);
}
}
awaitDone方法中调用的removeWaiter
private void removeWaiter(WaitNode node) {
if (node != null) {
//将node的thread域置空。
node.thread = null;
//下面过程中会将node从等待队列中移除，以thread域为null为依据，
//如果过程中发生了竞争，重试。
retry:
for (;;) {
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;
}
}
}
get方法中获取结果时调用的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);
}
再看下get(long timeout, TimeUnit unit)方法
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);
}
总结一下执get方法：
- 首先检查当前任务的状态，如果状态表示执行完成，进入第2步。
- 获取执行结果，也可能得到取消或者执行异常，get过程结束。
- 如果当前任务状态表示未执行或者正在执行，那么当前线程放入一个新建的等待节点，然后进入Treiber Stack进行阻塞等待。
- 如果任务被工作线程(对当前线程来说是其他线程)执行完毕，执行完毕时工作线程会唤醒Treiber Stack上等待的所有线程，所以当前线程被唤醒，清空当前等待节点上的线程域，然后进入第2步。
- 当前线程在阻塞等待结果过程中可能被中断，如果被中断，那么会移除当前线程在Treiber Stack上对应的等待节点，然后抛出中断异常，get过程结束。
- 当前线程也可能执行带有超时时间的阻塞等待，如果超时时间过了，还没得到执行结果，那么会除当前线程在Treiber Stack上对应的等待节点，然后抛出超时异常，get过程结束。
再看一下cancel方法
public boolean cancel(boolean mayInterruptIfRunning) {
if (state != NEW)
return false; //如果任务已经执行完毕，返回false。
if (mayInterruptIfRunning) {
//如果有中断任务的标志，尝试将任务状态设置为INTERRUPTING
if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
return false;
//上面设置成功的话，这里进行线程中断。
Thread t = runner;
if (t != null)
t.interrupt();
//最后将任务状态设置为INTERRUPTED，注意这里又是LazySet。
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
}
//如果没有中断任务的标志，尝试将任务状态设置为CANCELLED。
else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
return false;
//最后唤醒Treiber Stack中所有等待线程。
finishCompletion();
return true;
}
cancel方法在设置mayInterruptIfRunning为true的情况下，内部首先通过一个原子操作将state从NEW转变为INTERRUPTING，然后中断执行任务的线程，然后在通过一个LazySet的操作将state从INTERRUPTING转变为INTERRUPTED，由于后面这个操作对其他线程并不会立即可见，所以handlePossibleCancellationInterrupt才会有一个自旋等待state从INTERRUPTING变为INTERRUPTED的过程。
最后，看一下为什么这个FutureTask不像1.6那样基于AQS构建了
FutureTask有一部分类注释如下
/*
* Revision notes: This differs from previous versions of this
* class that relied on AbstractQueuedSynchronizer, mainly to
* avoid surprising users about retaining interrupt status during
* cancellation races.
*/
大概意思是：使用AQS的方式，可能会在取消发生竞争过程中诡异的保留了中断状态。这里之所以没有采用这种方式，是为了避免这种情况的发生。
那么什么情况下会发生呢？
ExecutorService executor = Executors.newCachedThreadPool();
executor.submit(task1).cancel(true);
executor.submit(task2);
上面的代码，虽然中断的是task1，但可能task2得到中断信号。
原因是什么呢？看下JDK1.6的FutureTask的中断代码：
boolean innerCancel(boolean mayInterruptIfRunning) {
for (;;) {
int s = getState();
if (ranOrCancelled(s))
return false;
if (compareAndSetState(s, CANCELLED))
break;
}
if (mayInterruptIfRunning) {
Thread r = runner;
if (r != null) //第1行
r.interrupt(); //第2行
}
releaseShared(0);
done();
return true;
}
结合上面代码例子看一下，如果主线程执行到第1行的时候，线程池可能会认为task1已经执行结束(被取消)，然后让之前执行task1的工作线程去执行task2，工作线程开始执行task2之后，然后主线程执行第2行(我们会发现并没有任何同步机制来阻止这种情况的发生)，这样就会导致task2被中断了。更多的相关信息参考这个Bug说明
所以，JDK1.7 FutureTask的handlePossibleCancellationInterrupt中将cancel(true)中的中断保留在当前run方法运行范围内了。
