文章目錄

• • 案例
  • join實現
  • FutureTask
  • Guava 的異步回調
  • 擴展
  • 關于我

案例

在深入理解 異步回調模式前，我們以一個經典案例來說明，即
數學家華羅庚先生的文章《統籌方法》，介紹了一個燒水泡茶的例子，文中提到最優的工序應該是下面這樣

用兩個線程 T1 和 T2 來完成燒水泡茶程序，T1 負責洗水壺、燒開水、泡茶這三道工序，T2 負責洗茶壺、洗茶杯、拿茶葉三道工序，其中 T1 在執行泡茶這道工序時需要等待 T2 完成拿茶葉的工序

這里實現方式有多中，我們先一最簡單的線程匯總的join來實現

join實現

在編寫程序前我們先要了解一下join的基礎知識

• A線程調用B線程的join方法，在B線程沒有執行完成錢，A線程一直處于阻塞狀態
• join是實例方法，需要用線程對象去調用
• 使用join線程合并線程無法獲取到合并線程的返回值，即無法知道燒水線程執行的結果。只能一直阻塞等待燒水線程結束

public class JoinDemo {public static final int SLEEP_TIME = 1000;public static void main(String[] args) {Thread hThread = new Thread(() -> {try {Thread.currentThread().setName("燒水線程");System.out.println("洗好水壺");System.out.println("灌好涼水");System.out.println("放在火上");Thread.sleep(SLEEP_TIME);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("水燒開了");});hThread.start();Thread wThread = new Thread(() -> {try {Thread.currentThread().setName("清洗線程");System.out.println("洗茶壺");System.out.println("洗茶杯");System.out.println("拿茶葉");Thread.sleep(SLEEP_TIME);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("洗茶葉完成");});wThread.start();// 主線程 1. 合并燒水線程try {hThread.join();wThread.join();System.out.println("泡泡茶喝");} catch (InterruptedException e) {e.printStackTrace();}} }

運行結果：

結果很完美，可惜我們無法獲取到燒水和洗茶葉線程的執行結果，如果要獲取到他們的執行結果我們就需要使用jdk1.5版本后提供的新的多線程類：
FutureTask

FutureTask

jdk為了彌補Runnable接口沒有返回值的問題，jdk提供了Callable接口

Callable接口是一個泛型接口，也是函數式接口，異常為Exception，不用強制捕獲，Callable接口和Runnable接口相比功能更強大，但是有個問題就是Callable接口實例不能作為Thread線程實例的target來使用，而Runnable接口可以直接作為Thread的構造參數傳入開啟線程。
而且Java線程類中只有Thread，為了解決這個問題，就有了我們今天的FutureTask類
FutureTask 的構造方法

public FutureTask(Callable<V> callable) {if (callable == null)throw new NullPointerException();this.callable = callable;this.state = NEW; // ensure visibility of callable}

FutureTask 接口主要的功能有如下三個：

判斷并發任務是否執行完成

獲取并發任務完成后的結果

取消并發執行中的任務

// 獲取任務執行結果，如果任務沒有執行完成，調用次方法線程會一直阻塞，知道任務完成 V get() // 和上面一樣功能，不同的是多了阻塞時間限制，在限制時間內沒有完成則拋出異常 public V get(long timeout, TimeUnit unit) // 獲取任務執行狀態，執行完成返回true public boolean isDone() // 獲取任務的取消狀態，如果任務完成前被取消，則返回true public boolean isCancelled() // 取消任務執行 public boolean cancel(boolean mayInterruptIfRunning)

在介紹完 FutureTask 類后我們來基于FutureTask 實現上面泡茶喝案例

public class FutureTaskDemo {public static final int SLEEP_TIME = 1000;public static void main(String[] args) {Callable<Boolean> hotWaterJob = new HotWaterJob();FutureTask<Boolean> hotWaterTask = new FutureTask<>(hotWaterJob);Thread hotWaterThread = new Thread(hotWaterTask, "燒水線程");Callable<Boolean> washJob = new WashJob();FutureTask<Boolean> washTask = new FutureTask<>(washJob);Thread washThread = new Thread(washTask, "清洗線程");hotWaterThread.start();washThread.start();try {Boolean hotWaterFlag = hotWaterTask.get();Boolean washFlag = washTask.get();drinkTea(hotWaterFlag, washFlag);} catch (Exception e) {e.printStackTrace();}}private static void drinkTea(Boolean hotWaterFlag, Boolean washFlag) {if (hotWaterFlag && washFlag) {System.out.println("喝茶");}}static class HotWaterJob implements Callable<Boolean> {@Overridepublic Boolean call() throws Exception {try {Thread.currentThread().setName("燒水線程");System.out.println("洗好水壺");System.out.println("灌好涼水");System.out.println("放在火上");Thread.sleep(SLEEP_TIME);} catch (InterruptedException e) {e.printStackTrace();return false;}System.out.println("水燒開了");return true;}}static class WashJob implements Callable<Boolean> {@Overridepublic Boolean call() throws Exception {try {Thread.currentThread().setName("清洗線程");System.out.println("洗茶壺");System.out.println("洗茶杯");System.out.println("拿茶葉");Thread.sleep(SLEEP_TIME);} catch (InterruptedException e) {e.printStackTrace();return false;}System.out.println("洗茶葉完成");return true;}} }

上面 通過 FutureTask.get處于異步阻塞狀態，在清洗燒水沒有完成前，主線程是做不了任何其他事情的，而異步阻塞效率往往比較低下，只是我們考慮使用異步回調機制，異步回調我們就需要使用一些三方jar了，比如常用的Guava FutureCallback

Guava 的異步回調

Guava對Java 的異步回調做了增強

引入新的接口ListenableFuture,繼承于Future接口，使得Java的Future異步任務再Guava中能被監控和獲取非阻塞的異步執行結果

引入新的接口 FutureCallback，更具異步結果完成不同的回調處理

要獲取 ListenableFuture實例必須通過線程池（Guava的線程池）來提交Callable任務來獲取

我們來基于Guava來實現燒水泡茶喝案例：

public class GuavaFutureDemo {public static final int SLEEP_TIME = 1000;public static void main(String[] args) {MainJob mainJob = new MainJob();new Thread(mainJob).start();// 燒水Callable<Boolean> hotWater = new HotWaterJob();Callable<Boolean> washJob = new WashJob();ExecutorService pool = Executors.newFixedThreadPool(5);// 構造guava線程池ListeningExecutorService guavaPool = MoreExecutors.listeningDecorator(pool);// 提交燒水任務ListenableFuture<Boolean> hotFuture = guavaPool.submit(hotWater);// 綁定異步回調，燒水完成后，將燒水任務的 waterFlag 設置為trueFutures.addCallback(hotFuture, new FutureCallback<Boolean>() {@Overridepublic void onSuccess(@Nullable Boolean aBoolean) {if (aBoolean) {mainJob.waterFlag = true;}}@Overridepublic void onFailure(Throwable throwable) {System.out.println("燒水失敗");}}, guavaPool);// 提交清洗茶杯任務ListenableFuture<Boolean> washFuture = guavaPool.submit(washJob);Futures.addCallback(washFuture, new FutureCallback<Boolean>() {@Overridepublic void onSuccess(@Nullable Boolean aBoolean) {if (aBoolean) {mainJob.washFlag = true;}}@Overridepublic void onFailure(Throwable throwable) {System.out.println("清洗茶壺失敗");}}, guavaPool);}static class HotWaterJob implements Callable<Boolean> {@Overridepublic Boolean call() throws Exception {try {Thread.currentThread().setName("燒水線程");System.out.println("洗好水壺");System.out.println("灌好涼水");System.out.println("放在火上");Thread.sleep(SLEEP_TIME);} catch (InterruptedException e) {e.printStackTrace();return false;}System.out.println("水燒開了");return true;}}static class WashJob implements Callable<Boolean> {@Overridepublic Boolean call() throws Exception {try {Thread.currentThread().setName("清洗線程");System.out.println("洗茶壺");System.out.println("洗茶杯");System.out.println("拿茶葉");Thread.sleep(SLEEP_TIME);} catch (InterruptedException e) {e.printStackTrace();return false;}System.out.println("洗茶葉完成");return true;}}static class MainJob implements Runnable {public static boolean waterFlag = false;public static boolean washFlag = false;@Overridepublic void run() {while (!waterFlag && !washFlag) {try {System.out.println("等待燒水洗杯中");Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}if (waterFlag && washFlag) {drinkTea(washFlag, waterFlag);}}}private void drinkTea(boolean washFlag, boolean waterFlag) {if (washFlag && waterFlag) {System.out.println("燒水洗杯完成，開始喝茶");} else if(!washFlag){System.out.println("杯子清洗失敗");} else if(!waterFlag){System.out.println("燒是失敗");}}}}

擴展

實際Netty中也擴展了實現了自己的異步回調機制，感興趣的可以自己去了解

關于我

? 覺得文章不錯請掃碼關注我吧

?

總結

以上是生活随笔為你收集整理的Future异步回调详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。