**移步[java多线程系列文章]
Java多线程（二十二）---LockSupport工具
Java 停止线程
一、概述
- Callable和Runnbale一样代表着任务，区别在于Callable有返回值并且可以抛出异常。
- 其使用如下：
public class CallableDemo {
static class SumTask implements Callable<Long> {
@Override
public Long call() throws Exception {
long sum = 0;
for (int i = 0; i < 9000; i++) {
sum += i;
}
return sum;
}
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("Start:" + System.nanoTime());
FutureTask<Long> futureTask = new FutureTask<Long>(new SumTask());
Executor executor=Executors.newSingleThreadExecutor();
executor.execute(futureTask);
System.out.println(futureTask.get());
System.out.println("End:" + System.nanoTime());
}
}
从上面的代码可以看到，使用到了一个FutureTask的变量并且还可以得到Callable执行的结果，那么这个FutureTask是什么呢?
二、 分析
2.0 Callable接口
public interface Callable<V> {
// 子类复写这个方法，一般都是耗时操作，并返回结果值
V call() throws Exception;
}
相当于Runnable的run方法，一般都是耗时操作，但是不一样的是，这个方法会返回结果值。
2.1 Future接口
- Future是一个接口，代表了一个异步计算的结果。
- 接口中的方法用来检查计算是否完成、等待完成和得到计算的结果。
- 当计算完成后，只能通过get()方法得到结果，get方法会阻塞直到结果准备好了。
- 如果想取消，那么调用cancel()方法。
- 其他方法用于确定任务是正常完成还是取消了。
- 一旦计算完成了，那么这个计算就不能被取消。
public interface Future<V> {
/**
* 取消当前的Future。会唤醒所有等待结果值的线程，抛出CancellationException异常
* @param mayInterruptIfRunning 是否中断 计算结果值的那个线程
* @return 返回true表示取消成功
*/
boolean cancel(boolean mayInterruptIfRunning);
// 当前的Future是否被取消，返回true表示已取消。
boolean isCancelled();
// 当前Future是否已结束。包括运行完成、抛出异常以及取消，都表示当前Future已结束
boolean isDone();
// 获取Future的结果值。如果当前Future还没有结束，那么当前线程就等待，
// 直到Future运行结束，那么会唤醒等待结果值的线程的。
V get() throws InterruptedException, ExecutionException;
// 获取Future的结果值。与get()相比较多了允许设置超时时间。
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
2.1 RunnableFuture接口
// 继承自Runnable和Future接口。
public interface RunnableFuture<V> extends Runnable, Future<V> {
void run();
}
RunnableFuture接口的作用就是它的子类可以当做Runnable接口使用，那么创建一个新线程的时候，就可以使用它的实例作为参数了。
2.2 FutureTask类
- FutureTask类实现了RunnableFuture接口，而RunnnableFuture接口继承了Runnable和Future接口，所以说FutureTask是一个提供异步计算的结果的任务。
- FutureTask可以用来包装Callable或者Runnbale对象。因为FutureTask实现了Runnable接口，所以FutureTask也可以被提交给Executor（如上面例子那样）。
2.3 FutureTask的状态
FutureTask中有一个表示任务状态的int值，初始为NEW。定义如下：
// 表示FutureTask当前的状态
private volatile int state;
// NEW 新建状态，表示这个FutureTask还没有开始运行
private static final int NEW = 0;
// COMPLETING 完成状态， 表示FutureTask任务已经计算完毕了，
// 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
private static final int COMPLETING = 1;
// FutureTask任务完结，正常完成，没有发生异常
private static final int NORMAL = 2;
// FutureTask任务完结，因为发生异常。
private static final int EXCEPTIONAL = 3;
// FutureTask任务完结，因为取消任务
private static final int CANCELLED = 4;
// FutureTask任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求。
private static final int INTERRUPTING = 5;
// FutureTask任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求。
private static final int INTERRUPTED = 6;
可能的状态转换包括：
- NEW -> COMPLETING -> NORMAL
- NEW -> COMPLETING -> EXCEPTIONAL
- NEW -> CANCELLED
- NEW -> INTERRUPTING -> INTERRUPTED
2.4 构造方法
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
}
第一个构造方法好理解；第二个方法是将Runnbale和结果组合成一个Callable，这个可以通过Excutors.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;
}
}
- 从上面可以看到RunnableAdapter实现了Callable并且在call方法中调用了Runnable的run方法，然后将结果返回，这其实就是一个适配器模式
- 所以说两个构造方法最终都是得到了一个Callable以及设置了初始状态为NEW。
2.5 run方法
- 当将FutureTask提交给Executor后，Executor执行FutureTask时会执行其run方法
// 开始运行FutureTask任务
public void run() {
// 如果状态state不是NEW，或者设置runner值失败
// 表示有别的线程在此之前调用run方法，并成功设置了runner值
// 保证了只有一个线程可以运行try 代码块中的代码。
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
//尝试调用Callable.call
try {
// 使用一个变量c记录callable，防止多线程情况下，
// callable直接被设置为null出现问题
Callable<V> c = callable;
// 只有c不为null且状态state为NEW的情况，
if (c != null && state == NEW) {
V result;
boolean ran;
try {
// 调用callable的call方法，并返回结果
result = c.call();
// 运行成功
ran = true;
} catch (Throwable ex) {
//出现异常了，调用setException方法
result = null;
ran = false;
// 设置异常
setException(ex);
}
// 如果运行成功，则设置结果
if (ran)
set(result);
}
} finally {
runner = null;
int s = state;
// 当状态大于或等于INTERRUPTING，调用handlePossibleCancellationInterrupt方法，
// 等待别的线程将状态设置成INTERRUPTED
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
- 从上面可以看到，任务可以被执行的前提是当前状态为NEW以及CAS当前执行线程成功，也就是runner值，代表执行Callable的线程
- 从这个看到run方法就是调用Callable的call方法，然后如果出现异常了就调用setException方法，如果成功执行了，那么调用set方法
2.6 set方法
- 当Callable成功执行后，会调用set方法将结果传出。
protected void set(V v) {
//完成NEW->COMPLETING->NORMAL状态转换
// 调用CAS函数，将状态state从NEW改成COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
// 延迟设置，将状态改成NORMAL，
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
// 调用finishCompletion唤醒所有等待结果的线程
finishCompletion();
}
}
- 先使用CAS函数将状态state从NEW变成COMPLETING，防止多线程冲突。
- 使用putOrderedInt方法设置状态state是NORMAL或EXCEPTIONAL。
- 调用finishCompletion方法唤醒所有等待结果的线程。
2.7 setException方法
当想得到FutureTask的结算结果时，调用get方法，get方法可以允许多个线程调用，下面的例子展示了多个线程调用get的情况。
//完成NEW->COMPLETING->EXCEPTIONAL状态转换
protected void setException(Throwable t) {
// 调用CAS函数，将状态state从NEW改成COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
// 延迟设置，将状态改成EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
// 调用finishCompletion唤醒所有等待结果的线程
finishCompletion();
}
}
- 先使用CAS函数将状态state从NEW变成COMPLETING，防止多线程冲突。
- 使用putOrderedInt方法设置状态state是NORMAL或EXCEPTIONAL。
- 调用finishCompletion方法唤醒所有等待结果的线程。
- 注: putOrderedInt方法的意义。因为state变量是被volatile关键字修饰,那么它会给state变量加一个内存屏障，来保证state变量的可见性和有序性，这样会消耗一些性能。
- 而putOrderedInt方法的意义，就是通过它来设置volatile修饰的变量，会取消这个内存屏障，也就是像普通变量一样了，不保证可见性了。
2.8 get方法
当想得到FutureTask的结算结果时，调用get方法，get方法可以允许多个线程调用
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("Start:" + System.nanoTime());
FutureTask<Long> futureTask = new FutureTask<Long>(new SumTask());
Executor executor=Executors.newSingleThreadExecutor();
executor.execute(futureTask);
for(int i=0;i<5;i++){
executor.execute(new Runnable() {
@Override
public void run() {
try {
System.out.println("get result "+futureTask.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
});
}
System.out.println(futureTask.get());
System.out.println("End:" + System.nanoTime());
}
该例子展示了一共有5个线程想得到FutureTask的结果，一旦调用get，那么该线程就会阻塞。
FutureTask的get方法实现如下：
public V get() throws InterruptedException, ExecutionException {
int s = state;
/**
* 状态小于等于COMPLETING，表示FutureTask任务还没有完结，
* 所以调用awaitDone方法，让当前线程等待
*/
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，表示FutureTask任务还没有完结，
* 所以调用awaitDone方法，让当前线程等待。
* 与get()不同的是，如果到了规定时间，任务状态仍然是小于等于COMPLETING，
* 那么就抛出TimeoutException超时异常
*/
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
// 返回结果值或者抛出异常
return report(s);
}
- 从上面的代码可以看到，如果当前任务的状态不大于COMPLETING，那么会调用awaitDone方法，这个方法会将调用的线程挂起；否则直接调用report方法返回结果。
- 在前面set和setException方法中可以得出结论：当状态从NEW变为COMPLETING后，才会将outcome赋值，也就是状态是NEW或者COMPLETING时，outcome都还未赋值，也就意味着计算仍在进行，那么此时想要get到结果，就必须等待。
2.9 awaitDone 将当前线程插入到等待队列中
- awaitDone的两个参数分别表示是否定时，以及定时的时间多少。
- get的另一个重载方法就提供了超时限制。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 (;;) {
// 如果当前线程中断标志位是true，
// 那么从列表中移除节点q，并抛出InterruptedException异常
if (Thread.interrupted()) {
// 调用removeWaiter方法从链表中移除节点q
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
//如果状态大于COMPLETING，说明已经计算已经完成了
if (s > COMPLETING) {
if (q != null)
// 将节点q线程设置为null，因为线程没有阻塞等待
q.thread = null;
return s;
}
//状态是COMPLETING，在set和setException方法中可以看到处于该状态马上就会进入下一个状态
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//新建一个等待节点
// 代码来到这里，表示状态是NEW，那么就需要将当前线程阻塞等待。
// 就是将它插入等待线程链表中，
else if (q == null)
q = new WaitNode();
//还没有入队，尝试入队
else if (!queued)
// 使用CAS函数将新节点添加到链表中，如果添加失败，那么queued为false，
// 下次循环时，会继续添加，知道成功。
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
// timed为true表示需要设置超时
else if (timed) {
// 得到剩余时间
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
//挂起指定时间
LockSupport.parkNanos(this, nanos);
}
//无限挂起
else
LockSupport.park(this);
}
}
上面的代码中有一个WaitNode类，该类表示等待节点，保存等待的线程以及下一个节点，是一个单链表结构，其定义如下:
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
awaitDone方法中进入死循环后，主要有几步：
- 如果线程被中断了，移除节点，抛出异常
- 如果状态大于COMPLETING，那么直接返回
- 如果状态是COMPLETING，在set和setException可以看到，处于COMPLETING是一个暂时状态，很快就会进入下一个状态，所以这儿就调用了Thread.yield()方法让步一下
- 如果状态是NEW且节点为null，那么创建一个节点
- 如果还没有将当前线程加入队列，那么将当前线程加入到等待队列中。由于WaitNode是一个单链表结构，FutureTask中保存了waiters的变量，就可以沿着该变量得到所有等待的线程
- 如果限制了时间，那么计算出生出超出时间，挂起指定时间。当解除挂起时，如果计算还未完成，那么将会由于没有时间了，调用removeWaiter方法移除节点。
- 如果没有限制时间，那么将线程无限挂起
上面几种情况下，都涉及了移除节点，removeWaiter方法就是删除单链表中一个节点的实现。
2.10 report
当线程被解除挂起，或计算已经完成后，将会get方法中将会调用report返回结果，其实现如下：
private V report(int s) throws ExecutionException {
Object x = outcome;
// 表示正常完结状态，所以返回结果值
if (s == NORMAL)
return (V)x;
// 大于或等于CANCELLED，都表示手动取消FutureTask任务，
// 所以抛出CancellationException异常
if (s >= CANCELLED)
throw new CancellationException();
// 否则就是运行过程中，发生了异常，这里就抛出这个异常
throw new ExecutionException((Throwable)x);
}
从上面可以看到report会根据任务的状态不同返回不同的结果。
- 如果计算正常结束，即状态是NORMAL，那么返回正确的计算结果
- 如果计算被取消了，即状态大于等于CANCELLED，那么抛出CancellationException
- 如果计算以异常结束，即状态是EXCEPTIONAL，那么抛出ExecutionException
2.11 finishCompletion方法
- 在set方法和setException方法中，当将结果赋值后，都调用了finishCompletion方法来移除和通知等待线程。
- 由于get方法中可以挂起了一群等待节点，那么当结果被计算出来了，自然应该通知那些等待线程。
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; // unlink to help gc
q = next;
}
break;
}
}
// 钩子方法。子类可以复写这个方法。
done();
callable = null; // to reduce footprint
}
- finishCompletion的实现比较简单，就是遍历等待线程的单链表，释放那些等待线程。
- 当线程被释放后，那么在awaitDone的死循环中就会进入下一个循环，由于状态已经变成了NORMAL或者EXCEPTIONAL，将会直接跳出循环。
- 释放了所有线程后，将会调用done()方法，FutureTask的done()方法默认没有任何实现，子类可以在该方法中调用完成回调以及记录操作等等。
2.12 cancel方法
- cancel方法用于取消Callable的计算。
- 参数mayInterruptIfRunning指明是否应该中断正在运行的任务，返回值表示取消是否成功了。
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 {
//最终状态INTERRUPTED
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
//释放等待线程
finishCompletion();
}
return true;
}
- 从上面可以看到如果是需要中断正在执行的任务，那么状态转换将会是NEW->INTERRPUTING->INTERRUPTED；
- 如果不需要中断正在执行的任务，那么状态转换将会是NEW->CANCELD。
- 不管是否中断，最终都会调用finishCompletion()完成对等待线程的释放。
- 当这些线程释放后，再进入到awaitDone中的循环时，返回的状态将会是大于等于CANCELD，在report方法中将会得到CancellationException异常。
2.13 isDone方法
- Future接口中isDone方法表明任务是否已经完成了，如果完成了，那么返回true，否则false。
下面是FutureTask的实现：
public boolean isDone() {
return state != NEW;
}
可以看到只要状态从初始状态NEW完成了一次转换，那么就说明任务已经被完成了。
总结
- Callable是一种可以返回结果的任务，这是它与Runnable的区别，但是通过适配器模式可以使Runnable与Callable类似。
- Future代表了一个异步的计算，可以从中得到计算结果、查看计算状态，其实现FutureTask可以被提交给Executor执行，多个线程可以从中得到计算结果。
- Callable和Future是配合使用的，当从Future中get结果时，如果结果还没被计算出来，那么线程将会被挂起，FutureTak内部使用一个单链表维持等待的线程；
- 当计算结果出来后，将会对等待线程解除挂起，等待线程就都可以得到计算结果了。
