未来的邮箱:Promise

按照Erik的介绍,我们可以把Promise当作是装Future的邮箱/容器。
可从它的两个方法签名可以体会下,你可以向邮箱里放成功的数据或者失败的Exception。

def successful[T](result: T): Promise[T]
def failed[T](exception: Throwable): Promise[T]

放了值之后,可以调用Promise的future() 得到一个已经完成了的Future。
Promise最核心的就是这个逻辑:你可以通过Promise.future()得到一个Future对象,而future里的计算结果是在什么其它地方(当然一般情况是在其它线程里、回调代码里)计算好放进去的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
val p = Promise[Int]()
val f = p.future

def produce() = Future {
Thread.sleep(500)
p.success(1)
println("Produce done")
}

def consume() = Future {
f.foreach(r => println(s"Get $r"))
println(s"Consume done")
}

produce()
consume()
StdIn.readLine("End?\n")

这段代码的打印如下内容,可以看到consume()方法已经执行完了才打印出“Get 1”。

Consume done
End?
Produce done
Get 1

这个也好理解,foreach只是针对future Success的情况提供了callback机制。需要注意的是Future可以通过onComplete, foreach注册多个callback,但是这些callback运行先后以及运行所在线程是没有保证的。这点区别于map、flatMap。

本来想试着用Promise实现 List[Future[T]] 到 Future[List[T]]的转换,找到foldLeft这种方式。如下:

1
2
3
def sequence[T](fts: List[Future[T]]): Future[List[T]] = {
fts.foldLeft(Future{ List.empty[T]})((acc, ft) => acc.flatMap(ts => ft.map(t => ts :+ t)))
}

画蛇添足地再体验下promise:

1
2
3
4
5
6
7
def sequenceByPromise[T](fts: List[Future[T]]): Future[List[T]] = {
val p = Promise[List[T]]()
val result = p.success(List.empty[T]).future

fts.foldLeft(result)((acc, ft) => acc.flatMap(ts => ft.map(t => ts :+ t)))
result
}

下面代码演示了如何把callback风格代码转为Future风格。在Akka actor框架里,如果需要异步的执行代码并且后面的代码需要这个异步执行的结果,我们就可以通过Promise把结果封装到Future里。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

trait CallbackBasedApi {
def computeIntAsync(continuation: Try[Int] => Unit): Unit
}

trait FutureBasedApi {
def computeIntAsync(): Future[Int]
}

def futurize(callbackBasedApi: CallbackBasedApi): FutureBasedApi = {
val p = Promise[Int]()

// 体会下把“Try=>Unit”当做complete的参数。
callbackBasedApi.computeIntAsync( t => p.complete(t))

new FutureBasedApi {
def computeIntAsync() = p.future
}
}

Reference:
https://docs.scala-lang.org/overviews/core/futures.html