JUC之FutureTask

傷城~ 2022-08-31 11:05 149阅读 0赞

# **█ Runnable** #

在Java中，启动一个线程通常是如下的形式：

Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+" is running");
        }
    });
    thread.start();

或者使用Lambda表达式：

Thread thread = new Thread(()->
            System.out.println(Thread.currentThread().getName()+" is running"));
    thread.start();

借助于**Thread**，在创建Thread对象新创建一个线程的同时，指定一个**Runnable**类型的参数，然后调用**start**方法启动线程，这样在改线程得到调度的时候，会执行Runnable对象中的**run**方法。

public Thread(Runnable target) {
        init(null, target, "Thread-" + nextThreadNum(), 0);
    }

@FunctionalInterface
    public interface Runnable {
        public abstract void run();
    }

但是从Runnable接口中定义的run方法返回值为void可以看出，该方法的不支持有返回值的，也就是当线程执行完成后，是不会有返回值的。

--------------------

# **█ **Callable #

为了解决Runnable无法提供返回值的情况，Java又提供了这样一个类，其是可以获取返回值的，即**Callable**。

@FunctionalInterface
    public interface Callable<V> {
        V call() throws Exception;
    }

从Callable接口中定义的方法call来看，其是有返回值的，返回值类型是通过泛型来指定的，即Callable<String> call = new Callable()\{\}，返回值类型则为String。

但是此时有个问题，Thread类型并没有提供任何一个构造器用于接收Callable类型的参数，比如如下这种构造器是不存在，Thread是只支持传入类型为Runnable类型的参数。所以需要将Callable类型的对象转换成Runnable。

public Thread(Callable target) {
        ......
    }

--------------------

# **█ **FutureTask #

![20210724193954680.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

从上面Callable接口的定义来看，其并没有继承Runnable接口，因此Callable类型的对象是无法直接强制转换成Runnable的，即如下这种强制转换是会报类型转换异常的**java.lang.ClassCastException**：

Callable c = new Callable() {
        @Override
        public Object call() throws Exception {
            return null;
        }
    };
    // 强制转换将出错
    Runnable r = (Runnable) c;

那既然强制转换是行不通的，那咋办？借助中间类。为了解决将Callable类型的对象转换成Runnable类型的对象，**FutureTask**便诞生了。

从继承图上看见，FutureTask实现了Runnable接口，因此一个FutureTask对象也就是Runnable对象，FutureTask对象其是可以强制转换成Runnable对象的，因此将FutureTask对象作为参数传进Thread构造器是合法的：

FutureTask futureTask = new FutureTask(() -> "hello world");
    Thread t = new Thread(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
    }

FutureTask提供了两个重载的构造器，分别支持Callable类型的参数和Runnable类型的参数。不同的是对于接收Runnable类型的参数，还需要提供一个返回值，因为本身run方法是没有返回值的，因此需要手动指定一个。而Callable的call方法本身是支持有返回值的，所以不需要手动指定了。

### Executors.callable ###

当构造器参数类型为Runnable时，FutureTask要将其转换成Callable类型，FutureTask借助了**Executors**中提供的一个方法callable（关于Executors的详细介绍，请戳《[JUC之Executors][JUC_Executors]》）：

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

将Runnable对象封装到**RunnableAdapter**对象中，RunnableAdapter是Executors中提供的一个内部类，实现了Callable接口。

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;
        }
    }

### 字段： ###

*  **state**

private volatile int state;

标记当前FutureTask对象的状态。状态有如下可选值：

// 新生状态，call方法等待执行
    private static final int NEW          = 0;
    // call方法已执行完成，等待处理结果
    private static final int COMPLETING   = 1;
    // call方法是正常执行完成的
    private static final int NORMAL       = 2;
    // call方法在执行过程中发生了异常
    private static final int EXCEPTIONAL  = 3;
    // call方法被取消执行
    private static final int CANCELLED    = 4;
    // call方法正在被阻断
    private static final int INTERRUPTING = 5;
    // call方法已经被阻断。
    private static final int INTERRUPTED  = 6;

*  **callable**

private Callable<V> callable;

用于接收创建FutureTask对象时指定的Callable或Runnable对象。

*  **outcome**

private Object outcome;

FutureTask提供了一个get方法用于获取Callable的call方法运行完成对应的结果，如果发生了异常，则为异常。

*  **runner**

private volatile Thread runner;

执行Callable的call方法的线程。

*  **waiters**

private volatile WaitNode waiters;

等待获取执行结果的线程集合，使用单向链表将等待获取结果的线程存放起来。

**WaitNode**是FutureTask中提供的一个内部类：

static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }

# 方法： #

*  **run**

因为FutureTask实现了Runnable接口，那必定要实现run方法，当Thread对象被调度时，运行run方法：

public void run() {
        if (state != 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)
                handlePossibleCancellationInterrupt(s);
        }
    }

①state != NEW

第一个判断该任务的状态state，必须是NEW状态时才能得到运行。然后借助UNSAFE的CAS操作来保证在多线程并发竞争的过程中线程安全。

②Callable<V> c = callable; c.call();

获取callable对象，调用call方法。

③set

call方法正常执行，调用set方法记录结果。

protected void set(V v) {
        // 使用CAS保证线程安全，更改state状态为COMPLETING
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            // 将结果记录到outcome全局变量中
            outcome = v;
            // 将state状态更改为NORMAL
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

④setException

call方法在执行的过程中发生了异常。

protected void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

发生了异常后处理结果的setException与正常执行处理结果的set方法，区别在于发生异常将state状态设置成EXCEPTIONAL，而正常执行是将state状态设置成NORMAL。不管正常执行获取结果，还是发生了异常，都是将结果记录在全局变量outcome上。

*  **finishCompletion**

callable的call方法执行完成后，处理后续工作。唤醒正在等待获取结果的线程们起来继续运行（这些等待线程什么时候被阻塞休眠的，将在后面的方法体现）。实现逻辑就是遍历waiters单向链表，调用**LockSupport.unpark**唤醒WaitNode中对象的线程。

private void finishCompletion() {
        // assert state > COMPLETING;
        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
    }

done：

未实现的方法，提供给子类去重写自定义具体的后续逻辑。

protected void done() { }

*  **get**

在方法finishCompletion中有唤醒等待线程的操作，那让等待线程休眠的操作就是在get方法中进行的。get方法在call方法执行完成返回结果之前是会一直处于阻塞状态的。

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

①s <= COMPLETING

当state状态值小于COMPLETING时，即此时的state为NEW，也就是Callable还没有被执行。

**awaitDone**：

将当前调用get方法等待获取执行结果的线程添加到waiters单向链表上，并让当前线程休眠。使用**UNSAFE**.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);将当前线程添加到waiters单向链表上，使用**LockSupport.park**让当前线程休眠，执行call方法执行完成后，在finishCompletion方法中将线程唤醒。

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;
            }
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            else if (q == null)
                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);
        }
    }

*  **report**

当call方法执行完成后，返回执行结果。正常执行的话就返回call方法返回值，如果执行过程中发生了异常，封装在**ExecutionException**中抛出。如果方法执行被取消或者被阻断了，则抛出**CancellationException**异常。

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);
    }

*  **get(long timeout, TimeUnit unit)**

get()方法的重载，该方法可以设置阻塞等待的时长。get()方法是在call方法没有执行完成返回结果之前，方法会一直阻塞不会继续往下执行。get(long timeout, TimeUnit unit)是可以设置超时等待的，如果在指定的时长内call方法还没有执行运行返回结果，则该get方法不会继续等待，而是直接抛出**TimeoutException**异常。

public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null)
            throw new NullPointerException();
        int s = state;
        if (s <= COMPLETING &&
            (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
            throw new TimeoutException();
        return report(s);
    }

*  **cancel**

取消执行方法或者阻断执行。

public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == 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;
    }

--------------------

# **█ 使用例子** #

（为了使代码看上去更清晰，就不使用Lambda表达式了）

①运行Callable

FutureTask<String> f = new FutureTask(new Callable() {
        @Override
        public Object call() throws Exception {
            return "hello world";
        }
    });
    Thread t = new Thread(f);
    t.start();
    
    try {
        String result = f.get();
        System.out.println(result);
    } catch (Exception e) {
        e.printStackTrace();
    }

运行结果：

![20210724195050367.png][]

②运行Runnable

FutureTask<String> f = new FutureTask(new Runnable() {
        @Override
        public void run() {
            System.out.println("Runnable run");
        }
    }, "hello world Runnable");
    Thread t = new Thread(f);
    t.start();
    
    try {
        String result = f.get();
        System.out.println(result);
    } catch (Exception e) {
        e.printStackTrace();
    }

运行结果： ![20210724195138578.png][]

③运行时发生异常

让call方法在执行过程中发生异常。

FutureTask<String> f = new FutureTask(new Callable() {
        @Override
        public Object call() throws Exception {
            int a = 1/0;
            return "hello world";
        }
    });

运行结果： ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

只有去调用get方法获取结果的时候才会抛出异常，如果没有调用get方法，是无法感知到异常的发生的。

[20210724193954680.png]: /images/20220829/4543c8081d474c1faf6dac7170e63728.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20220829/fe0107cf26e446908d92e6061528b863.png
[JUC_Executors]: https://blog.csdn.net/weixin_50518271/article/details/118768318?spm=1001.2014.3001.5501
[20210724195050367.png]: /images/20220829/81f794dc384e491983cd74819e6d7215.png
[20210724195138578.png]: /images/20220829/9d226df4006348e6829f4a57901778c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20220829/1ae7b38748104d919f5064052bf94a65.png