hangscer

scala-Future和Promise

2017/05/22

首先说明同步与异步,阻塞与非阻塞的问题:
Asynchronous vs. Synchronous

A method call is considered synchronous if the caller cannot make progress until the method returns a value or throws an exception. On the other hand, an asynchronous call allows the caller to progress after a finite number of steps, and the completion of the method may be signalled via some additional mechanism (it might be a registered callback, a Future, or a message).
A synchronous API may use blocking to implement synchrony, but this is not a necessity. A very CPU intensive task might give a similar behavior as blocking. In general, it is preferred to use asynchronous APIs, as they guarantee that the system is able to progress.

Non-blocking vs. Blocking

We talk about blocking if the delay of one thread can indefinitely delay some of the other threads. A good example is a resource which can be used exclusively by one thread using mutual exclusion. If a thread holds on to the resource indefinitely (for example accidentally running an infinite loop) other threads waiting on the resource can not progress. In contrast, non-blocking means that no thread is able to indefinitely delay others.
Non-blocking operations are preferred to blocking ones, as the overall progress of the system is not trivially guaranteed when it contains blocking operations.

以上文献摘自akka文档,一个方法之所以被称为同步方法,是因为直到该方法返回某值或者抛出异常,该方法的调用者才能得到结果(make progress)。如果一个异步调用需要通过额外的机制(比如callback,Future,message)。如果一个线程的延迟导致了另一个(一些)线程的延迟,那么久出现了阻塞(blocking)。一个例子就是一个资源被一个线程所独占,那么其他线程需要等待这个线程释放资源才能继续执行。

scala中的FuturePromise都是非阻塞的执行,既可以通过回调函数获取结果,但是也可以通过阻塞的方法串行获取结果。

Future

一个Future会持有一个值,虽然这个值在将来某个时间点才可用。

  1. 如果计算未完成,那么这个Future就未完成。
  2. 如果计算完成(得到结果或者异常),这个Future就完成了。

Future只能被指派一次,一旦Future给定了一个值或异常,它的结果不能修改。

1
2
3
4
5
6
7
8
9
10
11
object Main extends App {
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
val f:Future[Int]=Future{
Thread.sleep(100)//模拟某个耗时操作 比如网络请求
println("haha")
10
}
Thread.sleep(1000)
}
//haha

异步方法获取结果

目的获取结果,而不是控制执行过程。
scala提供了onSuccess等回调函数。其签名为:def onComplete[U](f: Try[T] => U)(implicit executor: ExecutionContext): Unit

1
2
3
4
f.onComplete({
case Success(i) => println(i)
case Failure(e) => e.printStackTrace()
})

以上代码采用偏函数形式,或者:

1
2
3
4
f.onComplete(result=>result match {
case Success(i)=>println(i)
case Failure(e)=>e.printStackTrace()
})

还可以注册多个回调:

1
2
3
4
5
6
f.onComplete(result=>result match {
case Success(i)=>println(i)
})
f.onComplete(result=>result match {
case Success(i)=>println(i+20)
})

注:多个回调函数之间并不保证执行顺序

同步方法获取结果

通过Await.result可以同步获取结果,或者超时或者异常。Await.ready等待计算完成,不返回结果。

1
val r=Await.result(f,Duration.Inf) //Await.result(f,1 seconds)

Promise

除了通过Future.apply创建Future对象,还可以使用Promise.future。如果说Future是一个只读的,值还没计算的占位符。那么Promise就是一个可写的,单次指派的容器。Promise可以通过调用success代表Future成功完成,failure方法抛出异常。或者更抽象的complete

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
object Main extends App {
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future,Promise}
// import scala.concurrent.duration._
val p=Promise[Int]
val f=p.future
val producer=Future{
p complete Try{
100
}
}
val consumer=Future{
f onComplete{
case Success(i)=>println(i)
case Failure(e)=>e.printStackTrace()
}
}
Thread.sleep(1000)
}