並發編程 Promise, Future 和 Callback
在並發編程中,我們通常會用到一組非阻塞的模型:Promise,Future 和 Callback。其中的 Future 表示一個可能還沒有實際完成的異步任務的結果,針對這個結果可以添加 Callback 以便在任務執行成功或失敗後做出對應的操作,而 Promise 交由任務執行者,任務執行者通過 Promise 可以標記任務完成或者失敗。 可以說這一套模型是很多異步非阻塞架構的基礎。
這一套經典的模型在 Scala、C# 中得到了原生的支持,但 JDK 中暫時還隻有無 Callback 的 Future 出現,當然也並非在 JAVA 界就沒有發展了,比如 Guava 就提供了ListenableFuture 接口,而 Netty 4+ 更是提供了完整的 Promise、Future 和 Listener 機製,在 Netty 的官方文檔 Using as a generic library 中也介紹了將 Netty 作為一個 lib 包依賴,並且使用 Listenable futures 的示例。在實際的項目使用中,發現 Netty 的 EventLoop 機製不一定適用其他場景,因此想去除對 EventLoop 的依賴,實現一個簡化版本。
參考 Scala 和 Netty 的代碼重新定義了接口和實現,先介紹下和 Netty 版本的區別:
- 去除了對 EventLoop 的依賴,Callback 的執行策略不同:任務未完成時添加的 Callback,會在結束任務的線程執行;任務完成後添加的 Callback 會在添加 Callback 線程立即執行
- 一個 Callback 執行後會立即被清理
- Callback 可以根據任務結果添加,支持添加以下三種 Callback: onComplete, onSuccess, onFailure, 不需要和 Netty 的 FutureListener 一樣大部分場景下都需要檢查 future.isSuccess 等
- 支持 Callback 的組合,Callback 包含一些函數式的方法,比如 compose 和 andThen 可以用來組合
- 使用 CountdownLatch 替換掉了 Netty 的 wait/notify 實現
- 去掉 Netty Future 一些不常使用的方法,同時補充一些模型間關聯的方法,比如 Promise.getFuture
然後再介紹幾個使用這個 commons-future 的示例:
- 異步執行任務,獲得 Future 後添加 Callback
final TaskPromise promise = new DefaultTaskPromise(); final TaskFuture future = promise.getFuture(); final CountDownLatch latch = new CountDownLatch(1); future.onComplete(new TaskCallback() { // 添加結束 Callback @Override public TaskFuture apply(TaskFuture f) { latch.countDown(); return f; } }); new Thread(new Runnable() { @Override public void run() { promise.setSuccess(null); } }).start(); latch.await();
- 異步執行任務,獲得 Future 後添加成功結束的 Callback
final TaskPromise promise = new DefaultTaskPromise(); final TaskFuture future = promise.getFuture(); final CountDownLatch latch = new CountDownLatch(1); future.onSuccess(new TaskCallback() { // 添加成功結束 Callback @Override public TaskFuture apply(TaskFuture f) { latch.countDown(); return f; } }); new Thread(new Runnable() { @Override public void run() { promise.setSuccess(null); } }).start(); latch.await();
- 異步執行任務,獲得 Future 後,添加失敗結束的組合 Callback
final TaskPromise promise = new DefaultTaskPromise(); final TaskFuture future = promise.getFuture(); final CountDownLatch latch = new CountDownLatch(2); future.onFailure(new TaskCallback() { @Override public TaskFuture apply(TaskFuture f) { latch.countDown(); return f; } }.andThen(new TaskCallback() { @Override public TaskFuture apply(TaskFuture f2) { latch.countDown(); return f2; } })); new Thread(new Runnable() { @Override public void run() { promise.setFailure(new IllegalStateException("cm")); } }).start(); latch.await();
- 異步執行任務,獲得 Future 後阻塞等待任務完成
final TaskPromise promise = new DefaultTaskPromise(); final TaskFuture future = promise.getFuture(); new Thread(new Runnable() { @Override public void run() { try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { } promise.setFailure(new IllegalStateException("cm")); } }).start(); future.await();
代碼倉庫: https://bitbucket.org/qiyi/commons-future
源文鏈接: https://isouth.org/archives/354.html
參考:
- https://docs.scala-lang.org/sips/completed/futures-promises.html
- https://scala-lang.org/
- https://github.com/netty/netty
- https://docs.oracle.com/javase/6/docs/api/java/util/concurrent/CountDownLatch.html
- https://biasedbit.com/countdownlatch-vs-wait-notify/
- 轉載自 並發編程網 - ifeve.com
最後更新:2017-05-23 18:02:22