jdk原生的future已经提供了异步操作，但是不能直接回调。guava对future进行了增强，核心接口就是ListenableFuture。如果已经开始使用了jdk8，可以直接学习使用原生的CompletableFuture，这是jdk从guava中吸收了精华新增的类。
ListenableFuture继承了Future，额外新增了一个方法，listener是任务结束后的回调方法，executor是执行回调方法的执行器(通常是线程池)。guava中对future的增强就是在addListener这个方法上进行了各种各样的封装，所以addListener是核心方法
void addListener(Runnable listener, Executor executor);
jdk原生FutureTask类是对Future接口的实现，guava中ListenableFutureTask继承了FutureTask并实现了ListenableFuture，guava异步回调最简单的使用：
//ListenableFutureTask通过静态create方法返回实例，还有一个重载方法，不太常用
ListenableFutureTask<String> task = ListenableFutureTask.create(new Callable<String>() {
@Override
public String call() throws Exception {
return "";
}
});
//启动任务
new Thread(task).start();
//增加回调方法，MoreExecutors.directExecutor()返回guava默认的Executor，执行回调方法不会新开线程，所有回调方法都在当前线程做(可能是主线程或者执行ListenableFutureTask的线程，具体可以看最后面的代码)。
//guava异步模块中参数有Executor的方法，一般还会有一个没有Executor参数的重载方法，使用的就是MoreExecutors.directExecutor()
task.addListener(new Runnable() {
@Override
public void run() {
System.out.println("done");
}
}, MoreExecutors.directExecutor());
//MoreExecutors.directExecutor()源码，execute方法就是直接运行，没有新开线程
public static Executor directExecutor() {
return DirectExecutor.INSTANCE;
}
private enum DirectExecutor implements Executor {
INSTANCE;
@Override
public void execute(Runnable command) {
command.run();
}
@Override
public String toString() {
return "MoreExecutors.directExecutor()";
}
}
一般使用异步模式的时候，都会用一个线程池来提交任务，不会像上面那样简单的开一个线程去做，那样效率太低下了，所以需要说说guava对jdk原生线程池的封装。
guava对原生线程池的增强都在MoreExecutor类中，guava对ExecutorService和ScheduledExecutorService的增强类似，这里只介绍ExecutorService的增强.
//真正干活的线程池
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
5,
5,
0,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new CustomizableThreadFactory("demo"),
new ThreadPoolExecutor.DiscardPolicy());
//guava的接口ListeningExecutorService继承了jdk原生ExecutorService接口，重写了submit方法，修改返回值类型为ListenableFuture
ListeningExecutorService listeningExecutor = MoreExecutors.listeningDecorator(poolExecutor);
//获得一个随着jvm关闭而关闭的线程池，通过Runtime.getRuntime().addShutdownHook(hook)实现
//修改ThreadFactory为创建守护线程，默认jvm关闭时最多等待120秒关闭线程池，重载方法可以设置时间
ExecutorService newPoolExecutor = MoreExecutors.getExitingExecutorService(poolExecutor);
//只增加关闭线程池的钩子，不改变ThreadFactory
MoreExecutors.addDelayedShutdownHook(poolExecutor, 120, TimeUnit.SECONDS);
//像线程池提交任务，并得到ListenableFuture
ListenableFuture<String> listenableFuture = listeningExecutor.submit(new Callable<String>() {
@Override
public String call() throws Exception {
return "";
}
});
//可以通过addListener对listenableFuture注册回调，但是通常使用Futures中的工具方法
Futures.addCallback(listenableFuture, new FutureCallback<String>() {
@Override
public void onSuccess(String result) {
}
@Override
public void onFailure(Throwable t) {
}
});
/**
* Futures.addCallback源码，其实就是包装了一层addListener，可以不加executor参数，使用上文说的DirectExecutor
* 需要说明的是不加Executor的情况，只适用于轻型的回调方法，如果回调方法很耗时占资源，会造成线程阻塞
* 因为DirectExecutor有可能在主线程中执行回调
*/
public static <V> void addCallback(final ListenableFuture<V> future, final FutureCallback<? super V> callback, Executor executor) {
Preconditions.checkNotNull(callback);
Runnable callbackListener =
new Runnable() {
@Override
public void run() {
final V value;
try {
value = getDone(future);
} catch (ExecutionException e) {
callback.onFailure(e.getCause());
return;
} catch (RuntimeException e) {
callback.onFailure(e);
return;
} catch (Error e) {
callback.onFailure(e);
return;
}
callback.onSuccess(value);
}
};
future.addListener(callbackListener, executor);
}
使用guava的异步链式执行
//当task1执行完毕会回调执行Function的apply方法，如果有task1有异常抛出，则task2也抛出相同异常，不执行apply
ListenableFuture<String> task2 = Futures.transform(task1, new Function<String, String>() {
@Override
public String apply(String input) {
return "";
}
});
ListenableFuture<String> task3 = Futures.transform(task2, new Function<String, String>() {
@Override
public String apply(String input) {
return "";
}
});
//处理最终的异步任务
Futures.addCallback(task3, new FutureCallback<String>() {
@Override
public void onSuccess(String result) {
}
@Override
public void onFailure(Throwable t) {
}
});
Futures.transform()和Futures.addCallback()都是对addListener做了封装，进行回调的设置，但是transform更适合用在链式处理的中间过程，addCallback更适合用在处理最终的结果上。
源码分析
我们先来看看listener的add方法
@Override
public void addListener(Runnable listener, Executor exec) {
executionList.add(listener, exec);
}
public void add(Runnable runnable, Executor executor) {
// Fail fast on a null. We throw NPE here because the contract of Executor states that it throws
// NPE on null listener, so we propagate that contract up into the add method as well.
checkNotNull(runnable, "Runnable was null.");
checkNotNull(executor, "Executor was null.");
// Lock while we check state. We must maintain the lock while adding the new pair so that
// another thread can't run the list out from under us. We only add to the list if we have not
// yet started execution.
synchronized (this) {
if (!executed) {
runnables = new RunnableExecutorPair(runnable, executor, runnables);
return;
}
}
// Execute the runnable immediately. Because of scheduling this may end up getting called before
// some of the previously added runnables, but we're OK with that. If we want to change the
// contract to guarantee ordering among runnables we'd have to modify the logic here to allow
// it.
executeListener(runnable, executor);
}
如果task已经执行完了，执行executeListener
private static void executeListener(Runnable runnable, Executor executor) {
try {
executor.execute(runnable);
} catch (RuntimeException e) {
// Log it and keep going -- bad runnable and/or executor. Don't punish the other runnables if
// we're given a bad one. We only catch RuntimeException because we want Errors to propagate
// up.
log.log(
Level.SEVERE,
"RuntimeException while executing runnable " + runnable + " with executor " + executor,
e);
}
}
如果task还没被执行，则放入队列中，这个队列是一个单链表,等待任务执行完，再依次执行这个队列所有的等待runnable。
private static final class RunnableExecutorPair {
final Runnable runnable;
final Executor executor;
@Nullable RunnableExecutorPair next;
RunnableExecutorPair(Runnable runnable, Executor executor, RunnableExecutorPair next) {
this.runnable = runnable;
this.executor = executor;
this.next = next;
}
}
实际上listener模式只是重写了FutureTask的done方法,我们知道在future task中任务执行后在finishCompletion方法会调用done方法
@Override
protected void done() {
executionList.execute();
}
public void execute() {
// Lock while we update our state so the add method above will finish adding any listeners
// before we start to run them.
RunnableExecutorPair list;
synchronized (this) {
if (executed) {
return;
}
executed = true;
list = runnables;
runnables = null; // allow GC to free listeners even if this stays around for a while.
}
// If we succeeded then list holds all the runnables we to execute. The pairs in the stack are
// in the opposite order from how they were added so we need to reverse the list to fulfill our
// contract.
// This is somewhat annoying, but turns out to be very fast in practice. Alternatively, we could
// drop the contract on the method that enforces this queue like behavior since depending on it
// is likely to be a bug anyway.
// N.B. All writes to the list and the next pointers must have happened before the above
// synchronized block, so we can iterate the list without the lock held here.
RunnableExecutorPair reversedList = null;
while (list != null) { // 反转单链表
RunnableExecutorPair tmp = list;
list = list.next;
tmp.next = reversedList;
reversedList = tmp;
}
while (reversedList != null) {
executeListener(reversedList.runnable, reversedList.executor);
reversedList = reversedList.next;
}
}
ListenableFuture的监听器模式设计很精炼，一目了然。也是实现transform等方法的基础。
下面我们来看看tranform实现的原理
ListenableFutureTask<String> task1 = ListenableFutureTask.create(new Callable<String>() {
@Override
public String call() throws Exception {
return "good";
}
});
ListenableFuture<String> task2 = Futures.transform(task1, new Function<String, String>() {
@Override
public String apply(String input) {
return "yes";
}
});
new Thread(task1).start();
try {
System.out.println(task2.get(10, TimeUnit.SECONDS));
} catch (Exception e) {
System.out.println(e);
}
源码分析
public static <I, O> ListenableFuture<O> transform(
ListenableFuture<I> input, Function<? super I, ? extends O> function) {
return AbstractTransformFuture.create(input, function);
}
static <I, O> ListenableFuture<O> create(
ListenableFuture<I> input, Function<? super I, ? extends O> function) {
checkNotNull(function);
TransformFuture<I, O> output = new TransformFuture<I, O>(input, function);
input.addListener(output, directExecutor()); // 其实还是调用了ListenableFuture的addListener方法
return output;
}
其实本质上tranform是new了一个新的ListenableFuture output,作为input的监听。当input future完成后，会处理监听的output future。
tranform后的future继承了AbstractFuture：
public V get(long timeout, TimeUnit unit)
throws InterruptedException, TimeoutException, ExecutionException {
// NOTE: if timeout < 0, remainingNanos will be < 0 and we will fall into the while(true) loop
// at the bottom and throw a timeoutexception.
long remainingNanos = unit.toNanos(timeout); // we rely on the implicit null check on unit.
if (Thread.interrupted()) {
throw new InterruptedException();
}
Object localValue = value;
if (localValue != null & !(localValue instanceof AbstractFuture.SetFuture)) {
return getDoneValue(localValue);
}
// we delay calling nanoTime until we know we will need to either park or spin
final long endNanos = remainingNanos > 0 ? System.nanoTime() + remainingNanos : 0;
long_wait_loop:
if (remainingNanos >= SPIN_THRESHOLD_NANOS) {
Waiter oldHead = waiters;
if (oldHead != Waiter.TOMBSTONE) {
Waiter node = new Waiter(); //封装成等待队列
do {
node.setNext(oldHead);
if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) {
while (true) {
LockSupport.parkNanos(this, remainingNanos); // 挂起
// Check interruption first, if we woke up due to interruption we need to honor that.
if (Thread.interrupted()) {
removeWaiter(node);
throw new InterruptedException();
}
// Otherwise re-read and check doneness. If we loop then it must have been a spurious
// wakeup
localValue = value;
if (localValue != null & !(localValue instanceof AbstractFuture.SetFuture)) {
return getDoneValue(localValue);
}
// timed out?
remainingNanos = endNanos - System.nanoTime();
if (remainingNanos < SPIN_THRESHOLD_NANOS) {
// Remove the waiter, one way or another we are done parking this thread.
removeWaiter(node); // 等待超时
break long_wait_loop; // jump down to the busy wait loop
}
}
}
oldHead = waiters; // re-read and loop.
} while (oldHead != Waiter.TOMBSTONE);
}
// re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a
// waiter.
return getDoneValue(value);
}
// If we get here then we have remainingNanos < SPIN_THRESHOLD_NANOS and there is no node on the
// waiters list
while (remainingNanos > 0) {
localValue = value;
if (localValue != null & !(localValue instanceof AbstractFuture.SetFuture)) {
return getDoneValue(localValue);
}
if (Thread.interrupted()) {
throw new InterruptedException();
}
remainingNanos = endNanos - System.nanoTime();
}
throw new TimeoutException();
}
当input future完成后，由于
input.addListener(output, directExecutor());
@Override
public void addListener(Runnable listener, Executor exec) {
executionList.add(listener, exec);
}
会运用ListenableFuture的监听器模式，完成tranform, 我们再来回顾下ListenableFuture的监听模式：
protected void done() {
executionList.execute();
}
public void execute() {
// Lock while we update our state so the add method above will finish adding any listeners
// before we start to run them.
RunnableExecutorPair list;
synchronized (this) {
if (executed) {
return;
}
executed = true;
list = runnables;
runnables = null; // allow GC to free listeners even if this stays around for a while.
}
// If we succeeded then list holds all the runnables we to execute. The pairs in the stack are
// in the opposite order from how they were added so we need to reverse the list to fulfill our
// contract.
// This is somewhat annoying, but turns out to be very fast in practice. Alternatively, we could
// drop the contract on the method that enforces this queue like behavior since depending on it
// is likely to be a bug anyway.
// N.B. All writes to the list and the next pointers must have happened before the above
// synchronized block, so we can iterate the list without the lock held here.
RunnableExecutorPair reversedList = null;
while (list != null) {
RunnableExecutorPair tmp = list;
list = list.next;
tmp.next = reversedList;
reversedList = tmp;
}
while (reversedList != null) {
executeListener(reversedList.runnable, reversedList.executor); // 执行listener
reversedList = reversedList.next;
}
}
private static void executeListener(Runnable runnable, Executor executor) {
try {
executor.execute(runnable);
} catch (RuntimeException e) {
....
}
}
public void execute(Runnable command) {
command.run();
}
这里的command是output
abstract class AbstractTransformFuture<I, O, F, T> extends AbstractFuture.TrustedFuture<O>
implements Runnable
AbstractTransformFuture:
public final void run() {
ListenableFuture<? extends I> localInputFuture = inputFuture;
F localFunction = function;
if (isCancelled() | localInputFuture == null | localFunction == null) {
return;
}
inputFuture = null;
function = null;
I sourceResult;
try {
sourceResult = getDone(localInputFuture); // 获取上一个future结果
} catch (Exception e) {
.... // 只看关键代码
}
T transformResult;
try {
transformResult = doTransform(localFunction, sourceResult); // input Future的结果作为输入
} catch (UndeclaredThrowableException e) {
....
}
setResult(transformResult);
}
看看doTransform方法
@Override
@Nullable
O doTransform(Function<? super I, ? extends O> function, @Nullable I input) {
return function.apply(input);
// TODO(lukes): move the UndeclaredThrowable catch block here?
}
小结
tranform的实现主要是依赖ListenableFuture的监听模式，转换后的future作为listener监听转换前的future， 转换前future的输出作为转换后future的输入。最好是在idea debug一下更容易理解。
网友评论