该篇文章与线程池的关系不是很大，由于它与FutureTask的实现非常的相似，因此放在了线程池系列。在学习ListenableFuture时一定要对比着FutureTask的相关知识点学习，了解两者的共同点、区别、适应场景。本节将详细讲解ListenableFure的默认实现AbstractFuture类，并结合FutureTask展开一些列的问题讨论。有关FutureTask的知识，请阅读我的另一篇文章《线程池系列一》-FutureTask原理讲解与源码剖析。本节标题与《线程池系列六》-Guava ListenableFutureTask非常相似，但是这两篇文章完全不同，ListenableFutureTask是基于FutureTask的实现，而本文主要讲解的是ListenableFuture的默认实现AbstractFuture类的使用
ListenableFuture<V>
ListenableFuture<V>接口继承Future<V>接口。Future接口是JDK定义的接口，其定义了与任务操作相关的方法，例如任务的取消：cancel()，判断任务的状态：isCancelled()、isDone()，获取任务结果：get()等。
ListenableFuture<V>是Guava（google提供的一个java开发包）定义的接口，其在Future接口的基础上添加了addListener()方法，源码如下：
void addListener(Runnable listener, Executor executor);
该方法主要用于给Future任务添加监听任务，需要主要的是：
- listener任务必须指定执行该任务的线程池executor(该executor一定不要与执行Future任务的线程池是同一个，否则会出现死锁情况)
- 监听任务执行的时机为Future任务执行完成(包括正确执行完成和任务抛异常终止)或者被取消。
AbstractFuture<V>内部类Sync
AbstractFuture的实现与FutureTask的实现非常相似，FutureTask使用unsafe包的CAS实现，而AbstractFuture使用的是AQS（AbstractQueuedSynchronizer，jdk锁实现的模板类，本文不做讲解），AbstractFuture将99%的操作全部都交于内部类Sync实现，下面讲解一下Sync类的实现
- 任务的状态信息
Sync将任务状态分为5种，分别为running、completing、completed、cancelled、interrupted。其与FutureTask任务的状态对比如下图所示：
FutureTask-AbstractFuture状态对比图.png
从图中可以看出，AbstractFure相比与FutureTask少了两种状态Exceptional和interrupting状态，其中Exceptional状态，在AbstractFure中通过添加一个Throwable类型的结果来实现（如果Throwable对象的值不为null，则说明是exceptional），而interrupting状态再FutureTask中的作用就不是很大，在AstractFurue中并没有设计该状态。
- 成员变量
与FutureTask不同，FutureTask中只有一个Object对象用来存放Future执行的结果，可以是正常结果，也可以是异常。在AbstractFuture中，将两种结果分开，正常结果放在value中，异常结果放在exception中，源码如下：
private V value;
private Throwable exception;
这也是为什么不设置exceptional状态就能区分正常和异常结果的原因。
- 锁方法的重写
该锁使用的是共享锁来实现，主要涉及两个方法：
- tryAcquireShared(int ignored) 该方法是在获取锁时调用，多个线程可以同时获取锁
- tryReleaseShared(int finalState) 该方法是在释放锁时调用
在锁的时候时，我们一般都是先尝试获取锁，然后处理临界资源，处理完成后释放锁。而在AbstractFuture中并不是这种常规的使用方式，其实现是必须有一个线程先调用releaseShared(int arg)释放锁，其他线程才能调用acquireShared(int arg)获取锁，否则，所有的获取锁线程都将会堵塞
源码如下：
protected int tryAcquireShared(int ignored) {
if (isDone()) {
return 1;
}
return -1;
}
@Override
protected boolean tryReleaseShared(int finalState) {
setState(finalState);
return true;
}
从源码中可以看出，调用tryAcquireShared()方法就是判断任务有没有完成（isDone()方法），任务完成成功获取锁，任务没有完成则等待。
那么问题就是任务什么时候完成，任务完成都会调用tryReleaseShared()方法，该方法用于任务完成或者取消时设置最终状态。
- 成员方法
一. 阻塞与非阻塞的get()方法
get()方法或间接调用获取锁操作，如果成功获取锁，则返回对应的结果，如果获取不到锁，则返回对应的异常
V get(long nanos) throws TimeoutException, CancellationException,
ExecutionException, InterruptedException {
//间接调用tryAcquireShared(arg)方法，支持中断，最长等待nanos时间
if (!tryAcquireSharedNanos(-1, nanos)) {
throw new TimeoutException("Timeout waiting for task.");
}
return getValue();
}
V get() throws CancellationException, ExecutionException,
InterruptedException {
//间接调用tryAcquireShared(arg)方法，支持中断，直到获取锁返回
acquireSharedInterruptibly(-1);
return getValue();
}
从源码中可以看出，两种get()方法都会间接的调用tryAcquireShared(arg)方法，且都支持中断。在成功获取锁之后，两者都会调用getValue()方法，其源码如下：
private V getValue() throws CancellationException, ExecutionException {
int state = getState();
switch (state) {
case COMPLETED:
if (exception != null) {
throw new ExecutionException(exception);
} else {
return value;
}
case CANCELLED:
case INTERRUPTED:
throw cancellationExceptionWithCause(
"Task was cancelled.", exception);
default:
throw new IllegalStateException(
"Error, synchronizer in invalid state: " + state);
}
}
该方法主要针对三种终止状态做处理：
- completed状态，该状态又分为正常结束和异常终止，通过判断exception是否为null进行区分，如果异常，则抛出异常，否则返回正常结束的结果
- cancelled和interrupted状态，统一处理抛出取消异常
- 其他情况，讲道理不会出现的状态，如果出现了抛出非法状态异常
二. 设值方法
该类方法都会间接调用 tryReleaseShared(int finalState)方法，使锁处于可获取状态，表示任务执行完成。其中包括set(@Nullable V v)设值正常值、setException(Throwable t)设值异常、cancel(boolean interrupt)取消任务，期源码如下：
boolean set(@Nullable V v) {
return complete(v, null, COMPLETED);
}
boolean setException(Throwable t) {
return complete(null, t, COMPLETED);
}
boolean cancel(boolean interrupt) {
return complete(null, null, interrupt ? INTERRUPTED : CANCELLED);
}
从源码中可以看出三者都调用了complete(@Nullable V v, @Nullable Throwable t, int finalState)方法，下面对该方法进行详细的讲解，其核心逻辑如下：
- 将任务状态由running修改为completing状态
- 如果状态修改成功，则对结果value和异常exception赋值，异常的赋值主要看方法参数finalState，改状态如果为cancelled或者interrupted则为异常赋值，否则赋值为参数t的值（t可能为null），然后调用releaseShared(finalState)（该方法间接调用tryReleaseShared(arg)）方法更新任务状态为最终状态。
- 如果状态修改失败，判断状态是否为completing状态，如果是则说明任务已经执行完成，只在赋值阶段，执行acquireShared(-1)获取锁操作，使自己阻塞至任务完成
源码如下：
private boolean complete(@Nullable V v, @Nullable Throwable t,
int finalState) {
boolean doCompletion = compareAndSetState(RUNNING, COMPLETING);
if (doCompletion) {
// If this thread successfully transitioned to COMPLETING, set the value
// and exception and then release to the final state.
this.value = v;
// Don't actually construct a CancellationException until necessary.
this.exception = ((finalState & (CANCELLED | INTERRUPTED)) != 0)
? new CancellationException("Future.cancel() was called.") : t;
releaseShared(finalState);
} else if (getState() == COMPLETING) {
// If some other thread is currently completing the future, block until
// they are done so we can guarantee completion.
acquireShared(-1);
}
return doCompletion;
}
AbstractFuture的状态转换全部都在该方法中了。
三. 状态判断方法
主要判断任务状态state的值，state的值存放在AQS中，再次不过多讲解，源码如下：
boolean isDone() {
return (getState() & (COMPLETED | CANCELLED | INTERRUPTED)) != 0;
}
boolean isCancelled() {
return (getState() & (CANCELLED | INTERRUPTED)) != 0;
}
boolean wasInterrupted() {
return getState() == INTERRUPTED;
}
AbstractFuture<V>
- 成员变量
private final Sync<V> sync = new Sync<V>();
private final ExecutionList executionList = new ExecutionList();
其中，sync已经讲解，ExecutionList主要用于执行listener，其实用可以参考我的另一篇博客《线程池系列六》-Guava ListenableFutureTask中有关于ExecutionList的详细讲解
- 成员方法
大部分的成员方法都是直接调用sync的对应方法，并没有做过多的操作，只是简单的将方法暴露给外部使用而已。其中，get(long timeout, TimeUnit unit)、get()、isDone()、isCancelled()、wasInterrupted()都是直接调用的sync的方法。
除了上述方法外，还有赋值操作和取消操作的方法，由于该类方法设计到任务完成回调listener方法的关系，因此不是简单的调用sync的方法，其实现如下所示：
protected boolean set(@Nullable V value) {
boolean result = sync.set(value);
if (result) {
executionList.execute();
}
return result;
}
如果设置成功，表示任务结束，则执行listener方法（executionList.execute()）
setException(Throwable throwable)方法与set(@Nullable V value)操作一直，多了一个非空判断而已，不再讲解。
cancel() 如果取消成功，则执行listener回调，如果参数为真，则interruptTask();该方法为抽象方法，子类可以实现。cancel()的源码如下：
public boolean cancel(boolean mayInterruptIfRunning) {
if (!sync.cancel(mayInterruptIfRunning)) {
return false;
}
executionList.execute();
if (mayInterruptIfRunning) {
interruptTask();
}
return true;
}
AbstractFuture与FutureTask的区别
- AbstractFuture通过AQS实现，FutureTask通过unsafe CAS实现，本质是一样的
- AbstractFuture 有五种状态，两种任务结果value和exception，而FutureTask有七种状态，任务执行结果只有一个outcome
- AbstractFuture没有实现Runnable接口，不能作为任务放到线程池中执行，而FutureTask可以
- AbstractFuture有接口回调，FutureTask没有，但是留下了回调的接口，可以重写done()方法
AbstractFuture的使用场景
AbstractFuture是一个抽象类，我们需要自定义子类来使用AbstractFuture，又因此AbstractFuture并没有实现Runnable接口，因此其不适合和线程池配合使用（子类同时实现Runnable接口也是可以的）。它经常用来与AsyncHttpClient配合使用，使用异步HttpClient发起请求，在请求的回调中根据请求的返回结果执行AbstractFuture对象set()、setException()、cancel()操作。
欢迎扫描下方二维码，关注公众号，我们可以进行技术交流，共同成长
qrcode_for_gh_5580beb3cba1_430.jpg
网友评论