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的異常恢復也提供了優雅的解決方案。