异步_Future与Callable

2023-10-24 16:35:52

当我们创建一个线程时，我们想获取线程运行完成后的结果，一般使用回调的方式。例如：

interface Callable {
    void call(int num);
}
public class FutureTest {


    public static void main(String[] args)  {

        Callable callable = new Callable() {
            @Override
            public void call(int num) {
                System.out.println("线程运行结果值 num=="+num);
            }
        };

        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("线程"+Thread.currentThread().getName()+" 开始");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                callable.call(100);
                System.out.println("线程"+Thread.currentThread().getName()+" 结束");
            }
        }, "t1").start();

        System.out.println("主线程结束");

    }
}

运行结果：

主线程结束
线程t1 开始
线程运行结果值 num==100
线程t1 结束

这种方式的实现有三个缺点：

  • 必须要创建回调接口。而且线程运行中可能产生异常，那么回调接口至少包含成功回调和错误回调两个方法。
  • 当线程运行后，只能等到线程回调接口，本身我们没有办法进行取消操作。
  • 如果要重复获取同样线程运行结果的值，还是只能重新运行线程。当然你也可以使用一个变量缓存结果值。

那么有没有一种优化的方式呢？java中提供了Future与Callable的模式。

Future与Callable

想一想我们为什么需要使用回调呢？那是因为结果值是由另一线程计算的，当前线程是不知道结果值什么时候计算完成，所以它传递一个回调接口给计算线程，当计算完成时，调用这个回调接口，回传结果值。

另一种方式就是当前线程去获取值的时候，如果计算结果的线程还没有计算完毕，那么当前线程就等待，直到计算完毕，会唤醒等待结果的线程。如果已经计算完毕了，就直接获取结果值。

所以我们来看一看Future与Callable是怎么实现的？

Callable接口

public interface Callable<V> {
    // 子类复写这个方法，一般都是耗时操作，并返回结果值
    V call() throws Exception;
}

相当于Runnable的run方法，一般都是耗时操作，但是不一样的是，这个方法会返回结果值。

Future接口

要想解决我们上面说的三个缺点，那么Future接口至少有两个方法：

  • get()方法：获取结果值，如果当前Future还没有结束，那么当前线程就等待，直到Future运行结束，那么会唤醒等待结果值的线程的。
  • cancel()方法：取消当前的Future。会唤醒所有等待结果值的线程。
public interface Future<V> {

    /**
     * 取消当前的Future。会唤醒所有等待结果值的线程，抛出CancellationException异常
     * @param mayInterruptIfRunning 是否中断 计算结果值的那个线程
     * @return 返回true表示取消成功
     */
    boolean cancel(boolean mayInterruptIfRunning);

    // 当前的Future是否被取消，返回true表示已取消。
    boolean isCancelled();

    // 当前Future是否已结束。包括运行完成、抛出异常以及取消，都表示当前Future已结束
    boolean isDone();

    // 获取Future的结果值。如果当前Future还没有结束，那么当前线程就等待，
    // 直到Future运行结束，那么会唤醒等待结果值的线程的。
    V get() throws InterruptedException, ExecutionException;

    // 获取Future的结果值。与get()相比较多了允许设置超时时间。
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

可以看出Callable代表任务，Future代表结果值，那它们是如何结合操作的呢？下面就分析一个非常有代表的类FutureTask。

FutureTask类

FutureTask实现RunnableFuture接口，那么这个RunnableFuture接口的作用是什么呢？

// 继承自Runnable和Future接口。
public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

RunnableFuture接口的作用就是它的子类可以当做Runnable接口使用，那么创建一个新线程的时候，就可以使用它的实例作为参数了。

FutureTask的状态

根据Future接口，我们至少判断FutureTask有三种状态，新建状态，完成状态和取消状态。但是因为还有异常和中断操作，状态多了几种。

// 表示FutureTask当前的状态
private volatile int state;
// NEW 新建状态，表示这个FutureTask还没有开始运行
private static final int NEW          = 0;
// COMPLETING 完成状态， 表示FutureTask任务已经计算完毕了，
// 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
private static final int COMPLETING   = 1;
// FutureTask任务完结，正常完成，没有发生异常
private static final int NORMAL       = 2;
// FutureTask任务完结，因为发生异常。
private static final int EXCEPTIONAL  = 3;
// FutureTask任务完结，因为取消任务
private static final int CANCELLED    = 4;
// FutureTask任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求。
private static final int INTERRUPTING = 5;
// FutureTask任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求。
private static final int INTERRUPTED  = 6;

状态说明：

  • NEW 新建状态: 是FutureTask刚创建时的状态，也就是说没有调用过这个FutureTask任务的run与runAndReset运行方法和cancel取消方法。
  • COMPLETING 完成状态: 表示FutureTask任务的run运行方法已经执行完成，有两种方式表示完成：一是run方法正常执行完毕，二是run方法抛出异常。在set和setException方法中设置。
  • NORMAL 正常完结状态：表示FutureTask任务顺利地执行完毕，得到运算结果，一般在set方法中设置。
  • EXCEPTIONAL 异常完结状态：表示FutureTask任务在运行过程中抛出异常，任务也表示完结，一般在setException方法中设置。
  • CANCELLED 取消状态：表示正常取消任务，但是不会对正在运行FutureTask任务的线程产生任何影响，也就是说FutureTask任务仍然会执行完毕，但是结果值无法设置了，因为任务已取消。通过调用cancel方法设置，参数mayInterruptIfRunning是false。
  • INTERRUPTING和INTERRUPTED：表示中断取消任务，也就是说它会调用正在运行FutureTask任务的线程的interrupt方法，进行中断请求。通过调用cancel方法设置，参数mayInterruptIfRunning是true。

这六种状态分为四类：

  • NEW状态：这个FutureTask任务还没有做任何操作，只有这个状态下的任务，我们可以调用run方法运行任务，或者调用cancel方法取消任务。
  • COMPLETING状态：表示run方法运行完成，但是还有一些后序操作没有执行，比如唤醒正在等待任务结果的线程。切记这个状态FutureTask任务没有完结，不能返回结果值。
  • NORMAL和EXCEPTIONAL状态：任务完结，可能是正常完结也可能是异常完结。
  • CANCELLED、INTERRUPTING和INTERRUPTED状态：取消任务，任务也是完结。区别就是CANCELLED只是改变状态，而INTERRUPTING不仅改变状态，还会对正在运行FutureTask任务的线程进行中断。INTERRUPTED表示已经调用了中断请求。

因为状态是int变量，所以可以根据大小来判断。state > COMPLETING：表示FutureTask任务已经完结了，可以返回结果值。state >= CANCELLED：表示任务已经取消了，也表示任务完结。

对于FutureTask任务，我们首先关心它的结果值，那么我们先看看它获取结果值的方法，从Future接口来看，它有两个获取结果值的方法。

获取结果值get方法

思考一下，因为FutureTask任务一般运行在另一个线程中，其他线程来获取任务的结果值，会发生什么情况呢？

  1. 如果FutureTask任务已经完结，那么就返回结果值。
  2. 如果FutureTask任务没有完结，那么当前线程就应该等待，直到任务运行完结，会唤醒这个等待结果的线程，返回结果值。
public V get() throws InterruptedException, ExecutionException {
    int s = state;
    /**
     * 状态小于等于COMPLETING，表示FutureTask任务还没有完结，
     * 所以调用awaitDone方法，让当前线程等待
     */
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    // 返回结果值或者抛出异常
    return report(s);
}

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    /**
     * 状态小于等于COMPLETING，表示FutureTask任务还没有完结，
     * 所以调用awaitDone方法，让当前线程等待。
     * 与get()不同的是，如果到了规定时间，任务状态仍然是小于等于COMPLETING，
     * 那么就抛出TimeoutException超时异常
     */
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    // 返回结果值或者抛出异常
    return report(s);
}

get方法流程与我们上面介绍的一样。s <= COMPLETING表示FutureTask任务没有完结，awaitDone方法可以将当前线程插入等待结果的线程队列中去，并阻塞当前线程。report方法返回结果值或者抛出异常。

/**
 * 返回运行结果，或者抛出异常。这个两种情况都表示FutureTask完结了
 */
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
    Object x = outcome;
    // 表示正常完结状态，所以返回结果值
    if (s == NORMAL)
        return (V)x;
    // 大于或等于CANCELLED，都表示手动取消FutureTask任务，
    // 所以抛出CancellationException异常
    if (s >= CANCELLED)
        throw new CancellationException();
    // 否则就是运行过程中，发生了异常，这里就抛出这个异常
    throw new ExecutionException((Throwable)x);
}

awaitDone方法我们后面在介绍，report方法就是根据传入的状态值s，来决定是抛出异常，还是返回结果值。

运行FutureTask任务

运行FutureTask任务也很简单，只要调用FutureTask任务的run方法，那么这个线程也是运行FutureTask任务的线程，取消任务时，可能会中断这个线程。

// 开始运行FutureTask任务
public void run() {
    // 如果状态state不是NEW，或者设置runner值失败
    // 表示有别的线程在此之前调用run方法，并成功设置了runner值
    // 保证了只有一个线程可以运行try 代码块中的代码。
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        // 使用一个变量c记录callable，防止多线程情况下，
        // callable直接被设置为null出现问题
        Callable<V> c = callable;
        // 只有c不为null且状态state为NEW的情况，
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                // 调用callable的call方法，并返回结果
                result = c.call();
                // 运行成功
                ran = true;
            } catch (Throwable ex) {
                // 发生异常
                result = null;
                ran = false;
                // 设置异常
                setException(ex);
            }
            // 如果运行成功，则设置结果
            if (ran)
                set(result);
        }
    } finally {
        runner = null;
        int s = state;
        // 当状态大于或等于INTERRUPTING，调用handlePossibleCancellationInterrupt方法，
        // 等待别的线程将状态设置成INTERRUPTED
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

其实run方法作用非常简单，就是调用callable的call方法返回结果值result，根据是否发生异常，调用set(result)或setException(ex)方法表示FutureTask任务完结。

不过因为FutureTask任务都是在多线程环境中使用，所以要注意并发冲突问题。注意在run方法中，我们没有使用synchronized代码块或者Lock来解决并发问题，而是使用了CAS这个乐观锁来实现并发安全，保证只有一个线程能运行FutureTask任务。

protected void set(V v) {
    // 调用CAS函数，将状态state从NEW改成COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        // 延迟设置，将状态改成NORMAL，
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        // 调用finishCompletion唤醒所有等待结果的线程
        finishCompletion();
    }
}

protected void setException(Throwable t) {
    // 调用CAS函数，将状态state从NEW改成COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        // 延迟设置，将状态改成EXCEPTIONAL
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        // 调用finishCompletion唤醒所有等待结果的线程
        finishCompletion();
    }
}

set与setException方法流程几乎一样：

  • 先使用CAS函数将状态state从NEW变成COMPLETING，防止多线程冲突。
  • 使用putOrderedInt方法设置状态state是NORMAL或EXCEPTIONAL。
  • 调用finishCompletion方法唤醒所有等待结果的线程。

注：putOrderedInt方法的意义。因为state变量是被volatile关键字修饰，根据我们以前的讲解，那么它会给state变量加一个内存屏障，来保证state变量的可见性和有序性，这样会消耗一些性能。而putOrderedInt方法的意义，就是通过它来设置volatile修饰的变量，会取消这个内存屏障，也就是像普通变量一样了，不保证可见性了。

取消FutureTask任务

 

上一篇：多线程


下一篇：纯前端JSON文件编辑器[0]