【Guava】并发编程ListenableFuture&Service

MoreExecutors

directExecutor

1. ExecutorService executor = Executors.newSingleThreadExecutor();
2. SettableFuture<Integer> future = SettableFuture.create();
3. // 使用其他线程去 set 对应的结果。
4. executor.submit(() -> {
5.     future.set(1);
6. });
8. Futures.addCallback(future, new FutureCallback<>() {
9.     @Override
10.     public void onSuccess(Integer result) {
11.         // main线程执行的
12.         System.out.println("result=" + result + "线程名：" + Thread.currentThread().getName());//main
13.     }
15.     @Override
16.     public void onFailure(Throwable t) {
17.     }
18. }, MoreExecutors.directExecutor());

复制代码

执行 callback 的线程池这里指定为 MoreExecutors#directExecutor ，那么这里执行打印 result 的线程是主线程

在 MoreExecutors#directExecutor 中，可以看到界说是如许的：

1. public final class MoreExecutors {
2.     // 省略了类内其他成员
3.     public static Executor directExecutor() {
4.         return DirectExecutor.INSTANCE;
5.   }
6. }

复制代码

以及

1. @GwtCompatible
2. @ElementTypesAreNonnullByDefault
3. enum DirectExecutor implements Executor {
4.   INSTANCE;
6.   @Override
7.   public void execute(Runnable command) {
8.     command.run();
9.   }
11.   @Override
12.   public String toString() {
13.     return "MoreExecutors.directExecutor()";
14.   }
15. }

复制代码

MoreExecutors#directExecutor 其实是一个假的线程池，表示直接执行。
再看下面这个例子：

1. ExecutorService executor = Executors.newSingleThreadExecutor();
2. SettableFuture<Integer> future = SettableFuture.create();
3. // 使用其他线程去 set 对应的结果。
4. executor.submit(() -> {
5.     // 增加线程 sleep 的逻辑。
6.     try {
7.         Thread.sleep(1000);
8.     } catch (InterruptedException e) {
9.         e.printStackTrace();
10.     }
11.     future.set(1);
12. });
14. Futures.addCallback(future, new FutureCallback<>() {
15.     @Override
16.     public void onSuccess(Integer result) {
17.         // 此时就会被 executor 的线程执行
18.         System.out.println("result=" + result + "线程名：" + Thread.currentThread().getName());//此时还未打印出来，主线程就结束了
19.     }
21.     @Override
22.     public void onFailure(Throwable t) {
23.     }
24. }, MoreExecutors.directExecutor());

复制代码

那么这里清晰了：

• 如果 future 已经完成，那么 MoreExecutor#directExecutor 表示当前线程；
  
• 如果 future 未完成，那么 MoreExecutor#directExecutor 就是将来完成 future 的线程。
  

因此其实详细执行回调的线程某种水平上是不确定的
ListenableFuture

引言

jdk原生的future已经提供了异步操作，但是不能直接回调。guava对future进行了增强，核心接口就是ListenableFuture。JDK8从guava中吸收了精华新增的类CompletableFuture，也可以直接看这个类的学习。
JUC 的 Future 接口提供了一种异步获取使命执行结果的机制，表示一个异步计算的结果。

1. ExecutorService executor = Executors.newFixedThreadPool(1);
2. Future<String> future = executor.submit(() -> {
3.     // 执行异步任务，返回一个结果
4.     return "Task completed";
5. });
6. // Blocked
7. String result = future.get();

复制代码

Executor 实际返回的是实现类 FutureTask，它同时实现了 Runnable 接口，因此可以手动创建异步使命。

1. FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
2.     @Override
3.     public String call() throws Exception {
4.         return "Hello";
5.     }
6. });
7.         
8. new Thread(futureTask).start();
9. System.out.println(futureTask.get());

复制代码

而 Guava 提供的 ListenableFuture 更进一步，允许注册回调，在使命完成后自动执行，实际也是使用它的实现类 ListenableFutureTask。

1. // 装饰原始的线程池
2. ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
3. ListenableFuture<String> future = listeningExecutorService.submit(() -> {
4.     // int i = 1 / 0;
5.     return "Hello";
6. });
8. // 添加回调 1
9. Futures.addCallback(future, new FutureCallback<String>() {
10.     // 任务成功时的回调
11.     @Override
12.     public void onSuccess(String result) {
13.         System.out.println(result);
14.     }
16.     // 任务失败时的回调
17.     @Override
18.     public void onFailure(Throwable t) {
19.         System.out.println("Error: " + t.getMessage());
20.     }
21. }, listeningExecutorService);
23. // 添加回调 2
24. future.addListener(new Runnable() {
25.     @Override
26.     public void run() {
27.         System.out.println("Done");
28.     }
29. }, listeningExecutorService);

复制代码

回调源码分析

先看下ListenableFuture接口界说：

1. public interface ListenableFuture<V> extends Future<V> {
2.     void addListener(Runnable listener, Executor executor);
3. }

复制代码

可以看到，这个接口在Future接口的底子上增长了addListener方法，允许我们注册回调函数。固然，在编程时可能不会直接使用这个接口，由于这个接口只能传Runnable实例。
addListener方法

1. @Override
2.   public void addListener(Runnable listener, Executor executor) {
3.     checkNotNull(listener, "Runnable was null.");
4.     checkNotNull(executor, "Executor was null.");
5.     // Checking isDone and listeners != TOMBSTONE may seem redundant, but our contract for
6.     // addListener says that listeners execute 'immediate' if the future isDone(). However, our
7.     // protocol for completing a future is to assign the value field (which sets isDone to true) and
8.     // then to release waiters, followed by executing afterDone(), followed by releasing listeners.
9.     // That means that it is possible to observe that the future isDone and that your listeners
10.     // don't execute 'immediately'.  By checking isDone here we avoid that.
11.     // A corollary to all that is that we don't need to check isDone inside the loop because if we
12.     // get into the loop we know that we weren't done when we entered and therefore we aren't under
13.     // an obligation to execute 'immediately'.
14.     if (!isDone()) {
15.       Listener oldHead = listeners; // 获取当前监听器的头结点
16.       if (oldHead != Listener.TOMBSTONE) {// 检查当前的头节点是否是TOMBSTONE。TOMBSTONE用来表示监听器列表不再接受新的监听器，通常是因为Future已经完成。
17.         Listener newNode = new Listener(listener, executor);//通过这个listener新增一个一个节点，节点中包含executor
18.         do {
19.           newNode.next = oldHead;//将newNode.next指向当前头结点，此时newNode就是头结点
20.           if (ATOMIC_HELPER.casListeners(this, oldHead, newNode)) {//检查头节点是否更新成功
21.             return;//更新成功就可以返回了
22.           }
23.           oldHead = listeners; // 重新执行 头插法
24.         } while (oldHead != Listener.TOMBSTONE);// 如果头节点变成了TOMBSTONE，则退出循环；并且
25.       }
26.     }
27.     // If we get here then the Listener TOMBSTONE was set, which means the future is done, call
28.     // the listener.
29.     executeListener(listener, executor);//执行到这里意味着监听器TOMBSTONE就设置好了，也就是future已经完成，可以直接调用监听器
30.   }

复制代码

这里其实就是在添加listener的方法中首先检查Future是否已经完成：

• 如果Future已经完成，那么就没有必要添加新的监听器，直接executeListener。
  
• 如果future没有完成，那么会新建一个Listener节点，并插入到链表头部（Listener就是一个链表）
  

如果已经完成，会直接执行executeListner 方法

1. private static void executeListener(Runnable runnable, Executor executor) {
2.   try {
3.     executor.execute(runnable);//直接使用listener拥有的线程executor执行
4.   } catch (Exception e) { // sneaky checked exception
5.     // Log it and keep going -- bad runnable and/or executor. Don't punish the other runnables if
6.     // we're given a bad one. We only catch Exception because we want Errors to propagate up.
7.     log.get()
8.         .log(
9.             Level.SEVERE,
10.             "RuntimeException while executing runnable "
11.                 + runnable
12.                 + " with executor "
13.                 + executor,
14.             e);
15.   }
16. }

复制代码

那么如果没有完成呢，在listener链表中的什么时候会执行？看后续的回调函数的触发内容

addCallback方法

Futures类还提供了另一个回调方法：addCallback方法

1. public static <V> void addCallback(
2.   final ListenableFuture<V> future,
3.   final FutureCallback<? super V> callback,
4.   Executor executor) {
5.         Preconditions.checkNotNull(callback);
6.         future.addListener(new CallbackListener<V>(future, callback), executor);//调用了addListener方法
7. }

复制代码

那哪些方法会来调用这个complete方法呢？

Service

Guava 的 Service 框架是一个用于管理服务生命周期的轻量级框架，可以资助我们把异步操作封装成一个Service服务。让这个服务有了运行状态(也可以理解成生命周期)，如许可以实时了解当前服务的运行状态。同时还可以添加监听器来监听服务运行状态之间的变化。
Guava里面的服务有五种状态，如下所示：

• Service.State.NEW: 服务创建状态
  
• Service.State.STARTING: 服务启动中
  
• Service.State.RUNNING：服务启动完成，正在运行中
  
• Service.State.STOPPING: 服务停止中
  
• Service.State.TERMINATED: 服务停止完成，竣事
  

全部的服务都需要实现Service接口，里面包罗了服务需要实现的一些基本方法，以下是Service接口：

1. private static final class CallbackListener<V> implements Runnable {
2.     final Future<V> future;
3.     final FutureCallback<? super V> callback;
5.     CallbackListener(Future<V> future, FutureCallback<? super V> callback) {
6.       this.future = future;
7.       this.callback = callback;
8.     }
10.     @Override
11.     public void run() {//回调时的逻辑
12.       if (future instanceof InternalFutureFailureAccess) {
13.         Throwable failure =
14.             InternalFutures.tryInternalFastPathGetFailure((InternalFutureFailureAccess) future);
15.         if (failure != null) {
16.           callback.onFailure(failure);
17.           return;
18.         }
19.       }
20.       final V value;
21.       try {
22.         value = getDone(future);//获取返回值
23.       } catch (ExecutionException e) {
24.         callback.onFailure(e.getCause());//如果发生了异常，则会调用onFailure方法通知异常
25.         return;
26.       } catch (RuntimeException | Error e) {
27.         callback.onFailure(e);//如果发生了异常，则会调用onFailure方法通知异常
28.         return;
29.       }
30.       callback.onSuccess(value);//将返回值调用FutureCallback实例的onSuccess方法执行注册的回调逻辑
31.     }
32. }

复制代码

那应该怎么来使用Service，需要实现的异步逻辑包装成服务呢．Guava里面已经给提供了三个底子实现类：

• AbstractService
  
• AbstractExecutionThreadService
  
• AbstractScheduledService
  

AbstractExecutionThreadService

AbstractExecutionThreadService可以把一个详细的异步操作封装成Service服务。说白了就是把之前在线程的实现逻辑封装成服务，把之前线程的详细实现逻辑搬到AbstractExecutionThreadService的实现方法run()方法去执行。
常用方法介绍

首先AbstractExecutionThreadService实现了Service，Service的方法在AbstractExecutionThreadService里面都有，AbstractExecutionThreadService新加了一些方法。如下所示：

1. private static void complete(AbstractFuture<?> param) {
2.     // Declare a "true" local variable so that the Checker Framework will infer nullness.
3.     AbstractFuture<?> future = param;//获取future
5.     Listener next = null;
6.     outer:
7.     while (true) {
8.       future.releaseWaiters();//通知所有执行的方法
9.       // We call this before the listeners in order to avoid needing to manage a separate stack data
10.       // structure for them.  Also, some implementations rely on this running prior to listeners
11.       // so that the cleanup work is visible to listeners.
12.       // afterDone() should be generally fast and only used for cleanup work... but in theory can
13.       // also be recursive and create StackOverflowErrors
14.       future.afterDone();
15.       // push the current set of listeners onto next
16.       next = future.clearListeners(next);//反转listener链表
17.       future = null;
18.       while (next != null) {
19.         Listener curr = next;//获取当前listener
20.         next = next.next;
21.         /*
22.          * requireNonNull is safe because the listener stack never contains TOMBSTONE until after
23.          * clearListeners.
24.          */
25.         Runnable task = requireNonNull(curr.task);
26.         if (task instanceof SetFuture) {
27.           SetFuture<?> setFuture = (SetFuture<?>) task;
28.           // We unwind setFuture specifically to avoid StackOverflowErrors in the case of long
29.           // chains of SetFutures
30.           // Handling this special case is important because there is no way to pass an executor to
31.           // setFuture, so a user couldn't break the chain by doing this themselves.  It is also
32.           // potentially common if someone writes a recursive Futures.transformAsync transformer.
33.           future = setFuture.owner;
34.           if (future.value == setFuture) {
35.             Object valueToSet = getFutureValue(setFuture.future);
36.             if (ATOMIC_HELPER.casValue(future, setFuture, valueToSet)) {
37.               continue outer;
38.             }
39.           }
40.           // other wise the future we were trying to set is already done.
41.         } else {
42.           /*
43.            * requireNonNull is safe because the listener stack never contains TOMBSTONE until after
44.            * clearListeners.
45.            */
46.           executeListener(task, requireNonNull(curr.executor));// 交给listener拥有的线程池进行处理
47.         }
48.       }
49.       break;
50.     }
51.   }

复制代码

AbstractExecutionThreadService类里面最重要的就是run()方法了，这个方法是服务需要详细实现的方法，服务需要处理的详细逻辑在这个方法里面做。
详细使用

1. public interface Service {
2.         //启动当前服务,只有当服务的状态是NEW的情况下才可以启动，否则抛出IllegalStateException异常
3.         @CanIgnoreReturnValue
4.         Service startAsync();
6.         //判断当前服务是否处在运行状态 (RUNNING)
7.         boolean isRunning();
9.         //获取当前服务的状态
10.         Service.State state();
12.         //停止当前服务
13.         @CanIgnoreReturnValue
14.         Service stopAsync();
16.         // 等待当前服务到达RUNNING状态
17.         void awaitRunning();
20.         // 在指定的时间内等待当前服务到达RUNNING状态，如果在指定时间没有达到则抛出TimeoutException
21.         void awaitRunning(long timeout, TimeUnit unit) throws TimeoutException;
23.         // 等待当前服务到达TERMINATED状态
24.         void awaitTerminated();
26.         //在指定的时间内等待当前服务达到TERMINATED状态，
27.         void awaitTerminated(long timeout, TimeUnit unit) throws TimeoutException;
29.         // 获取服务器失败的原因,在服务是FAILED的状态的时候调用该函数，否则抛出IllegalStateException异常
30.         Throwable failureCause();
32.         //监听当前服务的状态改变，executor参数表示，监听回调函数在哪里执行
33.         void addListener(Service.Listener listener, Executor executor);
34. }

复制代码

AbstractScheduledService

AbstractScheduledService可以把周期性的使命封装成一个服务。线程池也有一个周期性的线程池么，两者是一一对应的．
常用方法介绍

AbstractScheduledService也是一个服务所以Service里面的方法AbstractScheduledService也都有，同时，AbstractScheduledService也新增了一些其它方法

1. public class AbstractExecutionThreadService {
3.         // 开始执行服务逻辑的时候会调用，可以在里面做一些初始化的操作
4.         protected void startUp() throws Exception;
6.         // 当前服务需要执行的具体逻辑
7.         protected abstract void run() throws Exception;
9.         // 服务停止之后会调用的函数，可以在里面做 一些释放资源的处理
10.         protected void shutDown() throws Exception {}
12.     //比如在run方法里面有一个无线循环，可以在这个方法里面置状态，退出无线循环，让服务真正停止
13.         //调stopAsync函数的时候，会调用该方法
14.         protected void triggerShutdown() {}
15.        
16.         ...
17. }

复制代码

详细使用

自界说一个类继承AbstractScheduledService，实现一个非常简朴的周期性使命．

1. public class AbstractExecutionThreadServiceImpl extends AbstractExecutionThreadService {
3.         private volatile boolean running = true; //声明一个状态
5.         @Override
6.         protected void startUp() {
7.                 //TODO: 做一些初始化操作
8.         }
10.         @Override
11.         public void run() {
12.                 // 具体需要实现的业务逻辑，会在线程中执行
13.                 while (running) {
14.                         try {
15.                                 // 等待2s
16.                                 Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
17.                                 System.out.println("do our work.....");
18.                         } catch (Exception e) {
19.                                 //TODO: 处理异常，这里如果抛出异常，会使服务状态变为failed同时导致任务终止。
20.                         }
21.                 }
22.         }
24.         @Override
25.         protected void triggerShutdown() {
26.                 //TODO: 如果的run方法中有无限循环，可以在这里置状态，让其退出无限循环，stopAsync()里面会调用到该方法
27.                 running = false; //这里改变状态值，run方法中就能够得到响应。
28.         }
30.         @Override
31.         protected void shutDown() throws Exception {
32.                 //TODO: 可以做一些清理操作，比如关闭连接。shutDown() 是在线程的具体实现里面调用的
33.         }
34. }

复制代码

ServiceManager

ServiceManager是用来管理多个服务的，让对多个服务的操作变的更加方便，比如可以同时去启动多个服务，同时去停止多个服务等等。
常用方法介绍

[code]public class ServiceManager {        //构造函数，管理多个Service服务        public ServiceManager(Iterable