Future模式

发布时间:2024年01月24日

FutureTask类

原作者:源码笔记
在这里插入图片描述
??FutureTask实现RunnableFutrue。RunnableFutrue接口又继承Future和Runnable接口。所以,FutureTask既可以提交到线程异步执行,也可以通过Futrue接口获取结果。Future接口相当于当前任务的凭证,可以通过Future获取当前任务的执行情况。

FutrueTask的属性变量及构造函数

?? int state状态。
??(1)正常情况状态流转。NEW–>COMPLETING–>NORMAL
??(2)异常情况状态流转。NEW -> COMPLETING -> EXCEPTIONAL
??(3)取消,中断。
NEW -> CANCELLED 或 NEW -> INTERRUPTING -> INTERRUPTED
??Callable callable变量。记录CallAble接口的实现类
??Object outcome变量。Object类型的返回对象。
??Thread runner变量。执行任务的线程对象。
?? WaitNode waiters变量。get方法被阻塞的线程封装的节点对象。这是一个单链表结构。

?? 有两个构造函数。一个是入参为Callable callable的构造函数。callable赋值给变量callable。如下

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

?? 另一个构造函数,入参为Runnable runnable, V result。通过Executors.callable方法,将runnable转换为callable。result作为callable正常执行结束的返回值。

  public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }
public class FutureTask<V> implements RunnableFuture<V> {
 
    /**
     * 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;
    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() */
    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;

    /**
     * Returns result or throws exception for completed task.
     *
     * @param s completed state value
     */
    @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);
    }

    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Callable}.
     *
     * @param  callable the callable task
     * @throws NullPointerException if the callable is null
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Runnable}, and arrange that {@code get} will return the
     * given result on successful completion.
     *
     * @param runnable the runnable task
     * @param result the result to return on successful completion. If
     * you don't need a particular result, consider using
     * constructions of the form:
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @throws NullPointerException if the runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

FutureTask.run方法,用来执行异步任务

public void run() {
        // 【1】,为了防止多线程并发执行异步任务,这里需要判断线程满不满足执行异步任务的条件,有以下三种情况:
    // 1)若任务状态state为NEW且runner为null,说明还未有线程执行过异步任务,此时满足执行异步任务的条件,
    // 此时同时调用CAS方法为成员变量runner设置当前线程的值;
    // 2)若任务状态state为NEW且runner不为null,任务状态虽为NEW但runner不为null,说明有线程正在执行异步任务,
    // 此时不满足执行异步任务的条件,直接返回;
    // 1)若任务状态state不为NEW,此时不管runner是否为null,说明已经有线程执行过异步任务,此时没必要再重新
    // 执行一次异步任务,此时不满足执行异步任务的条件;
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        // 拿到之前构造函数传进来的callable实现类对象,其call方法封装了异步任务执行的逻辑
        Callable<V> c = callable;
        // 若任务还是新建状态的话,那么就调用异步任务
        if (c != null && state == NEW) {
            // 异步任务执行结果
            V result;
            // 异步任务执行成功还是始遍标志
            boolean ran;
            try {
                // 【2】,执行异步任务逻辑,并把执行结果赋值给result
                result = c.call();
                // 若异步任务执行过程中没有抛出异常,说明异步任务执行成功,此时设置ran标志为true
                ran = true;
            } catch (Throwable ex) {
                result = null;
                // 异步任务执行过程抛出异常,此时设置ran标志为false
                ran = false;
                // 【3】设置异常,里面也设置state状态的变化
                setException(ex);
            }
            // 【3】若异步任务执行成功,此时设置异步任务执行结果,同时也设置状态的变化
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        // 异步任务正在执行过程中,runner一直是非空的,防止并发调用run方法,前面有调用cas方法做判断的
        // 在异步任务执行完后,不管是正常结束还是异常结束,此时设置runner为null
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        // 线程执行异步任务后的任务状态
        int s = state;
        // 【4】如果执行了cancel(true)方法,此时满足条件,
        // 此时调用handlePossibleCancellationInterrupt方法处理中断
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
    }

??1.判断线程是否满足执行异步任务的条件:为了防止多线程并发执行异步任务,这里需要判断线程满不满足执行异步任务的条件;
??2.若满足条件,执行异步任务:因为异步任务逻辑封装在Callable.call方法中,此时直接调用Callable.call方法执行异步任务,然后返回执行结果;
??3.根据异步任务的执行情况做不同的处理:1) 若异步任务执行正常结束,此时调用set(result);来设置任务执行结果;2)若异步任务执行抛出异常,此时调用setException(ex);来设置异常
??4.异步任务执行完后的善后处理工作:不管异步任务执行成功还是失败,若其他线程有调用FutureTask.cancel(true),此时需要调用handlePossibleCancellationInterrupt方法处理中断
??判断线程满不满足执行异步任务条件时,runner是否为null是调用UNSAFE的CAS方法compareAndSwapObject来判断和设置的,同时compareAndSwapObject是通过成员变量runner的偏移地址runnerOffset来给runner赋值的,此外,成员变量runner被修饰为volatile是在多线程的情况下, 一个线程的volatile修饰变量的设值能够立即刷进主存,因此值便可被其他线程可见。

FutureTask的set和setException方法

??下面我们来看下当异步任务执行正常结束时,此时会调用set(result);方法:

protected void set(V v) {
    // 【1】调用UNSAFE的CAS方法判断任务当前状态是否为NEW,若为NEW,则设置任务状态为COMPLETING
    // 【思考】此时任务不能被多线程并发执行,什么情况下会导致任务状态不为NEW?
    // 答案是只有在调用了cancel方法的时候,此时任务状态不为NEW,此时什么都不需要做,
    // 因此需要调用CAS方法来做判断任务状态是否为NEW
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 【2】将任务执行结果赋值给成员变量outcome
        outcome = v;
        // 【3】将任务状态设置为NORMAL,表示任务正常结束
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        // 【4】调用任务执行完成方法,此时会唤醒阻塞的线程,调用done()方法和清空等待线程链表等
        finishCompletion();
    }
}

??可以看到当异步任务正常执行结束后,且异步任务没有被cancel的情况下,此时会做以下事情:将任务执行结果保存到FutureTask的成员变量outcome中的,赋值结束后会调用finishCompletion方法来唤醒阻塞的线程(阻塞线程为get方法获取结果阻塞挂起的线程),值得注意的是这里对应的任务状态变化是NEW -> COMPLETING -> NORMAL。
??当异步任务执行过程中抛出异常,此时会调用setException(ex);方法。

protected void setException(Throwable t) {
    // 【1】调用UNSAFE的CAS方法判断任务当前状态是否为NEW,若为NEW,则设置任务状态为COMPLETING
    // 【思考】此时任务不能被多线程并发执行,什么情况下会导致任务状态不为NEW?
    // 答案是只有在调用了cancel方法的时候,此时任务状态不为NEW,此时什么都不需要做,
    // 因此需要调用CAS方法来做判断任务状态是否为NEW
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 【2】将异常赋值给成员变量outcome
        outcome = t;
        // 【3】将任务状态设置为EXCEPTIONAL
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        // 【4】调用任务执行完成方法,此时会唤醒阻塞的线程,调用done()方法和清空等待线程链表等
        finishCompletion();
    }
}

??可以看到setException(Throwable t)的代码逻辑跟前面的set(V v)几乎一样,不同的是任务执行过程中抛出异常,此时是将异常保存到FutureTask的成员变量outcome中,还有,值得注意的是这里对应的任务状态变化是NEW -> COMPLETING -> EXCEPTIONAL。
??因为异步任务不管正常还是异常结束,此时都会调用FutureTask的finishCompletion方法来唤醒唤醒阻塞的线程,这里阻塞的线程是指我们调用Future.get方法时若异步任务还未执行完,此时该线程会阻塞。

private void finishCompletion() {
    // assert state > COMPLETING;
    // 取出等待线程链表头节点,判断头节点是否为null
    // 1)若线程链表头节点不为空,此时以“后进先出”的顺序(栈)移除等待的线程WaitNode节点
    // 2)若线程链表头节点为空,说明还没有线程调用Future.get()方法来获取任务执行结果,固然不用移除
    for (WaitNode q; (q = waiters) != null;) {
        // 调用UNSAFE的CAS方法将成员变量waiters设置为空
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                // 取出WaitNode节点的线程
                Thread t = q.thread;
                // 若取出的线程不为null,则将该WaitNode节点线程置空,且唤醒正在阻塞的该线程
                if (t != null) {
                    q.thread = null;
                    //【重要】唤醒正在阻塞的该线程
                    LockSupport.unpark(t);
                }
                // 继续取得下一个WaitNode线程节点
                WaitNode next = q.next;
                // 若没有下一个WaitNode线程节点,说明已经将所有等待的线程唤醒,此时跳出for循环
                if (next == null)
                    break;
                // 将已经移除的线程WaitNode节点的next指针置空,此时好被垃圾回收
                q.next = null; // unlink to help gc
                // 再把下一个WaitNode线程节点置为当前线程WaitNode头节点
                q = next;
            }
            break;
        }
    }
    // 不管任务正常执行还是抛出异常,都会调用done方法
    done();
    // 因为异步任务已经执行完且结果已经保存到outcome中,因此此时可以将callable对象置空了
    callable = null;        // to reduce footprint
}

??finishCompletion方法的作用就是不管异步任务正常还是异常结束,此时都要唤醒且移除线程等待链表的等待线程节点,这个链表实现的是一个是Treiber stack,因此唤醒(移除)的顺序是"后进先出"即后面先来的线程先被先唤醒(移除)。

FutureTask的handlePossibleCancellationInterrupt方法

??run方法里的最后有一个finally块,此时若任务状态state >= INTERRUPTING,此时说明有其他线程执行了cancel(true)方法,此时需要让出CPU执行的时间片段给其他线程执行,我们来看下具体的源码:

private void handlePossibleCancellationInterrupt(int s) {
    // It is possible for our interrupter to stall before getting a
    // chance to interrupt us.  Let's spin-wait patiently.
    // 当任务状态是INTERRUPTING时,此时让出CPU执行的机会,让其他线程执行
    if (s == INTERRUPTING)
        while (state == INTERRUPTING)
            Thread.yield(); // wait out pending interrupt

    // assert state == INTERRUPTED;

    // We want to clear any interrupt we may have received from
    // cancel(true).  However, it is permissible to use interrupts
    // as an independent mechanism for a task to communicate with
    // its caller, and there is no way to clear only the
    // cancellation interrupt.
    //
    // Thread.interrupted();
}

FutureTask.get方法,获取任务执行结果

前面我们起一个线程在其`run`方法中执行异步任务后,此时我们可以调用`FutureTask.get`方法来获取异步任务执行的结果。

// FutureTask.java

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 【1】若任务状态<=COMPLETING,说明任务正在执行过程中,此时可能正常结束,也可能遇到异常
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    // 【2】最后根据任务状态来返回任务执行结果,此时有三种情况:1)任务正常执行;2)任务执行异常;3)任务被取消
    return report(s);
}

??可以看到,如果任务状态state<=COMPLETING,说明异步任务正在执行过程中,此时会调用awaitDone方法阻塞等待;当任务执行完后,此时再调用report方法来报告任务结果,此时有三种情况:1)任务正常执行;2)任务执行异常;3)任务被取消。

FutureTask.awaitDone方法

??FutureTask.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 (;;) {
        // 如果当前获取任务执行结果的线程被中断,此时移除该线程WaitNode链表节点,并抛出InterruptedException
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }

        int s = state;
        // 【5】如果任务状态>COMPLETING,此时返回任务执行结果,其中此时任务可能正常结束(NORMAL),可能抛出异常(EXCEPTIONAL)
        // 或任务被取消(CANCELLED,INTERRUPTING或INTERRUPTED状态的一种)
        if (s > COMPLETING) {
            // 【问】此时将当前WaitNode节点的线程置空,其中在任务结束时也会调用finishCompletion将WaitNode节点的thread置空,
            // 这里为什么又要再调用一次q.thread = null;呢?
            // 【答】因为若很多线程来获取任务执行结果,在任务执行完的那一刻,此时获取任务的线程要么已经在线程等待链表中,要么
            // 此时还是一个孤立的WaitNode节点。在线程等待链表中的的所有WaitNode节点将由finishCompletion来移除(同时唤醒)所有
            // 等待的WaitNode节点,以便垃圾回收;而孤立的线程WaitNode节点此时还未阻塞,因此不需要被唤醒,此时只要把其属性置为
            // null,然后其有没有被谁引用,因此可以被GC。
            if (q != null)
                q.thread = null;
            // 【重要】返回任务执行结果
            return s;
        }
        // 【4】若任务状态为COMPLETING,此时说明任务正在执行过程中,此时获取任务结果的线程需让出CPU执行时间片段
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        // 【1】若当前线程还没有进入线程等待链表的WaitNode节点,此时新建一个WaitNode节点,并把当前线程赋值给WaitNode节点的thread属性
        else if (q == null)
            q = new WaitNode();
        // 【2】若当前线程等待节点还未入线程等待队列,此时加入到该线程等待队列的头部
        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);
        }
        // 【3】若没有超时设置,此时直接阻塞当前线程
        else
            LockSupport.park(this);
    }
}

??1. 首先awaitDone方法里面是一个死循环;
??2. 若获取结果的当前线程被其他线程中断,此时移除该线程WaitNode链表节点,并抛出InterruptedException;
??3. 如果任务状态state>COMPLETING,此时返回任务执行结果;
??4. 若任务状态为COMPLETING,此时获取任务结果的线程需让出CPU执行时间片段;
若q == null,说明当前线程还未设置到WaitNode节点,此时新建WaitNode节点并设置其thread属性为当前线程;
??5. 若queued==false,说明当前线程WaitNode节点还未加入线程等待链表,此时加入该链表的头部;
??6. 当timed设置为true时,此时该方法具有超时功能,关于超时的逻辑这里不详细分析;
??7. 当前面6个条件都不满足时,此时阻塞当前线程。

??我们分析到这里,可以直到执行异步任务只能有一个线程来执行,而获取异步任务结果可以多线程来获取,当异步任务还未执行完时,此时获取异步任务结果的线程会加入线程等待链表中,然后调用调用LockSupport.park(this);方法阻塞当前线程。直到异步任务执行完成,此时会调用finishCompletion方法来唤醒并移除线程等待链表的每个WaitNode节点,这里这里唤醒(移除)WaitNode节点的线程是从链表头部开始的,前面我们也已经分析过。

FutureTask.cancel方法,取消执行任务

public boolean cancel(boolean mayInterruptIfRunning) {
    // 【1】判断当前任务状态,若state == NEW时根据mayInterruptIfRunning参数值给当前任务状态赋值为INTERRUPTING或CANCELLED
    // a)当任务状态不为NEW时,说明异步任务已经完成,或抛出异常,或已经被取消,此时直接返回false。
    // TODO 【问题】此时若state = COMPLETING呢?此时为何也直接返回false,而不能发出中断异步任务线程的中断信号呢??
    // TODO 仅仅因为COMPLETING是一个瞬时态吗???
    // b)当前仅当任务状态为NEW时,此时若mayInterruptIfRunning为true,此时任务状态赋值为INTERRUPTING;否则赋值为CANCELLED。
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        // 【2】如果mayInterruptIfRunning为true,此时中断执行异步任务的线程runner(还记得执行异步任务时就把执行异步任务的线程就赋值给了runner成员变量吗)
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    // 中断执行异步任务的线程runner
                    t.interrupt();
            } finally { // final state
                // 最后任务状态赋值为INTERRUPTED
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    // 【3】不管mayInterruptIfRunning为true还是false,此时都要调用finishCompletion方法唤醒阻塞的获取异步任务结果的线程并移除线程等待链表节点
    } finally {
        finishCompletion();
    }
    // 返回true
    return true;
}

??以上代码中,当异步任务状态state != NEW时,说明异步任务已经正常执行完或已经异常结束亦或已经被cancel,此时直接返回false;当异步任务状态state = NEW时,此时又根据mayInterruptIfRunning参数是否为true分为以下两种情况:
??1.当mayInterruptIfRunning = false时,此时任务状态state直接被赋值为CANCELLED,此时不会对执行异步任务的线程发出中断信号,值得注意的是这里对应的任务状态变化是NEW -> CANCELLED。
??2.当mayInterruptIfRunning = true时,此时会对执行异步任务的线程发出中断信号,值得注意的是这里对应的任务状态变化是NEW -> INTERRUPTING -> INTERRUPTED。

??最后不管mayInterruptIfRunning为true还是false,此时都要调用finishCompletion方法唤醒阻塞的获取异步任务结果的线程并移除线程等待链表节点

??从FutureTask.cancel源码中我们可以得出答案,该方法并不能真正中断正在执行异步任务的线程,只能对执行异步任务的线程发出中断信号。如果执行异步任务的线程处于sleep、wait或join的状态中,此时会抛出InterruptedException异常,该线程可以被中断;此外,如果异步任务需要在while循环执行的话,此时可以结合以下代码来结束异步任务线程,即执行异步任务的线程被中断时,此时Thread.currentThread().isInterrupted()返回true,不满足while循环条件因此退出循环,结束异步任务执行线程,如下代码:

public Integer call() throws Exception {
    while (!Thread.currentThread().isInterrupted()) {
        // 业务逻辑代码
        System.out.println("running...");

    }
    return 666;
}

??注意:调用了FutureTask.cancel方法,只要返回结果是true,假如异步任务线程虽然不能被中断,即使异步任务线程正常执行完毕,返回了执行结果,此时调用FutureTask.get方法也不能够获取异步任务执行结果,此时会抛出CancellationException异常。

??因为调用了FutureTask.cancel方法,只要返回结果是true,此时的任务状态为CANCELLED或INTERRUPTED,同时必然会执行finishCompletion方法,而finishCompletion方法会唤醒获取异步任务结果的线程等待列表的线程,而获取异步任务结果的线程唤醒后发现状态s >= CANCELLED,此时就会抛出CancellationException异常了。

文章来源:https://blog.csdn.net/xiaopang2020/article/details/135823259
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。