在并发编程中,我们通常会用到一组非阻塞的模型: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
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
参考:
- http://docs.scala-lang.org/sips/completed/futures-promises.html
- http://scala-lang.org/
- https://github.com/netty/netty
- http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/CountDownLatch.html
- http://biasedbit.com/countdownlatch-vs-wait-notify/
来源:http://isouth.org/archives/354.html
Scala的futue和promise
scala的future和promise
异步操作的有两个经典接口:Future
和Promise
,其中的 Future
表示一个可能还没有实际完成的异步任务的结果,针对这个结果可以添加 Callback
以便在任务执行成功或失败后做出对应的操作,而 Promise
交由任务执行者,任务执行者通过 Promise
可以标记任务完成或者失败。 可以说这一套模型是很多异步非阻塞架构的基础。
scala的Future
表示一个异步操作的结果状态,它持有一个值,在未来的某个之间点可用,该值是异步操作的结果,当异步操作没有完成,那么Future
的isCompleted
为false
,当异步操作完成了且返回了值,那么Future
的isCompleted
返回true
且是success, 如果异步操作没有完成或者异常终止,那么Future
的isCompleted
也返回true但是是failed.
scala的Future
有一个重要特性,就是他只能被赋值一次,一旦Future
对象被赋值了或者设置为异常,那么它会变成不可变的
创建一个Future
最简单的方式就是调用future
函数:它会启动一个异步操作且返回一个包含了操作结果的。一个future
函数完成该值就可以用了,Future[T]
的泛型是异步操作结果的类型。下面是一个创建Future
的例子。
package com.joy.lession1
import scala.concurrent._
import ExecutionContext.Implicits.global
object FutureTest extends App{
val s = "Hello"
val f: Future[String] = future {
s + " future!"
}
f onSuccess {
case msg => println(msg)
}
println(s) //不加这句, f onSuccess就不执行
}
Future的apply()方法会构建一个异步操作且在未来某一个时刻返回一个值。Future的apply()方法内部使用了Promise
。
下面给Future添加一个回调函数。
import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.util.{Failure, Success, Try}
object FutureTest extends App{
val s = "Hello"
val f: Future[String] = future {
s + " future!"
}
f onComplete {
case Success(t) =>
{
println(t)
}
case Failure(e) =>
{
println(s"An error has occured: $e.getMessage")
}
}
println("dd")
}
Future
的值不知道什么时候可用,所以需要一种机制来获取异步操作的结果,一种是不停的查看Future
的完成状态,另一个采用阻塞的方式,scala提供了第二种方式的支持,使用scala.concurrent.Await
,它有两个方法,一个是Await.ready
,当Future的状态为完成时返回,一种是Await.result
,直接返回Future
持有的结果。Future
还提供了一些map
,filter
,foreach
等操作
Scala的Promise
可以被认为一个是可写的,静态单赋值,它可以创建一个Future
和完成一个Future(success complete
和failed complete)
promise
字面意思是承诺
,以为他可以控制异步操作的结果
import java.util.Timer
import java.util.TimerTask
import scala.concurrent._
object TimedEvent {
val timer = new Timer
/** Return a Future which completes successfully with the supplied value after secs seconds. */
def delayedSuccess[T](secs: Int, value: T): Future[T] = {
val result = Promise[T]
timer.schedule(new TimerTask() {
def run() = {
result.success(value)
}
}, secs * 1000)
result.future
}
/** Return a Future which completes failing with an IllegalArgumentException after secs
* seconds. */
def delayedFailure(secs: Int, msg: String): Future[Int] = {
val result = Promise[Int]
timer.schedule(new TimerTask() {
def run() = {
result.failure(new IllegalArgumentException(msg))
}
}, secs * 1000)
result.future
}
作者:半兽人
链接:http://www.orchome.com/250