1. 程式人生 > >併發程式設計 Promise, Future 和 Callback

併發程式設計 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 版本的區別:

  1. 去除了對 EventLoop 的依賴,Callback 的執行策略不同:任務未完成時新增的 Callback,會在結束任務的執行緒執行;任務完成後新增的 Callback 會在新增 Callback 執行緒立即執行
  2. 一個 Callback 執行後會立即被清理
  3. Callback 可以根據任務結果新增,支援新增以下三種 Callback: onComplete, onSuccess, onFailure, 不需要和 Netty 的 FutureListener 一樣大部分場景下都需要檢查 future.isSuccess 等
  4. 支援 Callback 的組合,Callback 包含一些函式式的方法,比如 compose 和 andThen 可以用來組合
  5. 使用 CountdownLatch 替換掉了 Netty 的 wait/notify 實現
  6. 去掉 Netty Future 一些不常使用的方法,同時補充一些模型間關聯的方法,比如 Promise.getFuture

然後再介紹幾個使用這個 commons-future 的示例:

  1. 非同步執行任務,獲得 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();
    
  2. 非同步執行任務,獲得 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();
    
  3. 非同步執行任務,獲得 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();
    
  4. 非同步執行任務,獲得 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();
    

源文連結: http://isouth.org/archives/354.html

參考: