Scala实战：并发-Future和Promise

并发编程是很困难的，特别是在你没有很好的设计与抽像你的功能层次时。传统的并发解决方案是采用多线程和共享变量，这使得随着代码的增长你很难找到错误根源。

Scala中采用了更好的方案，它不是只基于更低层次的线程的。Scala为用户提供了更高级的抽象：Futures和Promises（Akka还提供了基于actor模式的编程范式，是一种更高层次的并发编程抽象。本文主要讲解Futures和Promises，这里提供一些进一步学习的参考）。

Future

Future是持有某个值的对象，它完成一些计算并容许在“未来”的某一个时刻获取结果。它也能够是其它计算结果的结果（简单点说就是多个Future能够嵌套）。建立一个Future最简单的方式就调用它的apply方法，它将直接开始一个异步计算，并返回一个Future对象，它将包含计算结果。

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.{Success, Failure}

def computation(): Int = { 25 + 50 }
val theFuture = Future { computation() }

第一行导入了ExecutionContext.Implicits.global做为当前环境的一个默认执行上下文。如今先暂时无论它的具体含意，只要知道它会提供一个线程池，全部任务最终都会被提交给它来异步执行就能够了。在这个示例中先定义computation函数，并在Future { ... }代码块中调用。程序会使用上下文中的找到的线程池（由ExecutionContext.Implicits.global导入）并立刻开始异步执行。

访问Future的结果

前面说了，Future返回值有两种类型：Success和Failure。而Future在执行后提供了3个回调函数来让你访问结果，它们分别是：

• def onSuccess[U](pf: PartialFunction[T, U]): 传入一个偏函数，可使用模式匹配来处理你想要的结果
• def onFailure[U](pf: PartialFunction[Throwable, U]): 传入一个偏函数，可使用模式匹配来处理你想要的异常
• def onComplete[U](f: Try[T] => U): 传一个接受Try[T]（http://www.yangbajing.me/2013/02/16/Option-Either-Try/）类型的函数。

上面3个回调函数都要求返回一个类型为U的返回值，这也得益于Scala的类型自动推导功能，你能够减小不少的样版代码。

当Future完成后，咱们注册的回调函数将收到值。一个经常使用的注册回调函数是onComplete，它期待传入一个偏函数，并处理Success[T]和Failure[E]两种状况。（编函数将另文介绍）

theFuture.onComplate {
  case Success(result) => println(result)
  case Fialure(t) => println(s"Error: %{t.getMessage}")
}

能够看到，在Scala中写多线程代码是很是轻松惬意的。可是，你觉得使用Future只是简化了new Thead或new Runnable的代码量而以，那就大错特错了。Scala的Future不仅这些功能……

合并多个Future的结果

实际工做中，咱们常常遇到须要向多个来源同时异步请求数据的时候。这时咱们就须要等因此来源数据都返回后将结果集处理后再返回。使用Future.sequence方法，接收一个包含Future的列表来将一系列Future的结果汇总到一个List单一结果里输出。完整代码在：http://git.oschina.net/yangbajing/codes/08pacy2lubgqnkmv1xojd

val f1 = Future {
      TimeUnit.SECONDS.sleep(1)
      "f1"
    }

    val f2 = Future {
      TimeUnit.SECONDS.sleep(2)
      "f2"
    }

    val f3 = Future {
      TimeUnit.SECONDS.sleep(3)
      2342
    }

    val f4 = Future.sequence(Seq(f1, f2, f3))

    val results: List[Any] = Await.result(f4, 4.seconds)

    println(results) // 输出：List(f1, f2, 2342)

代码f1、f2、f3字义了3个异步操做并立刻执行。f4将3个异步操做的结果合并到一个List里返回，同时f4也是一个异步操做。除了采用Future.sequence提供的方便函数，咱们还可使用for comprehension特性来更灵活的合并多个Future的结果。

咱们把f4的操做改为使用for推导式形式：

val f4: Future[(String, String, Int)] =
      for {
        r2 <- f2
        r3 <- f3
        r1 <- f1
      } yield (r1.take(1), r2.drop(1), r3 + 1)

    val (f1Str, f2Str, f3Int) = Await.result(f4, 4.seconds)

    println(s"f1: $f1Str, f2: $f2Str, f3: $f3Int") // 输出：f1: f, f2: 2, f3: 2342

能够看到for推导式也可使用在Future上，r2 <- f2代码的含意是在f2这个Future执行完后将结果赋值给变量r2。与Future.sequence将多个线程的返回值合并到一个List不一样。使用for推导式，在yield语句部分你能够对每一个线程的运算结果作更自由的处理，并返回本身想要的类型（这得益于Scala强大的类型推导功能，不须要你显示的声明变量值类型）。

异常处理

前文代码，当Future代码块内有异常抛出时使用Future.sequence和for comprehension也会抛出异常，你将不能正确的得到结果。这时，可使用Future提供的recover方法处理异常，并把异常恢复成一个正确值返回。

def recover[U >: T](pf: PartialFunction[Throwable, U])

recover经常使用使用方式以下：

Future (6 / 0) recover { case e: ArithmeticException => 0 } // result: 0
Future (6 / 0) recover { case e: NotFoundException   => 0 } // result: exception
Future (6 / 2) recover { case e: ArithmeticException => 0 } // result: 3

接下来看看怎样使用recover来将处理异常并可以使返回值可正确应用于Future.sequence和for comprehension中。以以前的f2举例，修改代码以下：

val f2 = Future {
  throw new RuntimeException("throw exception")
}.recover {
  case e: Exception =>
    "handled exception"
}

采用上面的步骤将异常转换成一个值返回，f2就能够正确的应用到合并代码里了。

Promise

Promise是一个承若，它是一个可修改的对象。一个Promise能够在将来成功的完成一个任务（使用p.success来完成），也可能用来完成一个失败（经过返回一个异常，使用p.failure）。失败了的Promise，能够经过f.recover来处理故障。考虑一个把com.google.common.util.concurrent.FutureCallback<V>封装成Scala的Future的例子，看看Promise是怎样使用的。

val promise = Promise[R]()
    Futures.addCallback(
      resultSetFuture,
      new FutureCallback[ResultSet] {
        override def onFailure(t: Throwable): Unit = promise.failure(t)

        override def onSuccess(rs: ResultSet): Unit = promise.complete(Try(func(rs)))
      },
      ec)
    promise.future

总结

Scala使用Future和Promise对并发编程提供了快捷的支持，同时对多个Future结果的合并和Future的异常恢复也提供了优雅的解决方案。