• Java（八）FutureTask源码

    1.简述

    FutureTask位于java.util.concurrent.FutureTask 类中，主要用于处理异步任务，多线程编程，以及并发控制。

    1. 表示异步任务：FutureTask 是 Java 中表示异步任务的一种方式。它允许将一个任务提交给线程池执行，并在需要时获取任务的结果。

    2. 任务执行的封装：FutureTask 封装了一个任务（Callable 或 Runnable），使得你可以将任务的执行和结果的获取解耦。这允许你在一个线程中提交任务，然后在另一个线程中等待任务完成并获取结果，而不需要手动管理线程。

    3. 获取异步任务的结果：通过 FutureTask，你可以通过调用 get() 方法来获取异步任务的结果。如果任务尚未完成，get() 方法将会阻塞，直到任务完成。这对于需要等待任务结果的情况非常有用。

    4. 处理任务状态：FutureTask 提供了方法来查询任务的状态，如 isDone() 来检查任务是否已完成，isCancelled() 来检查任务是否被取消。这有助于你了解任务的执行状态。

    5. 取消任务：你可以通过调用 cancel(boolean mayInterruptIfRunning) 方法来取消任务的执行。mayInterruptIfRunning 参数允许你指定是否中断正在执行任务的线程。

    6. 异常处理：FutureTask 允许捕获任务执行过程中抛出的异常，以便更好地处理错误情况。

    7. 并发控制：FutureTask 可以与 ExecutorService 结合使用，帮助管理并发执行的任务。它使得你可以异步执行多个任务，然后在需要时获取它们的结果。

    8. 提高程序的响应性：通过使用 FutureTask，可以在程序执行期间执行耗时的任务，而不会阻塞主线程。这有助于提高程序的响应性，特别是在涉及到 I/O 操作或长时间计算的情况下。

    FutureTask 类的存在有助于处理异步任务和并发编程，提供了一种方便的方式来提交和管理任务，获取任务的结果以及处理任务的状态和异常。这对于构建高效的多线程应用程序和异步操作非常重要。

    2.接口类

    Callable接口

    对于需要执行的任务需要实现Callable接口，Callable接口定义如下:

    public interface Callable {
    /**
      * 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;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    可以看到Callable是个泛型接口，泛型V就是要call()方法返回的类型。Callable接口和Runnable接口很像，都可以被另外一个线程执行，但是正如前面所说的，Runnable不会返回数据也不能抛出异常。

    Future接口

    Future接口代表异步计算的结果，通过Future接口提供的方法可以查看异步计算是否执行完成，或者等待执行结果并获取执行结果，同时还可以取消执行。Future接口的定义如下:

    public interface Future {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
 }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • cancel():cancel()方法用来取消异步任务的执行。如果异步任务已经完成或者已经被取消，或者由于某些原因不能取消，则会返回false。如果任务还没有被执行，则会返回true并且异步任务不会被执行。如果任务已经开始执行了但是还没有执行完成，若mayInterruptIfRunning为true，则会立即中断执行任务的线程并返回true，若mayInterruptIfRunning为false，则会返回true且不会中断任务执行线程。
    • isCanceled():判断任务是否被取消，如果任务在结束(正常执行结束或者执行异常结束)前被取消则返回true，否则返回false。
    • isDone():判断任务是否已经完成，如果完成则返回true，否则返回false。需要注意的是：任务执行过程中发生异常、任务被取消也属于任务已完成，也会返回true。
    • get():获取任务执行结果，如果任务还没完成则会阻塞等待直到任务执行完成。如果任务被取消则会抛出CancellationException异常，如果任务执行过程发生异常则会抛出ExecutionException异常，如果阻塞等待过程中被中断则会抛出InterruptedException异常。
    • get(long timeout,Timeunit unit):带超时时间的get()版本，如果阻塞等待过程中超时则会抛出TimeoutException异常。

    FutureTask

    Future只是一个接口，不能直接用来创建对象，FutureTask是Future的实现类，
    FutureTask的继承图如下:

    可以看到,FutureTask实现了RunnableFuture接口，则RunnableFuture接口继承了Runnable接口和Future接口，所以FutureTask既能当做一个Runnable直接被Thread执行，也能作为Future用来得到Callable的计算结果。

    使用

    FutureTask一般配合ExecutorService来使用，也可以直接通过Thread来使用。

    package com.beautyboss.slogen.callback;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;


public class CallDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        /**
         * 第一种方式:Future + ExecutorService
         * Task task = new Task();
         * ExecutorService service = Executors.newCachedThreadPool();
         * Future future = service.submit(task1);
         * service.shutdown();
         */


        /**
         * 第二种方式: FutureTask + ExecutorService
         * ExecutorService executor = Executors.newCachedThreadPool();
         * Task task = new Task();
         * FutureTask futureTask = new FutureTask(task);
         * executor.submit(futureTask);
         * executor.shutdown();
         */

        /**
         * 第三种方式:FutureTask + Thread
         */

        // 2. 新建FutureTask,需要一个实现了Callable接口的类的实例作为构造函数参数
        FutureTask<Integer> futureTask = new FutureTask<Integer>(new Task());
        // 3. 新建Thread对象并启动
        Thread thread = new Thread(futureTask);
        thread.setName("Task thread");
        thread.start();

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Thread [" + Thread.currentThread().getName() + "] is running");

        // 4. 调用isDone()判断任务是否结束
        if (!futureTask.isDone()) {
            System.out.println("Task is not done");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        int result = 0;
        try {
            // 5. 调用get()方法获取任务结果,如果任务没有执行完成则阻塞等待
            result = futureTask.get();
        } catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println("result is " + result);

    }

    // 1. 继承Callable接口,实现call()方法,泛型参数为要返回的类型
    static class Task implements Callable<Integer> {

        @Override
        public Integer call() throws Exception {
            System.out.println("Thread [" + Thread.currentThread().getName() + "] is running");
            int result = 0;
            for (int i = 0; i < 100; ++i) {
                result += i;
            }

            Thread.sleep(3000);
            return result;
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85

    3.重点方法

    构造函数

    先从FutureTask的构造函数看起，FutureTask有两个构造函数，其中一个如下：

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    这个构造函数会把传入的Callable变量保存在this.callable字段中，该字段定义为private Callable callable;用来保存底层的调用，在被执行完成以后会指向null,接着会初始化state字段为NEW。state字段用来保存FutureTask内部的任务执行状态，一共有7中状态，每种状态及其对应的值如下：

    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;

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    其中需要注意的是state是volatile类型的，也就是说只要有任何一个线程修改了这个变量，那么其他所有的线程都会知道最新的值。

    为了后面更好的分析FutureTask的实现，这里有必要解释下各个状态。

    • NEW:表示是个新的任务或者还没被执行完的任务。这是初始状态。
    • COMPLETING:任务已经执行完成或者执行任务的时候发生异常，但是任务执行结果或者异常原因还没有保存到outcome字段(outcome字段用来保存任务执行结果，如果发生异常，则用来保存异常原因)的时候，状态会从NEW变更到COMPLETING。但是这个状态会时间会比较短，属于中间状态。
    • NORMAL:任务已经执行完成并且任务执行结果已经保存到outcome字段，状态会从COMPLETING转换到NORMAL。这是一个最终态。
    • EXCEPTIONAL:任务执行发生异常并且异常原因已经保存到outcome字段中后，状态会从COMPLETING转换到EXCEPTIONAL。这是一个最终态。
    • CANCELLED:任务还没开始执行或者已经开始执行但是还没有执行完成的时候，用户调用了cancel(false)方法取消任务且不中断任务执行线程，这个时候状态会从NEW转化为CANCELLED状态。这是一个最终态。
    • INTERRUPTING: 任务还没开始执行或者已经执行但是还没有执行完成的时候，用户调用了cancel(true)方法取消任务并且要中断任务执行线程但是还没有中断任务执行线程之前，状态会从NEW转化为INTERRUPTING。这是一个中间状态。
    • INTERRUPTED:调用interrupt()中断任务执行线程之后状态会从INTERRUPTING转换到INTERRUPTED。这是一个最终态。

    有一点需要注意的是，所有值大于COMPLETING的状态都表示任务已经执行完成(任务正常执行完成，任务执行异常或者任务被取消)。

    另外一个构造函数如下，

     public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
}

    • 1
    • 2
    • 3
    • 4

    这个构造函数会把传入的Runnable封装成一个Callable对象保存在callable字段中，同时如果任务执行成功的话就会返回传入的result。这种情况下如果不需要返回值的话可以传入一个null。

    顺带看下Executors.callable()这个方法，这个方法的功能是把Runnable转换成Callable，代码如下:

    public static  Callable callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter(task, result);
}

    • 1
    • 2
    • 3
    • 4
    • 5

    可以看到这里采用的是适配器模式，调用RunnableAdapter(task, result)方法来适配，实现如下:

    static final class RunnableAdapter implements Callable {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    这个适配器很简单，就是简单的实现了Callable接口，在call()实现中调用Runnable.run()方法，然后把传入的result作为任务的结果返回。

    在new了一个FutureTask对象之后，接下来就是在另一个线程中执行这个Task,无论是通过直接new一个Thread还是通过线程池，执行的都是run()方法，接下来就看看run()方法的实现。

    run()

    run()方法实现如下:

    public void run() {
    // 1. 状态如果不是NEW，说明任务或者已经执行过，或者已经被取消，直接返回
    // 2. 状态如果是NEW，则尝试把当前执行线程保存在runner字段中
    // 如果赋值失败则直接返回
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                // 3. 执行任务
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                // 4. 任务异常
                setException(ex);
            }
            if (ran)
                // 4. 任务正常执行完毕
                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;
        // 5. 如果任务被中断，执行中断处理
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39

    run()方法首先会

    1. 判断当前任务的state是否等于NEW,如果不为NEW则说明任务或者已经执行过，或者已经被取消，直接返回。
    2. 如果状态为NEW则接着会通过unsafe类把任务执行线程引用CAS的保存在runner字段中，如果保存失败，则直接返回。
    3. 执行任务。
    4. 如果任务执行发生异常，则调用setException()方法保存异常信息。setException()方法如下：
    protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
       UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
         finishCompletion();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    在setException()方法中

    • 首先会CAS的把当前的状态从NEW变更为COMPLETING状态。
    • 把异常原因保存在outcome字段中，outcome字段用来保存任务执行结果或者异常原因。
    • CAS的把当前任务状态从COMPLETING变更为EXCEPTIONAL。这个状态转换对应着上图中的二。
    • 调用finishCompletion()。关于这个方法后面在分析。
    1. 如果任务成功执行则调用set()方法设置执行结果，该方法实现如下:
     protected void set(V v) {
  if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
 }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    这个方法跟上面分析的setException()差不多，

    • 首先会CAS的把当前的状态从NEW变更为COMPLETING状态。
    • 把任务执行结果保存在outcome字段中。
    • CAS的把当前任务状态从COMPLETING变更为NORMAL。这个状态转换对应着上图中的一。
    • 调用finishCompletion()。

    发起任务线程跟执行任务线程通常情况下都不会是同一个线程，在任务执行线程执行任务的时候，任务发起线程可以查看任务执行状态、获取任务执行结果、取消任务等等操作，接下来分析下这些操作。

    get()

    任务发起线程可以调用get()方法来获取任务执行结果，如果此时任务已经执行完毕则会直接返回任务结果，如果任务还没执行完毕，则调用方会阻塞直到任务执行结束返回结果为止。get()方法实现如下:

     public V get() throws InterruptedException, ExecutionException {
     int s = state;
     if (s <= COMPLETING)
         s = awaitDone(false, 0L);
    return report(s);
 }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    get()方法实现比较简单，会

    1. 判断任务当前的state <= COMPLETING是否成立。前面分析过，COMPLETING状态是任务是否执行完成的临界状态。
    2. 如果成立，表明任务还没有结束(这里的结束包括任务正常执行完毕，任务执行异常，任务被取消)，则会调用awaitDone()进行阻塞等待。
    3. 如果不成立表明任务已经结束，调用report()返回结果。

    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 (;;) {
        // 1. 判断阻塞线程是否被中断,如果被中断则在等待队
        // 列中删除该节点并抛出InterruptedException异常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }

        // 2. 获取当前状态，如果状态大于COMPLETING
        // 说明任务已经结束(要么正常结束，要么异常结束，要么被取消)
        // 则把thread显示置空，并返回结果
        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        // 3. 如果状态处于中间状态COMPLETING
        // 表示任务已经结束但是任务执行线程还没来得及给outcome赋值
        // 这个时候让出执行权让其他线程优先执行
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        // 4. 如果等待节点为空，则构造一个等待节点
        else if (q == null)
            q = new WaitNode();
        // 5. 如果还没有入队列，则把当前节点加入waiters首节点并替换原来waiters
        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;
            }
            // 6. 阻塞等待特定时间
            LockSupport.parkNanos(this, nanos);
        }
        else
            // 6. 阻塞等待直到被其他线程唤醒
            LockSupport.park(this);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51

    awaitDone()中有个死循环，每一次循环都会

    1. 判断调用get()的线程是否被其他线程中断，如果是的话则在等待队列中删除对应节点然后抛出InterruptedException异常。
    2. 获取任务当前状态，如果当前任务状态大于COMPLETING则表示任务执行完成，则把thread字段置null并返回结果。
    3. 如果任务处于COMPLETING状态，则表示任务已经处理完成(正常执行完成或者执行出现异常)，但是执行结果或者异常原因还没有保存到outcome字段中。这个时候调用线程让出执行权让其他线程优先执行。
    4. 如果等待节点为空，则构造一个等待节点WaitNode。
    5. 如果第四步中新建的节点还没如队列，则CAS的把该节点加入waiters队列的首节点。
    6. 阻塞等待。

    假设当前state=NEW且waiters为NULL,也就是说还没有任何一个线程调用get()获取执行结果，这个时候有两个线程threadA和threadB先后调用get()来获取执行结果。再假设这两个线程在加入阻塞队列进行阻塞等待之前任务都没有执行完成且threadA和threadB都没有被中断的情况下(因为如果threadA和threadB在进行阻塞等待结果之前任务就执行完成或线程本身被中断的话，awaitDone()就执行结束返回了)，执行过程是这样的，以threadA为例:

    1. 第一轮for循环，执行的逻辑是q == null,所以这时候会新建一个节点q。第一轮循环结束。
    2. 第二轮for循环，执行的逻辑是!queue，这个时候会把第一轮循环中生成的节点的netx指针指向waiters，然后CAS的把节点q替换waiters。也就是把新生成的节点添加到waiters链表的首节点。如果替换成功，queued=true。第二轮循环结束。
    3. 第三轮for循环，进行阻塞等待。要么阻塞特定时间，要么一直阻塞知道被其他线程唤醒。

    在threadA和threadB都阻塞等待之后的waiters结果

    cancel(boolean)

    用户可以调用cancel(boolean)方法取消任务的执行，cancel()实现如下:

     1 public boolean cancel(boolean mayInterruptIfRunning) {
 2     // 1. 如果任务已经结束，则直接返回false
 3     if (state != NEW)
 4         return false;
 5     // 2. 如果需要中断任务执行线程
 6     if (mayInterruptIfRunning) {
 7         // 2.1. 把任务状态从NEW转化到INTERRUPTING
 8         if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
 9             return false;
10         Thread t = runner;
11         // 2.2. 中断任务执行线程
12         if (t != null)
13             t.interrupt();
14         // 2.3. 修改状态为INTERRUPTED
15         UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
16     }
17     // 3. 如果不需要中断任务执行线程，则直接把状态从NEW转化为CANCELLED
18     else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
19         return false;
20     // 4.
21     finishCompletion();
22     return true;
23 }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    cancel()方法会做下面几件事:

    ​ 1 .判断任务当前执行状态，如果任务状态不为NEW，则说明任务或者已经执行完成，或者执行异常，不能被取消，直接返回false表示执行失败。

    1. 判断需要中断任务执行线程，则
    • 把任务状态从NEW转化到INTERRUPTING。这是个中间状态。
    • 中断任务执行线程。
    • 修改任务状态为INTERRUPTED。这个转换过程对应上图中的四。

    1. 如果不需要中断任务执行线程，直接把任务状态从NEW转化为CANCELLED。如果转化失败则返回false表示取消失败。这个转换过程对应上图中的四。

    2. 调用finishCompletion()。

  • 相关阅读:
    vue3 父子组件间的通信
    Web基础与HTTP协议
    使用集成开发环境来开发Go项目
    何去何从的并行计算
    JVM的三种常见GC：Minor GC、Major GC与Full GC
    MySQL-2（14000字详解）
    Java面向对象七大原则以及设计模式单例模式和工厂模式简单工厂模式
    Java 反射
    知识体系图谱
    docker.6-Dockerfile精讲及新型容器镜像构建技术
  • 原文地址：https://blog.csdn.net/qq_42583549/article/details/133847087