异步_Future与Callable
作者:互联网
当我们创建一个线程时,我们想获取线程运行完成后的结果,一般使用回调的方式。例如:
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任务一般运行在另一个线程中,其他线程来获取任务的结果值,会发生什么情况呢?
- 如果FutureTask任务已经完结,那么就返回结果值。
- 如果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任务
标签:异步,结果,state,Callable,任务,Future,线程,FutureTask 来源: https://www.cnblogs.com/xfeiyun/p/15870031.html