并发是一个困难问题,但是通过强大和强大的抽象能够显著的简化工作。为了简化问题,Gauva使用ListenableFuture扩展了JDK的Future接口,
**我们强烈建议你在你所有的代码中使用ListenableFuture代替Future,**因为:
传统的Future代表着异步计算的结果:一个计算可能也可能没有完成生成结果。Future可以是一个程序计算的句柄,一个来自服务的承诺提供给我们一个结果。
ListenableFuture允许你注册一个回调函数,一旦计算是完成或者计算是已经完成,回调函数会立马执行。这个简单的添加能有效的支持多种Future接口不支持的操作。
基础的操作被ListenableFuture添加是通过addListener(Runnable, Executor),它指定了当Future的计算完成之后,指定的Runnable将会在Executor中执行。
大多数用户更喜欢使用Futures.addCallback(ListenableFuture, FutureCallback, Executor),一个FutueCallback有两个方法:
相对于JDKExecutorService.submit(Callable) 方法初始化一个异步的计算。Guava提供了一个ListeningExecutorService 接口,它能在任何ExecutorService返回一个正常Future的地方,返回一个ListenableFuture。为了将ExecutorService转换成ListeningExecutorService ,仅需要使用MoreExecutors.listeningDecorator(executorService):
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool( Runtime.getRuntime().availableProcessors())); ListenableFuture<String> listenableFuture = service.submit(() -> callDemo()); Futures.addCallback(listenableFuture, new FutureCallback<String>() { @Override public void onSuccess(String result) { System.out.println("Run Ok : " + result); // service.shutdown(); } @Override public void onFailure(Throwable t) { System.out.println("Run Fail: " + t); } }, service); |
或者,如果你从一个基于FutureTask的API进行转换,Guava提供了ListenableFutureTask.create(Callable)和ListenableFutureTask.create(Runnable, V) 。并不像JDK,ListenableFutureTask并不意味着直接扩展。
如果您更喜欢设置未来值的抽象而不是实现计算值的方法,请考虑扩展 AbstractFuture<V> 或直接使用 SettableFuture。
如果你必须将另一个API提供的Future转换成ListenableFuture,你没有选择除了使用重量级的JdkFutureAdapters.listenInPoolThread(Future) ,将Future转化成ListenableFuture。无论什么时候,只要有可能,都应该修改源代码去返回一个ListenableFuture。
使用ListenableFuture的重要原因是,它让异步的链操作成为可能。
很多有效的其它操作使用ListenableFuture是支持的,而单独使用Future是不支持的。不同的操作能够被不同的Executor执行,并且单个 ListenableFuture 可以有多个等待它的操作。
当多个操作应该在另一个操作开始时立即开始时——“扇出”——ListenableFuture 才起作用:它触发所有请求的回调。通过稍微多一点的工作,我们能“扇入”,或触发ListenableFuture进行计算只要若干个其它的Future执行完(看:Futures.allAsList)。
方法 | 描述 | 延伸 |
---|---|---|
Futures.transformAsync(ListenableFuture, AsyncFunction, Executor) | 返回一个新的ListenableFuture,它的结果是应用AsyncFunction。AsyncFunction是对ListenableFuture结果的加工 | |
Futures.transform(ListenableFuture, Function, Executor) | 返回一个新的ListenableFuture,它的结果是应用Function。Function是对ListenableFuture结果的加工 | |
Futures.allAsList(Iterable<ListenableFuture<V>>) | 返回一个ListenableFuture,它的值是一个list包含了每一个输入输入future的值。如果任何的future失败或取消,这个future也失败或取消。 | |
Futures.successfulAsList(Iterable<ListenableFuture>) | 返回一个ListenableFuture,它的值是一个list包含了每一个输入输入future的值。如果相对应的future失败或取消,将用null代替。 |
一个AsyncFunction 提供了一个方法ListenableFuture<B> apply(A a)。它用于异步传输一个值。
1 2 |
ListenableFuture<ListenableFuture<String>> nestedFuture = executorService.submit(() -> otherExecutorService.submit(() -> callDemo())); |
上面的代码是不正确的, 因为如果外部Future的取消与外部Future的完成竞争,则该取消不会传播到内部Future。 使用 get() 或侦 听器检查其他Future的失败也是一个常见错误,但除非特别注意,否则从 otherCallable 抛出的异常将被抑制。
为了避免这种情况,Guava 的所有Future处理方法(以及一些来自 JDK)都有 *Async 版本,可以安全地解开这个嵌套: