FutureTask分析

这是基于jdk1.8,doc文档有注明前一个版本是基于AQS来实现阻塞等待的,1.8改为基于一个int的state来实现。使用AQS的方式,可能会在取消发生竞争过程中诡异的保留了中断状态。这里之所以没有采用这种方式,是为了避免这种情况的发生!(只有用到线程池或者其他工作线程复用的情况下会发生。)

如下面的场景:

ThreadPoolExecutor executor = ...;  
executor.submit(task1).cancel(true);  
executor.submit(task2);  

主线程中断的是task1,但是有可能被中断的是task2.(这里的线程池执行task1和task2可能是同一工作线程完成的,这个是关键!!!)

看jdk1.6的实现:

boolean innerCancel(boolean mayInterruptIfRunning) {  
 for (;;) {  
  int s = getState();  
  if (ranOrCancelled(s))  
      return false;  
  if (compareAndSetState(s, CANCELLED))  
      break;  
 }  
    if (mayInterruptIfRunning) {  
        Thread r = runner;  
//  要理解这里,需要结合ThreadPoolExecutor源码(getTask和runWorker这两个方法,runWorker拿到一个Runnable的任务的话就直接调用其run方法了(FutureTask也是一个Runnable))
        if (r != null)  当主线程调用cancel执行到这里的时候,可能task1刚好已经执行过了,执行task1的工作线程被复用去执行task2,不巧的是,task2收到了这个中断信号!!!!!
            r.interrupt(); // 收到中断信号(此时虽然还是同一个工作线程,但是已经在执行task2了)
    }  
    releaseShared(0);  
    done();  
    return true;  
}  

为了防止中断错任务,jdk1.7,1.8版本中做了以下改进:

  • 新增INTERRUPTING,INTERRUPTED等状态
  • 弃用AQS,直接用一个volatile的state来控制状态
  • cancel时首先设置状态为INTERRUPTING, run方法结束前会查看状态如果为INTERRUPTING,就自旋等待变为INTERRUPTED(这样就能保证被中断的一定是当前任务)

1.类声明

// 实现了RunnableFuture这个叼逼接口
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();
}

2.属性

    /*
     *Possible state transitions:
     * 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;
    private static final int EXCEPTIONAL  = 3;
    private static final int CANCELLED    = 4;
    private static final int INTERRUPTING = 5; // 以下这两个是新增的状态,以前没有,也是中断bug的修复
    private static final int INTERRUPTED  = 6;

    /** The underlying callable; nulled out after running */
    // 执行的任务,执行完后会被置空
    private Callable<V> callable;
    /** The result to return or exception to throw from get() */
    // 任务执行结果,没有使用volatile是因为它与state总是同时出现,形成happen-before关系
    private Object outcome; // non-volatile, protected by state reads/writes
    /** The thread running the callable; CASed during run() */
   // 真正执行任务的线程
    private volatile Thread runner;
    /** Treiber stack of waiting threads */
    // 队列,存储那些在等待任务执行结果的线程。
    private volatile WaitNode waiters;

3.run方法

public void run() {
        // 当前状态不为new状态,或者cas设置当前执行线程出错(cas比较对象为null)
        // 将直接返回,cas与状态的检验将严格限制只会有一条线程真正执行。
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) { // 这里的state为new的检查非常重要,如果当前有线程调用了cancel方法,state一定已经不为new,就不会继续执行下面的callable了,所以任务还没开始之前调用cancel,无论参数是true还是false,都会阻止任务的继续执行
                V result;
                boolean ran; // 标记是否执行成功,必须有这个标记,因为r
                try {
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);  // 设置异常信息,并改变当前state 
                }
                if (ran)
                    set(result); // 设置执行结果,并改变state
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null; // 最后才设置runner为null,因为cas设置runner时以null为比较对象的,所以只要他不为空,就不会有多条线程进入执行run方法。
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)  // 如果当前正处于中断进行中(就是cancel方法设置了状态为中断中,但是还未发送中断信号完成),一致自旋等待中断。这里的实现是为什么弃用AQS的根本原因,防止中断信号中断了另外的任务。
                handlePossibleCancellationInterrupt(s);
        }
    }

// 中断保留在当前run方法运行范围内(防止中断错了任务)
private void handlePossibleCancellationInterrupt(int s) {
       // 等待中断线程完成中断,中断线程执行后一定会将state置为INTERRUPTED
        if (s == INTERRUPTING)
            while (state == INTERRUPTING)
                Thread.yield(); // wait out pending interrupt
    }

// 执行完后设置结果,setException类似
protected void set(V v) {
        // 只有状态为new才设置
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            // 唤醒所有等待线程,setException,cancel等方法都会调用finishCompletion
            finishCompletion();
        }
    }

4.cancel方法

public boolean cancel(boolean mayInterruptIfRunning) {
        // 任务在执行callable的run方法之后直到set或者setException时,
        // 才会先cas将状态从new改为cpmpleting,所以在state已经不为new时,已经没有必要去中断退出任务了,因为callable都已经执行完成了。
        // 如果state为new,则cas会将state置为INTERRUPTING或者CANCELLED,如果是INTERRUPTING,
        // 则与handlePossibleCancellationInterrupt方法对应,handlePossibleCancellationInterrupt会自旋等待state被置为INTERRUPTTED才会结束
        // 这里cas修改了state状态,在run方法正式执行之前会检查状态是否为new,成功调用了cancel后,run方法检测一定会失败。
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) { // 能走到这里,当前线程要么还没开始执行(没开始执行的线程永远也不会开始了,原因已经在上面讲解(cas已经修改了状态)),要么正在执行中(因为之前的state是new), 所以如果mayInterruptIfRunning设置为了false,将不会中断线程,任其执行结束,为true则发送中断信号
                try {
                    Thread t = runner;
                    if (t != null) 
                        t.interrupt(); // 中断正在运行的线程
                } finally { // final state
                    // 发送完中断信号,将状态设置为最终状态-----中断结束,此时自旋的handlePossibleCancellationInterrupt也会随后结束自旋
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            // 唤醒所有等待结果的线程
            finishCompletion();
        }
        return true;
    }

5.get方法

public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING) // 如果还没有执行完成,进入等待
            s = awaitDone(false, 0L);
        return report(s); // 已经执行完成,直接返回结果
    }

@SuppressWarnings("unchecked")
    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); // Exceptional状态
    }

下面仔细看最关键的awaitDone方法,如何做到等待结果出来的。

 // 麻痹的,这一坨if else写的很吊嘛,很巧妙啊。。。这思想值得借鉴。
 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的顺序。。。。哈哈哈哈
            if (Thread.interrupted()) { // 如果收到中断信号,响应中断,并移除等待队列中对应的节点
                removeWaiter(q);
                throw new InterruptedException();
            }

            int s = state;
            if (s > COMPLETING) { // 如果已经运行结束了,就不用等了。
                if (q != null)
                    q.thread = null;
                return s;
            }
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield(); // 如果处于结束过程中这个状态,说明callable已经执行完成,马上就会完全结束,自旋等待一会
            else if (q == null) // 到这里就必须新建一个节点进队列去了。
                q = new WaitNode();
            else if (!queued)  // cas将节点入队列
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            else if (timed) { // 如果有时间限制
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) { // 超时则直接返回并移除
                    removeWaiter(q);
                    return state;
                }
                // 挂起当前线程,nanos超时
                LockSupport.parkNanos(this, nanos);
            }
            else
                LockSupport.park(this);  // 直接挂起当前线程
        }
    }
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 218,284评论 6 506
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 93,115评论 3 395
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 164,614评论 0 354
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 58,671评论 1 293
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 67,699评论 6 392
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 51,562评论 1 305
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,309评论 3 418
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 39,223评论 0 276
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,668评论 1 314
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,859评论 3 336
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,981评论 1 348
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,705评论 5 347
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,310评论 3 330
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,904评论 0 22
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,023评论 1 270
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 48,146评论 3 370
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,933评论 2 355

推荐阅读更多精彩内容