并发编程 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();
    

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

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

参考:

相关文章
|
5月前
|
存储 前端开发 安全
C++一分钟之-未来与承诺:std::future与std::promise
【6月更文挑战第27天】`std::future`和`std::promise`是C++异步编程的关键工具,用于处理未完成任务的结果。`future`代表异步任务的结果容器,可阻塞等待或检查结果是否就绪;`promise`用于设置`future`的值,允许多线程间通信。常见问题包括异常安全、多重获取、线程同步和未检查状态。解决办法涉及智能指针管理、明确获取时机、确保线程安全以及检查未来状态。示例展示了使用`std::async`和`future`执行异步任务并获取结果。
102 2
|
2月前
|
前端开发 JavaScript
解决异步问题,教你如何写出优雅的promise和async/await,告别callback回调地狱!
该文章教授了如何使用Promise和async/await来解决异步编程问题,从而避免回调地狱,使代码更加清晰和易于管理。
解决异步问题,教你如何写出优雅的promise和async/await,告别callback回调地狱!
|
6月前
|
前端开发 Go
Golang深入浅出之-Go语言中的异步编程与Future/Promise模式
【5月更文挑战第3天】Go语言通过goroutines和channels实现异步编程,虽无内置Future/Promise,但可借助其特性模拟。本文探讨了如何使用channel实现Future模式,提供了异步获取URL内容长度的示例,并警示了Channel泄漏、错误处理和并发控制等常见问题。为避免这些问题,建议显式关闭channel、使用context.Context、并发控制机制及有效传播错误。理解并应用这些技巧能提升Go语言异步编程的效率和健壮性。
334 5
Golang深入浅出之-Go语言中的异步编程与Future/Promise模式
|
6月前
|
前端开发 JavaScript
js开发:请解释Promise是什么,以及它如何解决回调地狱(callback hell)问题。
Promise是JavaScript解决异步操作回调地狱的工具,代表未来可能完成的值。传统的回调函数嵌套导致代码难以维护,而Promise通过链式调用`.then()`和`.catch()`使异步流程清晰扁平。每个异步操作封装为Promise,成功时`.then()`传递结果,出错时`.catch()`捕获异常。ES6的`async/await`进一步简化Promise的使用,使异步代码更接近同步风格。
91 1
|
6月前
|
存储 前端开发 安全
【C++并发编程】std::future、std::async、std::packaged_task与std::promise的深度探索(三)
【C++并发编程】std::future、std::async、std::packaged_task与std::promise的深度探索
81 0
|
6月前
|
存储 设计模式 前端开发
【C++并发编程】std::future、std::async、std::packaged_task与std::promise的深度探索(二)
【C++并发编程】std::future、std::async、std::packaged_task与std::promise的深度探索
117 0
|
6月前
|
并行计算 前端开发 安全
【C++并发编程】std::future、std::async、std::packaged_task与std::promise的深度探索(一)
【C++并发编程】std::future、std::async、std::packaged_task与std::promise的深度探索
186 0
|
6月前
|
前端开发 C++
C++11实用技术(三)std::future、std::promise、std::packaged_task、async
C++11实用技术(三)std::future、std::promise、std::packaged_task、async
93 0
|
前端开发 C++ 容器
C++并发与多线程(四)async、future、packaged_task、promise、shared_future(下)
C++并发与多线程(四)async、future、packaged_task、promise、shared_future(下)
|
前端开发 Go C++
C++并发与多线程(四)async、future、packaged_task、promise、shared_future(上)
C++并发与多线程(四)async、future、packaged_task、promise、shared_future(上)
114 0