首页 > 代码库 > ListenableFuture in Guava

ListenableFuture in Guava

ListenableFuture的说明 

  并发编程是一个难题,但是一个强大而简单的抽象可以显著的简化并发的编写。出于这样的考虑,Guava 定义了 ListenableFuture接口并继承了JDK concurrent包下的Future 接口,ListenableFuture 允许你注册回调方法(callbacks),在运算(多线程执行)完成的时候进行调用,  或者在运算(多线程执行)完成后立即执行。这样简单的改进,使得可以明显的支持更多的操作,这样的功能在JDK concurrent中的Future是不支持的。 在高并发并且需要大量Future对象的情况下,推荐尽量使用ListenableFuture来代替..

  ListenableFuture 中的基础方法是addListener(Runnable, Executor), 该方法会在多线程运算完的时候,在Executor中执行指定的Runnable。

ListenableFuture的创建和使用

  对应JDK中的 ExecutorService.submit(Callable) 提交多线程异步运算的方式,Guava 提供了ListeningExecutorService 接口, 该接口返回 ListenableFuture, 而相应的ExecutorService 返回普通的 Future。将 ExecutorService 转为 ListeningExecutorService,可以使用MoreExecutors.listeningDecorator(ExecutorService)进行装饰。举例说明:

  

ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));

然后我们可以向这个ListeningExecutorService提交Callable任务

final ListenableFuture<String> future =  pool.submit(new Callable<String>() {            @Override            public String call() throws Exception {                Thread.sleep(1000*3);                return     "Task done !";            }        });

然后我们添加Listener:

future.addListener(new Runnable() {                @Override                public void run() {                    try {                        final String contents = future.get();                        System.out.println(contents);                    } catch (InterruptedException e) {                        e.printStackTrace();                    } catch (ExecutionException e) {                        e.printStackTrace();                    }                }            }, MoreExecutors.sameThreadExecutor());

我们看看上面的代码,确实不怎么优雅,我们需要处理抛出的异常,需要自己通过future.get()获得前面计算的值。有没有更加简便的方法呢?当然有,Guava提供了一个简便方法来替代上面的写法:

Futures.addCallback(future, new FutureCallback<String>() {                @Override                public void onSuccess(String result) {                    System.out.println(result);                }                @Override                public void onFailure(Throwable t) {                    t.printStackTrace();                }            });

完成代码如下:

package concurrency;import com.google.common.util.concurrent.FutureCallback;import com.google.common.util.concurrent.Futures;import com.google.common.util.concurrent.ListenableFuture;import com.google.common.util.concurrent.ListeningExecutorService;import com.google.common.util.concurrent.MoreExecutors;import java.util.concurrent.Callable;import java.util.concurrent.Executors;/** * Created by hupeng on 2014/9/24. */public class ListenableFutureTest {    public static void main(String[] args) throws InterruptedException {        ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));        final ListenableFuture<String> future = pool.submit(new Callable<String>() {            @Override            public String call() throws Exception {                Thread.sleep(1000 * 2);                return "Task done !";            }        });//            future.addListener(new Runnable() {//                @Override//                public void run() {//                    try {//                        final String contents = future.get();//                        System.out.println(contents);//                    } catch (InterruptedException e) {//                        e.printStackTrace();//                    } catch (ExecutionException e) {//                        e.printStackTrace();//                    }//                }//            }, MoreExecutors.sameThreadExecutor());        Futures.addCallback(future, new FutureCallback<String>() {            @Override            public void onSuccess(String result) {                System.out.println(result);            }            @Override            public void onFailure(Throwable t) {                t.printStackTrace();            }        });        Thread.sleep(5 * 1000);  //wait for task done        pool.shutdown();    }}

 

ListenableFuture in Guava