Guava 並行 Futures實例

Future可以用來構建復雜的異步操作，方法不是返回一個值，而是一個Future對象。創建Future對象的過程（比如調用Future異步函數接口），不會阻塞當前線程操作，而且對象第一個次創建沒有值，但以后可以通過這個對象獲取這個值。Guava中的ListenableFuture接口對java.util.concurrent.Future接口做了進一步拓展，並且提供了Futures靜態工具類，大大方便了我們的使用。本文主要介紹Guava Future的使用，給出了幾個使用的例子。

Code Test Case

多任務並發執行,不阻塞當前線程

    @Test
    public void should_run_future_tasks_in_parallel() throws Exception {
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
        ListenableFuture<?> task1 = service.submit(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(2000);
                    System.out.println("future task1 done.....");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        ListenableFuture<?> task2 = service.submit(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(2000);
                    System.out.println("future task2 done.....");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        System.out.println("main task done.....");
        Thread.sleep(3000);
    }

可以使用get操作獲取Future值，阻塞當前線程，直到異步操作邏輯處理完畢

    @Test
    public void should_block_the_current_thread() throws Exception {
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
        ListenableFuture<Integer> task = service.submit(new Callable<Integer>() {
            public Integer call() throws Exception {
                Thread.sleep(2000);
                System.out.println("future task done......");
                return 1;
            }
        });
        System.out.println(task.get());
        System.out.println("main task done.....");
    }

get函數提供下面兩個接口，用戶可以根據需要選擇是否添加超時。

    V get() throws InterruptedException, ExecutionException; 
    V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;

處理異步Future，可以使用callback,在callback函數中對異步處理的結果進行處理。callback函數不阻塞當前線程。

    @Test
    public void should_call_back_the_future_task() throws Exception {
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
        ListenableFuture<Integer> task = service.submit(new Callable<Integer>() {
            public Integer call() throws Exception {
                Thread.sleep(2000);
                System.out.println("future task done......");
                return 1;
            }
        });
        Futures.addCallback(task, new FutureCallback() {
            public void onSuccess(Object o) {
                System.out.println("異步處理成功,result="+o);
            }

            public void onFailure(Throwable throwable) {
                System.out.println("異步處理失敗,e="+throwable);
            }
        });

        System.out.println("main task done.....");
        Thread.sleep(3000);
    }

例子中使用的Futures.addCallBack函數，第一個參數為我們要處理的異步Future task,它可以是一個數據庫處理，可以是一個外部模塊API請求等；第二個參數我們使用的是FutureCallBack匿名構造對象，對象內實現兩個方法，onSuccess和onFailure。future task處理成功，沒有任何異常則分支進入onSuccess處理，否則進入onFailure分支。

將Future對象轉化為另一個Future對象

例子中將task的結果轉化為Boolean類型的future對象

    @Test
    public void should_transform_to_another_future_obj() throws Exception {
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
        ListenableFuture<Integer> task = service.submit(new Callable<Integer>() {
            public Integer call() throws Exception {
                Thread.sleep(2000);
                System.out.println("future task done......");
                return 1;
            }
        });
        ListenableFuture<Boolean> transform = Futures.transform(task, new AsyncFunction<Integer, Boolean>() {
            public ListenableFuture<Boolean> apply(Integer integer) throws Exception {
                return integer > 0 ? Futures.immediateFuture(Boolean.TRUE):
                        Futures.immediateFuture(Boolean.FALSE);
            }
        });
        System.out.println(transform.get());
        Thread.sleep(3000);
    }

創建一個Future帶值對象（非null）可以使用下面的接口

    public static <V> ListenableFuture<V> immediateFuture(@Nullable V value) {

Conclusion

Future在異步處理方面具有強大的功能，在分布式系統中組件異步通信，具有很好的應用。本文給出了Guava Future使用的幾個實例，希望能對大家有所幫助。

更多ListenableFuture接口可以參考官方API文檔：
< https://github.com/google/guava/wiki/ListenableFutureExplained >