Guava併發：ListenableFuture與RateLimiter示例

ListenableFuture顧名思義就是能夠監聽的Future，它是對java原生Future的擴展加強 RateLimiter相似於JDK的信號量Semphore，他用來限制對資源併發訪問的線程數，本文介紹RateLimiter使用

Guava併發 ListenableFuture RateLimiter

 

目錄[-]

• 概念

• 代碼示例

• Guava版本

 源碼：http://www.jinhusns.com/Products/Download/?type=xcj

概念

        ListenableFuture 顧名思義就是能夠監聽的Future，它是對java原生Future的擴展加強。咱們知道Future表示一個異步計算任務，當任務完成時能夠獲得計算 結果。若是咱們但願一旦計算完成就拿到結果展現給用戶或者作另外的計算，就必須使用另外一個線程不斷的查詢計算狀態。這樣作，代碼複雜，並且效率低下。使用 ListenableFuture Guava幫咱們檢測Future是否完成了，若是完成就自動調用回調函數，這樣能夠減小併發程序的複雜度。      

        推薦使用第二種方法，由於第二種方法能夠直接獲得Future的返回值，或者處理錯誤狀況。本質上第二種方法是經過調動第一種方法實現的，作了進一步的封裝。

另外ListenableFuture還有其餘幾種內置實現：

1. SettableFuture：不須要實現一個方法來計算返回值，而只須要返回一個固定值來作爲返回值，能夠經過程序設置此Future的返回值或者異常信息

2. CheckedFuture： 這是一個繼承自ListenableFuture接口，他提供了checkedGet()方法，此方法在Future執行發生異常時，能夠拋出指定類型的異常。

    

    RateLimiter相似於JDK的信號量Semphore，他用來限制對資源併發訪問的線程數，本文介紹RateLimiter使用

 

代碼示例

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96

import  java.util.concurrent.Callable; import  java.util.concurrent.ExecutionException; import  java.util.concurrent.Executors; import  java.util.concurrent.TimeUnit;   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  com.google.common.util.concurrent.RateLimiter;   public  class  ListenableFutureDemo {      public  static  void  main(String[] args) {          testRateLimiter();          testListenableFuture();      }        /**       * RateLimiter相似於JDK的信號量Semphore，他用來限制對資源併發訪問的線程數       */      public  static  void  testRateLimiter() {          ListeningExecutorService executorService = MoreExecutors                  .listeningDecorator(Executors.newCachedThreadPool());            RateLimiter limiter = RateLimiter.create( 5.0 );  // 每秒不超過4個任務被提交            for  ( int  i =  0 ; i <  10 ; i++) {              limiter.acquire();  // 請求RateLimiter, 超過permits會被阻塞                final  ListenableFuture<Integer> listenableFuture = executorService                      .submit( new  Task( "is " + i));          }      }        public  static  void  testListenableFuture() {          ListeningExecutorService executorService = MoreExecutors                  .listeningDecorator(Executors.newCachedThreadPool());            final  ListenableFuture<Integer> listenableFuture = executorService                  .submit( new  Task( "testListenableFuture" ));                      //同步獲取調用結果          try  {              System.out.println(listenableFuture.get());          }  catch  (InterruptedException e1) {              e1.printStackTrace();          }  catch  (ExecutionException e1) {              e1.printStackTrace();          }                    //第一種方式          listenableFuture.addListener( new  Runnable() {              @Override              public  void  run() {                  try  {                      System.out.println( "get listenable future's result "                              + listenableFuture.get());                  }  catch  (InterruptedException e) {                      e.printStackTrace();                  }  catch  (ExecutionException e) {                      e.printStackTrace();                  }              }          }, executorService);            //第二種方式          Futures.addCallback(listenableFuture,  new  FutureCallback<Integer>() {              @Override              public  void  onSuccess(Integer result) {                  System.out                          .println( "get listenable future's result with callback "                                  + result);              }                @Override              public  void  onFailure(Throwable t) {                  t.printStackTrace();              }          });      } }   class  Task  implements  Callable<Integer> {      String str;      public  Task(String str){          this .str = str;      }      @Override      public  Integer call()  throws  Exception {          System.out.println( "call execute.."  + str);          TimeUnit.SECONDS.sleep( 1 );          return  7 ;      } }

 

 

Guava版本

?

1 2 3 4 5

< dependency >              < groupId >com.google.guava</ groupId >              < artifactId >guava</ artifactId >              < version >14.0.1</ version >          </ dependency >