jdk原生的future已經(jīng)提供了異步操作树灶，但是不能直接回調(diào)杉女。guava對(duì)future進(jìn)行了增強(qiáng)由捎，核心接口就是ListenableFuture萄传。如果已經(jīng)開始使用了jdk8携悯，可以直接學(xué)習(xí)使用原生的CompletableFuture缔恳，這是jdk從guava中吸收了精華新增的類仰冠。
ListenableFuture繼承了Future蜕依，額外新增了一個(gè)方法典勇，listener是任務(wù)結(jié)束后的回調(diào)方法劫哼，executor是執(zhí)行回調(diào)方法的執(zhí)行器(通常是線程池)。guava中對(duì)future的增強(qiáng)就是在addListener這個(gè)方法上進(jìn)行了各種各樣的封裝割笙，所以addListener是核心方法
void addListener(Runnable listener, Executor executor);
jdk原生FutureTask類是對(duì)Future接口的實(shí)現(xiàn)权烧，guava中ListenableFutureTask繼承了FutureTask并實(shí)現(xiàn)了ListenableFuture，guava異步回調(diào)最簡單的使用：
//ListenableFutureTask通過靜態(tài)create方法返回實(shí)例伤溉，還有一個(gè)重載方法般码，不太常用
ListenableFutureTask<String> task = ListenableFutureTask.create(new Callable<String>() {
@Override
public String call() throws Exception {
return "";
}
});
//啟動(dòng)任務(wù)
new Thread(task).start();
//增加回調(diào)方法，MoreExecutors.directExecutor()返回guava默認(rèn)的Executor乱顾，執(zhí)行回調(diào)方法不會(huì)新開線程板祝，所有回調(diào)方法都在當(dāng)前線程做(可能是主線程或者執(zhí)行ListenableFutureTask的線程，具體可以看最后面的代碼)走净。
//guava異步模塊中參數(shù)有Executor的方法券时，一般還會(huì)有一個(gè)沒有Executor參數(shù)的重載方法，使用的就是MoreExecutors.directExecutor()
task.addListener(new Runnable() {
@Override
public void run() {
System.out.println("done");
}
}, MoreExecutors.directExecutor());
//MoreExecutors.directExecutor()源碼伏伯，execute方法就是直接運(yùn)行橘洞，沒有新開線程
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()";
}
}
一般使用異步模式的時(shí)候，都會(huì)用一個(gè)線程池來提交任務(wù)说搅，不會(huì)像上面那樣簡單的開一個(gè)線程去做炸枣，那樣效率太低下了，所以需要說說guava對(duì)jdk原生線程池的封裝弄唧。
guava對(duì)原生線程池的增強(qiáng)都在MoreExecutor類中适肠，guava對(duì)ExecutorService和ScheduledExecutorService的增強(qiáng)類似，這里只介紹ExecutorService的增強(qiáng).
//真正干活的線程池
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);
//獲得一個(gè)隨著jvm關(guān)閉而關(guān)閉的線程池，通過Runtime.getRuntime().addShutdownHook(hook)實(shí)現(xiàn)
//修改ThreadFactory為創(chuàng)建守護(hù)線程背伴，默認(rèn)jvm關(guān)閉時(shí)最多等待120秒關(guān)閉線程池，重載方法可以設(shè)置時(shí)間
ExecutorService newPoolExecutor = MoreExecutors.getExitingExecutorService(poolExecutor);
//只增加關(guān)閉線程池的鉤子，不改變ThreadFactory
MoreExecutors.addDelayedShutdownHook(poolExecutor, 120, TimeUnit.SECONDS);
//像線程池提交任務(wù)傻寂，并得到ListenableFuture
ListenableFuture<String> listenableFuture = listeningExecutor.submit(new Callable<String>() {
@Override
public String call() throws Exception {
return "";
}
});
//可以通過addListener對(duì)listenableFuture注冊(cè)回調(diào)息尺，但是通常使用Futures中的工具方法
Futures.addCallback(listenableFuture, new FutureCallback<String>() {
@Override
public void onSuccess(String result) {
}
@Override
public void onFailure(Throwable t) {
}
});
/**
* Futures.addCallback源碼，其實(shí)就是包裝了一層addListener疾掰，可以不加executor參數(shù)搂誉，使用上文說的DirectExecutor
* 需要說明的是不加Executor的情況，只適用于輕型的回調(diào)方法静檬，如果回調(diào)方法很耗時(shí)占資源炭懊，會(huì)造成線程阻塞
* 因?yàn)镈irectExecutor有可能在主線程中執(zhí)行回調(diào)
*/
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的異步鏈?zhǔn)綀?zhí)行
//當(dāng)task1執(zhí)行完畢會(huì)回調(diào)執(zhí)行Function的apply方法，如果有task1有異常拋出拂檩，則task2也拋出相同異常侮腹，不執(zhí)行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 "";
}
});
//處理最終的異步任務(wù)
Futures.addCallback(task3, new FutureCallback<String>() {
@Override
public void onSuccess(String result) {
}
@Override
public void onFailure(Throwable t) {
}
});
Futures.transform()和Futures.addCallback()都是對(duì)addListener做了封裝，進(jìn)行回調(diào)的設(shè)置稻励，但是transform更適合用在鏈?zhǔn)教幚淼闹虚g過程父阻，addCallback更適合用在處理最終的結(jié)果上。
源碼分析
我們先來看看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已經(jīng)執(zhí)行完了望抽，執(zhí)行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還沒被執(zhí)行加矛，則放入隊(duì)列中，這個(gè)隊(duì)列是一個(gè)單鏈表,等待任務(wù)執(zhí)行完煤篙，再依次執(zhí)行這個(gè)隊(duì)列所有的等待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;
}
}
實(shí)際上listener模式只是重寫了FutureTask的done方法,我們知道在future task中任務(wù)執(zhí)行后在finishCompletion方法會(huì)調(diào)用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) { // 反轉(zhuǎn)單鏈表
RunnableExecutorPair tmp = list;
list = list.next;
tmp.next = reversedList;
reversedList = tmp;
}
while (reversedList != null) {
executeListener(reversedList.runnable, reversedList.executor);
reversedList = reversedList.next;
}
}
ListenableFuture的監(jiān)聽器模式設(shè)計(jì)很精煉，一目了然辑奈。也是實(shí)現(xiàn)transform等方法的基礎(chǔ)苛茂。
下面我們來看看tranform實(shí)現(xiàn)的原理
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()); // 其實(shí)還是調(diào)用了ListenableFuture的addListener方法
return output;
}
其實(shí)本質(zhì)上tranform是new了一個(gè)新的ListenableFuture output,作為input的監(jiān)聽。當(dāng)input future完成后身害，會(huì)處理監(jiān)聽的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(); //封裝成等待隊(duì)列
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); // 等待超時(shí)
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();
}
當(dāng)input future完成后，由于
input.addListener(output, directExecutor());
@Override
public void addListener(Runnable listener, Executor exec) {
executionList.add(listener, exec);
}
會(huì)運(yùn)用ListenableFuture的監(jiān)聽器模式塌鸯，完成tranform, 我們?cè)賮砘仡櫹翷istenableFuture的監(jiān)聽模式：
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); // 執(zhí)行l(wèi)istener
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); // 獲取上一個(gè)future結(jié)果
} catch (Exception e) {
.... // 只看關(guān)鍵代碼
}
T transformResult;
try {
transformResult = doTransform(localFunction, sourceResult); // input Future的結(jié)果作為輸入
} 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?
}
小結(jié)
tranform的實(shí)現(xiàn)主要是依賴ListenableFuture的監(jiān)聽模式侍瑟，轉(zhuǎn)換后的future作為listener監(jiān)聽轉(zhuǎn)換前的future， 轉(zhuǎn)換前future的輸出作為轉(zhuǎn)換后future的輸入丙猬。最好是在idea debug一下更容易理解涨颜。
