【并发设计模式】FutureTask
2019-06-07 本文已影响0人
良辰夜
只谈谈脉络，不说正常人能够看懂的地方。
1.UML图
基本可以看出，我们需要实现run方法，而返回Future接口对象。
而我们可以看出，Future接口主要是一些run在执行过程中的状态，在面对多线程的状态扭转，我们通常用状态机来实现。
UML图
2.1 状态一览
/**
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
private volatile int state;
private static final int NEW = 0;//init的时候
private static final int COMPLETING = 1;//表示已经执行完毕，但是结果变量却没有写outcome的状态。
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;//打断结束
非常形象的状态列表（注意图上数字并不是实际值）
2.2 awaitDone
这是一个阻塞方法，阻塞到"完成"，即state>COMPLETING或者阻塞到nanos ns结束。
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()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//因为state一次切换大几率要变了，所以使用yield更好，比挂起响应要快。
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)//我觉得没意义，已经放在queued里面
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);
}
}
由于它是futureTask核心方法之一，所以我们详细剖析(别想多了，我就想多写几行)。
方法比较精彩的地方：
- 自旋，多次判断state状态，可以对比下Bit.reserveMemory方法的自旋过程。
- LockSupport和Thread. interrupted搭配，必须要自旋。(这点AQS也是如此)
- CAS加入队列，保证队列线程安全，非常新颖nice。
方法的疑问:
- 用LockSupport和Object.wait和notifyall谁比较好？
- 该方法存在某种意义的内存泄露。
有可能set方法线程在finishCompletion方法执行到done()时，get方法线程刚好执行queued = UNSAFE.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);这样最后变量waiters无法被赋值null。
这个bug最好解决方法：
int s = state;
if (s > COMPLETING) {
removeWaiter(q);
return s;
}
当然，作者也可能这样想的： 执行到这一步，也就基本说明FutureTask类状态已经到了最后阶段，即使被复用，这个waiters 这个最多保存 调用get线程数量的 WaitNode对象,这样的也不算浪费，可能下一次就被回收了，所以仅仅做q.thread = null就可以了
2.3 关于和happens-before相关方法
set(V v)、setException(Throwable t)
参见我之前写的:happens-before应用之FutureTask
2.4 关于CAS的链表队列的增删改查
把FutureTask里面里面关于waiters的添加、删除、清空提出来了。
private volatile WaitNode waiters;
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
public void add(){
while (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q));
}
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;
}
}
}
public void clear(){
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;
}
}
}
从添加到删掉到清空，作者把CAS用到极致了。
其中最为难理解的应该是removeWaiter这个方法。
作者首先将node.thread = null;，这样我们只需要判断node是否为null就可以了。作者为什么要这么做呢？
其主要原因，作者想判断这个删除动作是否和其他线程冲突。
这是waiters的节点，按照顺序A、B、C、D，依次排开。其中P表示prev，N表示next。
当两个线程同时执行removeWaiter时，我们针对的对象是AC,AD,BD，其实并没有什么影响，但是如果针对的是AB,BC,CD，我们发现线程可能发送不安全操作。那么作者采用一种乐观锁的方式？给每个元素一个被删除的标志，含义这个标志说明要被删除，每次线程在执行删除结束后就判断这个标记，如果存在标记，说明线程可能发生冲突，从新执行一次remove操作，这个标记就是thread==null，所以作者执行remove操作后，需要判断pred.thread == null，如果是两个线程分别触发相邻元素的remove操作，但出现线程安全问题，删除后面元素的线程一定触发这个thread==null这个判断。(注意可见性的保障来自volatile申明的WaitNode和waiters元素)
注意：
volatile修饰的变量如果是对象或数组之类的，其含义是对象获数组的地址具有可见性，但是数组或对象内部的成员改变不具备可见性：
2.5 关于run和cancel方法
没什么特别精彩的地方，中规中矩吧。
public void run() {
if (state != NEW || //不是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)//等待cancel执行完，这里有点没想通，有意义吗？
handlePossibleCancellationInterrupt(s);
}
}
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&//不是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;
}
整个设计，最突出的地方在调用interrupt，来打断等待任务，我们最好在代码里面判断interrupt，一旦出现interrupt，说明外部线程调用cancel了，我们需要快速返回。
