并发编程 Promise, Future 和 Callback

时间 2019-11-08

标签并发编程 promise future callback 繁體版

原文原文链接

在并发编程中，咱们一般会用到一组非阻塞的模型：Promise，Future 和 Callback。其中的 Future 表示一个可能尚未实际完成的异步任务的结果，针对这个结果能够添加 Callback 以便在任务执行成功或失败后作出对应的操做，而 Promise 交由任务执行者，任务执行者经过 Promise 能够标记任务完成或者失败。能够说这一套模型是不少异步非阻塞架构的基础。html

这一套经典的模型在 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 的依赖，实现一个简化版本。java

参考 Scala 和 Netty 的代码从新定义了接口和实现，先介绍下和 Netty 版本的区别:git

去除了对 EventLoop 的依赖，Callback 的执行策略不一样：任务未完成时添加的 Callback，会在结束任务的线程执行；任务完成后添加的 Callback 会在添加 Callback 线程当即执行github
一个 Callback 执行后会当即被清理编程
Callback 能够根据任务结果添加，支持添加如下三种 Callback: onComplete, onSuccess, onFailure, 不须要和 Netty 的 FutureListener 同样大部分场景下都须要检查 future.isSuccess 等api
支持 Callback 的组合，Callback 包含一些函数式的方法，好比 compose 和 andThen 能够用来组合promise
使用 CountdownLatch 替换掉了 Netty 的 wait/notify 实现架构
去掉 Netty Future 一些不常使用的方法，同时补充一些模型间关联的方法，好比 Promise.getFuture并发

而后再介绍几个使用这个 commons-future 的示例:oracle

异步执行任务，得到 Future 后添加 Callback

01 final TaskPromise promise = new DefaultTaskPromise();

02 final TaskFuture future = promise.getFuture();

03 final CountDownLatch latch = new CountDownLatch(1);

04 future.onComplete(new TaskCallback() { // 添加结束 Callback

05     @Override

06     public TaskFuture apply(TaskFuture f) {

07         latch.countDown();

08         return f;

09     }

10 });

11 new Thread(new Runnable() {

12     @Override

13     public void run() {

14         promise.setSuccess(null);

15     }

16 }).start();

17 latch.await();
异步执行任务，得到 Future 后添加成功结束的 Callback

01 final TaskPromise promise = new DefaultTaskPromise();

02 final TaskFuture future = promise.getFuture();

03 final CountDownLatch latch = new CountDownLatch(1);

04 future.onSuccess(new TaskCallback() { // 添加成功结束 Callback

05     @Override

06     public TaskFuture apply(TaskFuture f) {

07         latch.countDown();

08         return f;

09     }

10 });

11 new Thread(new Runnable() {

12     @Override

13     public void run() {

14         promise.setSuccess(null);

15     }

16 }).start();

17 latch.await();
异步执行任务，得到 Future 后，添加失败结束的组合 Callback

01 final TaskPromise promise = new DefaultTaskPromise();

02 final TaskFuture future = promise.getFuture();

03 final CountDownLatch latch = new CountDownLatch(2);

04 future.onFailure(new TaskCallback() {

05     @Override

06     public TaskFuture apply(TaskFuture f) {

07         latch.countDown();

08         return f;

09     }

10 }.andThen(new TaskCallback() {

11     @Override

12     public TaskFuture apply(TaskFuture f2) {

13         latch.countDown();

14         return f2;

15     }

16 }));

17 new Thread(new Runnable() {

18     @Override

19     public void run() {

20         promise.setFailure(new IllegalStateException("cm"));

21     }

22 }).start();

23 latch.await();
异步执行任务，得到 Future 后阻塞等待任务完成

01 final TaskPromise promise = new DefaultTaskPromise();

02 final TaskFuture future = promise.getFuture();

03 new Thread(new Runnable() {

04     @Override

05     public void run() {

06         try {

07             TimeUnit.SECONDS.sleep(2);

08         } catch (InterruptedException e) {

09         }

10         promise.setFailure(new IllegalStateException("cm"));

11     }

12 }).start();

13 future.await();

代码仓库: https://bitbucket.org/qiyi/commons-future

源文连接: http://isouth.org/archives/354.html

参考: