MoreExecutors
directExecutor
ExecutorService executor = Executors.newSingleThreadExecutor(); SettableFuture<Integer> future = SettableFuture.create(); // 使用其他线程去 set 对应的结果。 executor.submit(() -> { future.set(1); }); Futures.addCallback(future, new FutureCallback<>() { @Override public void onSuccess(Integer result) { // main线程执行的 System.out.println("result=" + result + "线程名：" + Thread.currentThread().getName());//main } @Override public void onFailure(Throwable t) { } }, MoreExecutors.directExecutor()); 执行 callback 的线程池这里指定为 MoreExecutors#directExecutor ，那么这里执行打印 result 的线程是主线程
在 MoreExecutors#directExecutor 中，可以看到定义是这样的：
public final class MoreExecutors { // 省略了类内其他成员 public static Executor directExecutor() { return DirectExecutor.INSTANCE; } } 以及
enum DirectExecutor implements Executor { INSTANCE; public void execute(Runnable command) { command.run(); } public String toString() { return "MoreExecutors.directExecutor()"; } } MoreExecutors#directExecutor 其实是一个假的线程池，表示直接执行。
再看下面这个例子：
ExecutorService executor = Executors.newSingleThreadExecutor(); SettableFuture<Integer> future = SettableFuture.create(); // 使用其他线程去 set 对应的结果。 executor.submit(() -> { // 增加线程 sleep 的逻辑。 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } future.set(1); }); Futures.addCallback(future, new FutureCallback<>() { @Override public void onSuccess(Integer result) { // 此时就会被 executor 的线程执行 System.out.println("result=" + result + "线程名：" + Thread.currentThread().getName());//此时还未打印出来，主线程就结束了 } @Override public void onFailure(Throwable t) { } }, MoreExecutors.directExecutor());
那么这里清晰了：
- 如果 future 已经完成，那么
MoreExecutor#directExecutor表示当前线程； - 如果 future 未完成，那么
MoreExecutor#directExecutor就是未来完成 future 的线程。
因此其实具体执行回调的线程某种程度上是不确定的
ListenableFuture
引言
jdk原生的future已经提供了异步操作，但是不能直接回调。guava对future进行了增强，核心接口就是ListenableFuture。JDK8从guava中吸收了精华新增的类CompletableFuture，也可以直接看这个类的学习。
JUC 的 Future 接口提供了一种异步获取任务执行结果的机制，表示一个异步计算的结果。
ExecutorService executor = Executors.newFixedThreadPool(1); Future<String> future = executor.submit(() -> { // 执行异步任务，返回一个结果 return "Task completed"; }); // Blocked String result = future.get(); Executor 实际返回的是实现类 FutureTask，它同时实现了 Runnable 接口，因此可以手动创建异步任务。
FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() { public String call() throws Exception { return "Hello"; } }); new Thread(futureTask).start(); System.out.println(futureTask.get()); 而 Guava 提供的 ListenableFuture 更进一步，允许注册回调，在任务完成后自动执行，实际也是使用它的实现类 ListenableFutureTask。
// 装饰原始的线程池 ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1)); ListenableFuture<String> future = listeningExecutorService.submit(() -> { // int i = 1 / 0; return "Hello"; }); // 添加回调 1 Futures.addCallback(future, new FutureCallback<String>() { // 任务成功时的回调 @Override public void onSuccess(String result) { System.out.println(result); } // 任务失败时的回调 @Override public void onFailure(Throwable t) { System.out.println("Error: " + t.getMessage()); } }, listeningExecutorService); // 添加回调 2 future.addListener(new Runnable() { @Override public void run() { System.out.println("Done"); } }, listeningExecutorService); 回调源码剖析
先看下ListenableFuture接口定义：
public interface ListenableFuture<V> extends Future<V> { void addListener(Runnable listener, Executor executor); } 可以看到，这个接口在Future接口的基础上增加了addListener方法，允许我们注册回调函数。当然，在编程时可能不会直接使用这个接口，因为这个接口只能传Runnable实例。
addListener方法
public void addListener(Runnable listener, Executor executor) { checkNotNull(listener, "Runnable was null."); checkNotNull(executor, "Executor was null."); // Checking isDone and listeners != TOMBSTONE may seem redundant, but our contract for // addListener says that listeners execute 'immediate' if the future isDone(). However, our // protocol for completing a future is to assign the value field (which sets isDone to true) and // then to release waiters, followed by executing afterDone(), followed by releasing listeners. // That means that it is possible to observe that the future isDone and that your listeners // don't execute 'immediately'. By checking isDone here we avoid that. // A corollary to all that is that we don't need to check isDone inside the loop because if we // get into the loop we know that we weren't done when we entered and therefore we aren't under // an obligation to execute 'immediately'. if (!isDone()) { Listener oldHead = listeners; // 获取当前监听器的头结点 if (oldHead != Listener.TOMBSTONE) {// 检查当前的头节点是否是TOMBSTONE。TOMBSTONE用来表示监听器列表不再接受新的监听器，通常是因为Future已经完成。 Listener newNode = new Listener(listener, executor);//通过这个listener新增一个一个节点，节点中包含executor do { newNode.next = oldHead;//将newNode.next指向当前头结点，此时newNode就是头结点 if (ATOMIC_HELPER.casListeners(this, oldHead, newNode)) {//检查头节点是否更新成功 return;//更新成功就可以返回了 } oldHead = listeners; // 重新执行 头插法 } while (oldHead != Listener.TOMBSTONE);// 如果头节点变成了TOMBSTONE，则退出循环；并且 } } // If we get here then the Listener TOMBSTONE was set, which means the future is done, call // the listener. executeListener(listener, executor);//执行到这里意味着监听器TOMBSTONE就设置好了，也就是future已经完成，可以直接调用监听器 } 这里其实就是在添加listener的方法中首先检查Future是否已经完成：
- 如果Future已经完成，那么就没有必要添加新的监听器，直接executeListener。
- 如果future没有完成，那么会新建一个Listener节点，并插入到链表头部（Listener就是一个链表）
如果已经完成，会直接执行executeListner 方法
private static void executeListener(Runnable runnable, Executor executor) { try { executor.execute(runnable);//直接使用listener拥有的线程executor执行 } catch (Exception e) { // sneaky checked exception // 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 Exception because we want Errors to propagate up. log.get() .log( Level.SEVERE, "RuntimeException while executing runnable " + runnable + " with executor " + executor, e); } } 那么如果没有完成呢，在listener链表中的什么时候会执行？看后续的回调函数的触发内容
addCallback方法
Futures类还提供了另一个回调方法：addCallback方法
public static <V> void addCallback( final ListenableFuture<V> future, final FutureCallback<? super V> callback, Executor executor) { Preconditions.checkNotNull(callback); future.addListener(new CallbackListener<V>(future, callback), executor);//调用了addListener方法 } 这里调用了ListenableFuture接口的addListener方法，传入了一个CallbackListener实例。而这个实例由需要传入future和一个Callback实例，所以这个回调是可以拿到返回值的。本质上是guava基于Runnable封了一个回调接口。
看下这个CallbackListener接口：
private static final class CallbackListener<V> implements Runnable { final Future<V> future; final FutureCallback<? super V> callback; CallbackListener(Future<V> future, FutureCallback<? super V> callback) { this.future = future; this.callback = callback; } public void run() {//回调时的逻辑 if (future instanceof InternalFutureFailureAccess) { Throwable failure = InternalFutures.tryInternalFastPathGetFailure((InternalFutureFailureAccess) future); if (failure != null) { callback.onFailure(failure); return; } } final V value; try { value = getDone(future);//获取返回值 } catch (ExecutionException e) { callback.onFailure(e.getCause());//如果发生了异常，则会调用onFailure方法通知异常 return; } catch (RuntimeException | Error e) { callback.onFailure(e);//如果发生了异常，则会调用onFailure方法通知异常 return; } callback.onSuccess(value);//将返回值调用FutureCallback实例的onSuccess方法执行注册的回调逻辑 } } 那么这个回调函数什么时候会执行？看后续的回调函数的触发内容
回调函数的触发
那么这些回调方法什么时候会触发呢？
private static void complete(AbstractFuture<?> param) { // Declare a "true" local variable so that the Checker Framework will infer nullness. AbstractFuture<?> future = param;//获取future Listener next = null; outer: while (true) { future.releaseWaiters();//通知所有执行的方法 // We call this before the listeners in order to avoid needing to manage a separate stack data // structure for them. Also, some implementations rely on this running prior to listeners // so that the cleanup work is visible to listeners. // afterDone() should be generally fast and only used for cleanup work... but in theory can // also be recursive and create StackOverflowErrors future.afterDone(); // push the current set of listeners onto next next = future.clearListeners(next);//反转listener链表 future = null; while (next != null) { Listener curr = next;//获取当前listener next = next.next; /* * requireNonNull is safe because the listener stack never contains TOMBSTONE until after * clearListeners. */ Runnable task = requireNonNull(curr.task); if (task instanceof SetFuture) { SetFuture<?> setFuture = (SetFuture<?>) task; // We unwind setFuture specifically to avoid StackOverflowErrors in the case of long // chains of SetFutures // Handling this special case is important because there is no way to pass an executor to // setFuture, so a user couldn't break the chain by doing this themselves. It is also // potentially common if someone writes a recursive Futures.transformAsync transformer. future = setFuture.owner; if (future.value == setFuture) { Object valueToSet = getFutureValue(setFuture.future); if (ATOMIC_HELPER.casValue(future, setFuture, valueToSet)) { continue outer; } } // other wise the future we were trying to set is already done. } else { /* * requireNonNull is safe because the listener stack never contains TOMBSTONE until after * clearListeners. */ executeListener(task, requireNonNull(curr.executor));// 交给listener拥有的线程池进行处理 } } break; } } 那哪些方法会来调用这个complete方法呢？
Service
Guava 的 Service 框架是一个用于管理服务生命周期的轻量级框架，可以帮助我们把异步操作封装成一个Service服务。让这个服务有了运行状态(也可以理解成生命周期)，这样可以实时了解当前服务的运行状态。同时还可以添加监听器来监听服务运行状态之间的变化。
Guava里面的服务有五种状态，如下所示：
- Service.State.NEW: 服务创建状态
- Service.State.STARTING: 服务启动中
- Service.State.RUNNING：服务启动完成，正在运行中
- Service.State.STOPPING: 服务停止中
- Service.State.TERMINATED: 服务停止完成，结束
所有的服务都需要实现Service接口，里面包括了服务需要实现的一些基本方法，以下是Service接口：
public interface Service { //启动当前服务,只有当服务的状态是NEW的情况下才可以启动，否则抛出IllegalStateException异常 @CanIgnoreReturnValue Service startAsync(); //判断当前服务是否处在运行状态 (RUNNING) boolean isRunning(); //获取当前服务的状态 Service.State state(); //停止当前服务 @CanIgnoreReturnValue Service stopAsync(); // 等待当前服务到达RUNNING状态 void awaitRunning(); // 在指定的时间内等待当前服务到达RUNNING状态，如果在指定时间没有达到则抛出TimeoutException void awaitRunning(long timeout, TimeUnit unit) throws TimeoutException; // 等待当前服务到达TERMINATED状态 void awaitTerminated(); //在指定的时间内等待当前服务达到TERMINATED状态， void awaitTerminated(long timeout, TimeUnit unit) throws TimeoutException; // 获取服务器失败的原因,在服务是FAILED的状态的时候调用该函数，否则抛出IllegalStateException异常 Throwable failureCause(); //监听当前服务的状态改变，executor参数表示，监听回调函数在哪里执行 void addListener(Service.Listener listener, Executor executor); } 那应该怎么来使用Service，需要实现的异步逻辑包装成服务呢．Guava里面已经给提供了三个基础实现类：
- AbstractService
- AbstractExecutionThreadService
- AbstractScheduledService
AbstractExecutionThreadService
AbstractExecutionThreadService可以把一个具体的异步操作封装成Service服务。说白了就是把之前在线程的实现逻辑封装成服务，把之前线程的具体实现逻辑搬到AbstractExecutionThreadService的实现方法run()方法去执行。
常用方法介绍
首先AbstractExecutionThreadService实现了Service，Service的方法在AbstractExecutionThreadService里面都有，AbstractExecutionThreadService新加了一些方法。如下所示：
public class AbstractExecutionThreadService { // 开始执行服务逻辑的时候会调用，可以在里面做一些初始化的操作 protected void startUp() throws Exception; // 当前服务需要执行的具体逻辑 protected abstract void run() throws Exception; // 服务停止之后会调用的函数，可以在里面做 一些释放资源的处理 protected void shutDown() throws Exception {} //比如在run方法里面有一个无线循环，可以在这个方法里面置状态，退出无线循环，让服务真正停止 //调stopAsync函数的时候，会调用该方法 protected void triggerShutdown() {} ... } AbstractExecutionThreadService类里面最重要的就是run()方法了，这个方法是服务需要具体实现的方法，服务需要处理的具体逻辑在这个方法里面做。
具体使用
public class AbstractExecutionThreadServiceImpl extends AbstractExecutionThreadService { private volatile boolean running = true; //声明一个状态 @Override protected void startUp() { //TODO: 做一些初始化操作 } @Override public void run() { // 具体需要实现的业务逻辑，会在线程中执行 while (running) { try { // 等待2s Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); System.out.println("do our work....."); } catch (Exception e) { //TODO: 处理异常，这里如果抛出异常，会使服务状态变为failed同时导致任务终止。 } } } @Override protected void triggerShutdown() { //TODO: 如果的run方法中有无限循环，可以在这里置状态，让其退出无限循环，stopAsync()里面会调用到该方法 running = false; //这里改变状态值，run方法中就能够得到响应。 } @Override protected void shutDown() throws Exception { //TODO: 可以做一些清理操作，比如关闭连接。shutDown() 是在线程的具体实现里面调用的 } } AbstractScheduledService
AbstractScheduledService可以把周期性的任务封装成一个服务。线程池也有一个周期性的线程池么，两者是一一对应的．
常用方法介绍
AbstractScheduledService也是一个服务所以Service里面的方法AbstractScheduledService也都有，同时，AbstractScheduledService也新增了一些其它方法
public class AbstractScheduledService { ... //周期任务的具体逻辑在这个里面实现 protected abstract void runOneIteration() throws Exception; //启动周期任务之前调用，可以在里面做一些初始化的操作 protected void startUp() throws Exception; //周期任务停止之后调用，可以在里面做 一些释放资源的处理 protected void shutDown() throws Exception {} //指定当前周期任务在哪个ScheduledExecutorService里面调用 //Scheduler.newFixedDelaySchedule() protected abstract Scheduler scheduler(); ... } 具体使用
自定义一个类继承AbstractScheduledService，实现一个非常简单的周期性任务．
public class AbstractScheduledServiceImpl extends AbstractScheduledService { protected void startUp() throws Exception { //TODO: 做一些初始化操作 } @Override protected void shutDown() throws Exception { //TODO: 可以做一些清理操作，比如关闭连接。shutDown() 是在线程的具体实现里面调用的 } @Override protected void runOneIteration() throws Exception { // 每次周期任务的执行逻辑 try { System.out.println("do work...."); } catch (Exception e) { //TODO: 处理异常，这里如果抛出异常，会使服务状态变为failed同时导致任务终止。 } } @Override protected Scheduler scheduler() { // 5s执行一次的Scheduler return Scheduler.newFixedDelaySchedule(1, 5, TimeUnit.SECONDS); } } ServiceManager
ServiceManager是用来管理多个服务的，让对多个服务的操作变的更加方便，比如可以同时去启动多个服务，同时去停止多个服务等等。
常用方法介绍
public class ServiceManager { //构造函数，管理多个Service服务 public ServiceManager(Iterable<? extends Service> services); //给ServiceManager增加状态监听器 public void addListener(Listener listener, Executor executor); public void addListener(Listener listener); //开始启动ServiceManager里面所有Service服务 public ServiceManager startAsync(); //等待ServiceManager里面所有Service服务达到Running状态 public void awaitHealthy(); public void awaitHealthy(long timeout, TimeUnit unit) throws TimeoutException; //停止ServiceManager里面所有Service服务 public ServiceManager stopAsync(); //等待ServiceManager里面所有Service服务达到终止状态 public void awaitStopped(); public void awaitStopped(long timeout, TimeUnit unit) throws TimeoutException; //ServiceManager里面所有Service服务是否都达到了Running状态 public boolean isHealthy(); //以状态为索引返回当前所有服务的快照 public ImmutableMultimap<State, Service> servicesByState(); //返回一个Map对象，记录被管理的服务启动的耗时、以毫秒为单位，同时Map默认按启动时间排序 public ImmutableMap<Service, Long> startupTimes(); } 具体使用
public void serviceManagerTest() { // 定义两个服务 AbstractExecutionThreadServiceImpl service0 = new AbstractExecutionThreadServiceImpl(); AbstractScheduledServiceImpl service1 = new AbstractScheduledServiceImpl(); List<Service> serviceList = Lists.newArrayList(service0, service1); // ServiceManager里面管理这两个服务 ServiceManager serviceManager = new ServiceManager(serviceList); // 添加监听 serviceManager.addListener(new ServiceManager.Listener() { @Override public void healthy() { super.healthy(); System.out.println("healthy"); } @Override public void stopped() { super.stopped(); System.out.println("stopped"); } @Override public void failure(Service service) { super.failure(service); System.out.println("failure"); } }); // 启动服务，等待所有的服务都达到running状态 serviceManager.startAsync().awaitHealthy(); // 等待30s Uninterruptibles.sleepUninterruptibly(30, TimeUnit.SECONDS); // 停止服务 serviceManager.stopAsync().awaitStopped(); } 
