这里我们主要探讨如何使用JDK中的Future实现异步编程,这包含
在Java并发包(JUC包)中Future代表着异步计算结果,Future中提供了一系列方法用来
Future类共有5个接口方法,下面我们来一一讲解
等待异步计算任务完成,并返回结果;
尝试取消任务的执行;
FutureTask代表了一个可被取消的异步计算任务,该类实现了Future接口,比如提供了启动和取消任务、查询任务是否完成、获取计算结果的接口。
public class AsyncFutureExample {
public static String doSomethingA() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("--- doSomethingA---");
return "TaskAResult";
}
public static String doSomethingB() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("--- doSomethingB---");
return "TaskBResult";
}
public static void main(String[] args) throws InterruptedException, ExecutionException {
long start = System.currentTimeMillis();
// 1.创建future任务
FutureTask<String> futureTask = new FutureTask<String>(() -> {
String result = null;
try {
result = doSomethingA();
} catch (Exception e) {
e.printStackTrace();
}
return result;
});
// 2.开启异步单元执行任务A
Thread thread = new Thread(futureTask, "threadA");
thread.start();
// 3.执行任务B
String taskBResult = doSomethingB();
// 4.同步等待线程A运行结束
String taskAResult = futureTask.get();
// 5.打印两个任务执行结果
System.out.println(taskAResult + " " + taskBResult);
System.out.println(System.currentTimeMillis() - start);
}
}
当然我们也可以把FutureTask提交到线程池来执行,使用线程池运行方式的代码如下:
// 0自定义线程池
private final static int AVALIABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
private final static ThreadPoolExecutor POOL_EXECUTOR = new ThreadPoolExecutor(AVALIABLE_PROCESSORS,
AVALIABLE_PROCESSORS * 2, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(5),
new ThreadPoolExecutor.CallerRunsPolicy());
public static void main(String[] args) throws InterruptedException, ExecutionException {
long start = System.currentTimeMillis();
// 1.创建future任务
FutureTask<String> futureTask = new FutureTask<String>(() -> {
String result = null;
try {
result = doSomethingA();
} catch (Exception e) {
e.printStackTrace();
}
return result;
});
// 2.开启异步单元执行任务A
POOL_EXECUTOR.execute(futureTask);
// 3.执行任务B
String taskBResult = doSomethingB();
// 4.同步等待线程A运行结束
String taskAResult = futureTask.get();
// 5.打印两个任务执行结果
System.out.println(taskAResult + " " + taskBResult);
System.out.println(System.currentTimeMillis() - start);
}
如上可知代码0创建了一个线程池,代码2添加异步任务到线程池,这里我们是调用了线程池的execute方法把futureTask提交到线程池的,其实下面代码与上面是等价的:
public static void main(String[] args) throws InterruptedException, ExecutionException {
long start = System.currentTimeMillis();
// 1.开启异步单元执行任务A
Future<String> futureTask = POOL_EXECUTOR.submit(() -> {
String result = null;
try {
result = doSomethingA();
} catch (Exception e) {
e.printStackTrace();
}
return result;
});
// 2.执行任务B
String taskBResult = doSomethingB();
// 3.同步等待线程A运行结束
String taskAResult = futureTask.get();
// 4.打印两个任务执行结果
System.out.println(taskAResult + " " + taskBResult);
System.out.println(System.currentTimeMillis() - start);
}
这里代码1调用了线程池的submit方法提交了一个任务到线程池,然后返回了一个futureTask对象。
由上图可知 FutureTask实现了Future接口的所有方法,并且实现了Runnable接口,所以其是可执行任务,可以投递到线程池或者线程来执行。
FutureTask中变量state是一个使用volatile关键字修饰(用来解决多线程下内存不可见问题 ,用来记录任务状态,任务状态枚举值如下:
任务可能的状态转换路径
从上述转换可知,任务最终只有四种终态:NORMAL、EXCEPTIONAL、CANCELLED、INTERRUPTED
,另外可知任务的状态值是从上到下递增的。
/** The underlying callable; nulled out after running */
private Callable<V> callable;
callable是有返回值的可执行任务,创建FutureTask对象时,可以通过构造函数传递该任务。
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
outcome是任务运行的结果,可以通过get系列方法来获取该结果。另外,outcome这里没有被修饰为volatile,是因为变量state已经被volatile修饰了,这里是借用volatile的内存语义来保证写入outcome时会把值刷新到主内存,读取时会从主内存读取,从而避免多线程下内存不可见问题
/** The thread running the callable; CASed during run() */
private volatile Thread runner;
runner变量,记录了运行该任务的线程,这个是在FutureTask的run方法内使用CAS函数设置的。
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;
waiters变量是一个WaitNode节点,是用Treiber stack实现的无锁栈,栈顶元素用waiters代表。栈用来记录所有等待任务结果的线程节点
其定义为:
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
可知其是一个简单的链表,用来记录所有等待结果被阻塞的线程。
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
由上述代码可知,构造函数内保存了传递到callable任务的callable变量,并且将任务状态设置为NEW,这里由于state为volatile修饰,所以写入state的值可以保证callable的写入也会被刷入主内存,以避免多线程下内存不可见问题。
另外还有一个构造函数:
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW;
}
该函数传入一个Runnable类型的任务,由于该任务是不具有返回值的,所以这里使用Executors.callable方法进行适配,适配为Callable类型的任务。
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
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;
}
}
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;//state变量的偏移地址
private static final long runnerOffset;//runner变量的偏移地址
private static final long waitersOffset;//waiters变量的偏移地址
static {
try {
//获取UNSAFE的实例
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
//获取变量state的偏移地址
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
//获取变量runner的偏移地址
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
//获取变量waiters变量的偏移地址
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
如上代码分别获取了FutureTask中几个变量在FutureTask对象内的内存地址偏移量,以便实现用UNSAFE的CAS操作来操作这些变量。
该方法是任务的执行体,线程是调用该方法来具体运行任务的,如果任务没有被取消,则该方法会运行任务,并且将结果设置到outcome变量中,其代码如下:
public void run() {
//1.如果任务不是初始化的NEW状态,或者使用CAS设置runner为当前线程失败,则直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
//2.如果任务不为null,并且任务状态为NEW,则执行任务
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
//2.1执行任务,如果OK则设置ran标记为true
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
//2.2执行任务出现异常,则标记false,并且设置异常
result = null;
ran = false;
setException(ex);
}
//3.任务执行正常,则设置结果
if (ran)
set(result);
}
} finally {
runner = null;
int s = state;
//4.为了保证调用cancel(true)的线程在该run方法返回前中断任务执行的线程
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
private void handlePossibleCancellationInterrupt(int s) {
//为了保证调用cancel在该run方法返回前中断任务执行的线程
//这里使用Thread.yield()让run方法执行线程让出CPU执行权,以便让
//cancel(true)的线程执行cancel(true)中的代码中断任务线程
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
}
protected void setException(Throwable t) {
//2.2.1
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
//2.2.1.1
finishCompletion();
}
}
由上述代码可知,使用CAS尝试设置任务状态state为COMPLETING,如果CAS成功,则把异常信息设置到outcome变量,并且设置任务状态为EXCEPTIONAL终止状态,然后调用finishCompletion,其代码如下:
private void finishCompletion() {
//a遍历链表节点
for (WaitNode q; (q = waiters) != null;) {
//a.1 CAS设置当前waiters节点为null
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
//a.1.1
for (;;) {
//唤醒当前q节点对应的线程
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
//获取q的下一个节点
WaitNode next = q.next;
if (next == null)
break;
q.next = null; //help gc
q = next;
}
break;
}
}
// b所有阻塞的线程都被唤醒后,调用done方法
done();
callable = null; // callable设置为null
}
上述代码比较简单,即当任务已经处于终态后,激活waiters链表中所有由于等待获取结果而被阻塞的线程,并从waiters链表中移除它们,等所有由于等待该任务结果的线程被唤醒后,调用done()方法,done默认实现为空实现。
上面我们讲了当任务执行过程中出现异常后的处理方法,下面我们看下代码3,了解当任务是正常执行完毕后set(result)的实现:
protected void set(V v) {
//3.1
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
如代码3.1所示,使用CAS尝试设置任务状态state为COMPLETING,如果CAS成功,则把任务结果设置到outcome变量,并且将任务状态设置为NORMAL终止状态,然后调用finishCompletion唤醒所有因为等待结果而被阻塞的线程。
其代码如下:
public V get() throws InterruptedException, ExecutionException {
//1.获取状态,如有需要则等待
int s = state;
if (s <= COMPLETING)
//等待任务终止
s = awaitDone(false, 0L);
//2.返回结果
return report(s);
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
//1.1超时时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
//1.2 循环,等待任务完成
for (;;) {
//1.2.1任务被中断,则移除等待线程节点,抛出异常
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
//1.2.2 任务状态>COMPLETING说明任务已经终止
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//1.2.3任务状态为COMPLETING
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//1.2.4为当前线程创建节点
else if (q == null)
q = new WaitNode();
//1.2.5 添加当前线程节点到链表
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
//1.2.6 设置了超时时间
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
//1.2.7没有设置超时时间
else
LockSupport.park(this);
}
}
cancel方法的代码如下:
public boolean cancel(boolean mayInterruptIfRunning) {
//1.如果任务状态为New则使用CAS设置任务状态为INTERRUPTING或者CANCELLED
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
//2.如果设置了中断正常执行任务线程,则中断
try {
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
//3.移除并激活所有因为等待结果而被阻塞的线程
finishCompletion();
}
return true;
}
当我们创建一个FutureTask时,其任务状态初始化为NEW,当我们把任务提交到线程或者线程池后,会有一个线程来执行该FutureTask任务,具体是调用其run方法来执行任务。
在任务执行过程中,我们可以在其他线程调用FutureTask的get()方法来等待获取结果,如果当前任务还在执行,则调用get的线程会被阻塞然后放入FutureTask内的阻塞链表队列;
多个线程可以同时调用get方法,这些线程可能都会被阻塞并放到阻塞链表队列中。
当任务执行完毕后会把结果或者异常信息设置到outcome变量,然后会移除和唤醒FutureTask内阻塞链表队列中的线程节点,进而这些由于调用FutureTask的get方法而被阻塞的线程就会被激活。
FutureTask虽然提供了用来检查任务是否执行完成、等待任务执行结果、获取任务执行结果的方法,但是这些特色并不足以让我们写出简洁的并发代码,比如它并不能清楚地表达多个FutureTask之间的关系。
另外,为了从Future获取结果,我们必须调用get()方法,而该方法还是会在任务执行完毕前阻塞调用线程,这明显不是我们想要的。
我们真正想要的是:
为了克服FutureTask的局限性,以及满足我们对异步编程的需要,JDK8中提供了CompletableFuture。