GUAVA-ListenableFuture实现回调

Posted 2020-09-29 gaojy

随着软件开发的不断进步，在实际的开发应用中，可能一次请求需要查询若干次数据库或者调用若干次第三方，按照传统的串行执行的话，会大大增加响应时间，无法满足业务需求，更无法满足用户迫切需要响应迅速的愿望。对此，我们需要针对网络请求或内部调用中包含的“多任务”进行异步处理，并行去执行这些“任务”，这样就就会大大减小响应时间。本文是基于guava中的ListenableFuture来实现的。

 测试代码：

 1 package com.study1;
 2 
 3 import java.util.Random;
 4 import java.util.concurrent.Callable;
 5 import java.util.concurrent.Executors;
 6 import java.util.concurrent.locks.LockSupport;
 7 
 8 import com.google.common.util.concurrent.FutureCallback;
 9 import com.google.common.util.concurrent.Futures;
10 import com.google.common.util.concurrent.ListenableFuture;
11 import com.google.common.util.concurrent.ListeningExecutorService;
12 import com.google.common.util.concurrent.MoreExecutors;
13 /**
14  * GUAVA  ListenableFuture
15  * @author gaojy
16  *
17  */
18 public class TestListenableFuture {
19     //定义一个线程池，用于处理所有任务
20     final static ListeningExecutorService service 
21                 = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
22 
23     public static void main(String[] args){
24         Long t1 = System.currentTimeMillis();
25         // 任务1  
26         ListenableFuture<Boolean> booleanTask = service.submit(new Callable<Boolean>() {
27             @Override
28             public Boolean call() throws Exception {
29                 Thread.sleep(10000);
30                 return true;
31             }
32         });
33 
34         /**
35          * 
36          */
37         Futures.addCallback(booleanTask, new FutureCallback<Boolean>() {
38             @Override
39             public void onSuccess(Boolean result) {
40                 System.err.println("BooleanTask: " + result);
41             }
42 
43             @Override
44             public void onFailure(Throwable t) {
45             }
46         });
47 
48         // 任务2
49         ListenableFuture<String> stringTask = service.submit(new Callable<String>() {
50             @Override
51             public String call() throws Exception {
52                 Thread.sleep(10000);
53                 return "Hello World";
54             }
55         });
56 
57         Futures.addCallback(stringTask, new FutureCallback<String>() {
58             @Override
59             public void onSuccess(String result) {
60                 System.err.println("StringTask: " + result);
61             }
62 
63             @Override
64             public void onFailure(Throwable t) {
65             }
66         });
67 
68         // 任务3
69         ListenableFuture<Integer> integerTask = service.submit(new Callable<Integer>() {
70             @Override
71             public Integer call() throws Exception {
72                 Thread.sleep(10000);
73                 return new Random().nextInt(100);
74             }
75         });
76 
77         Futures.addCallback(integerTask, new FutureCallback<Integer>() {
78             @Override
79             public void onSuccess(Integer result) {
80                 try {
81                     Thread.sleep(5000);
82                 } catch (InterruptedException e) {
83                     e.printStackTrace();
84                 }
85                 System.err.println("IntegerTask: " + result);
86             }
87 
88             @Override
89             public void onFailure(Throwable t) {
90             }
91         });
92 
93         // 执行时间
94         System.err.println("time: " + (System.currentTimeMillis() - t1));
95         
96     }
97 }

 

测试结果：

源码分析：

在26行中ListeningExecutorService的submit（）方法：

 public ListenableFuture submit(Runnable task, Object result)
    {
        // 初始化了ListenableFutureTask对象
        ListenableFutureTask ftask = ListenableFutureTask.create(task, result);
        //执行task，实际上调用了Callable对象的call()方法
        execute(ftask);
        return ftask;
    }

再来查看一下Futures.addCallback的方法：

 public static void addCallback(ListenableFuture future, FutureCallback callback, Executor executor)
    {
      //对回调task进行检查
        Preconditions.checkNotNull(callback);
    //创建一个新的Runnable对象，并放到一个指定的线程中，执行。
   //这个Runnable对象主要任务就是获得future的结果，并根据结果调用回调函数的相应方法
        Runnable callbackListener = new Runnable(future, callback) {

            public void run()
            {
                Object value;
                try
                {
  //获取future执行结果，内部调用future.get（），产生堵塞，而不影响main主线程的执行，当获取到value时，就调用callback的onSuccess（）方法
                    value = Uninterruptibles.getUninterruptibly(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);
            }

            final ListenableFuture val$future;
            final FutureCallback val$callback;

            
            {
                future = listenablefuture;
                callback = futurecallback;
                super();
            }
        }
;
        future.addListener(callbackListener, executor);
    }