SDP（13）： Scala.Future - far from completion，毫不能用來作甩手掌櫃

  在前面幾篇關於數據庫引擎的討論裏不少的運算函數都返回了scala.Future類型的結果，由於我覺得這樣就能夠很方便的實現了non-blocking效果。不管任何複雜的數據處理操做，只要把它們包在一個Future{...}裏扔給系統運算就算完事不理了，立刻能夠把關注放到編程的其它部分了。在3月17日的深圳scala用戶meetup裏我作了個關於scala函數式編程的分享，裏面我提到如今使用最多的函數組件就是scala.Future了。我想這應該在scala用戶羣裏是個比較廣泛的現象：你們都認爲這是實現non-blocking最直接的一種方式。不過當我在meetup後回想到scala.Future時忽然意識到它是一種即時運算值strict-value，看看下面這個例子：

 import scala.concurrent.duration._ val fs = Future {println("run now..."); System.currentTimeMillis() } //> run now... //| fs : scala.concurrent.Future[Long] = List()
  Await.result(fs, 1.second)             //> res0: Long = 1465907784714
  Thread.sleep(1000) Await.result(fs, 1.second)             //> res1: Long = 1465907784714

能夠看到fs是在Future構建時即時運算的，並且只會運算一次。若是scala Future中包括了能產生反作用的代碼，在構建時就會當即產生反作用。因此咱們是沒法使用scala Future來編寫純函數的，以下：

val progA：Future[A] = for { b <- readFromB _ <- writeToLocationA(a) r <- getResult } yield r /* location A content updated */ ... /* later */ val progB: Future[B] = for { a <- readFromA _ <- updateLocationA c <- getResult } ... val program: Future[Unit] = for { _ <- progA _ <- progB } yield()

在上面這個例子裏最終的目的是運算program：由progA,progB兩個子程序組成。這兩個子程序在構建的時候已經開始了運算，隨時都會更新localionA產生反作用。想象一下若是progA，progB是埋藏在其它一大堆源代碼裏的話program的運算結果確定是沒法預測的。換言之用Future來進行函數式組合就是在給本身挖坑嘛，最起碼要記住這些Future的構建順序，而這個要求在大型的協做開發軟件工程裏基本上是不可能的事。除了沒法安全進行函數組合外scala.Future還缺乏運算和線程控制的功能，好比：

沒法控制何時開始運算

沒法控制在在哪一個線程運算

沒法終止開始運算的程序

缺乏有效的異常處理機制如fallback，retry等

scalaz和monix函數組件庫裏都提供了Task來輔助Future實現函數組合。scalaz.Task是基於scalaz.Future的：

sealed abstract class Future[+A] { ... object Future { case class Now[+A](a: A) extends Future[A] case class Async[+A](onFinish: (A => Trampoline[Unit]) => Unit) extends Future[A] case class Suspend[+A](thunk: () => Future[A]) extends Future[A] case class BindSuspend[A,B](thunk: () => Future[A], f: A => Future[B]) extends Future[B] case class BindAsync[A,B](onFinish: (A => Trampoline[Unit]) => Unit, f: A => Future[B]) extends Future[B] ...

scalaz.Future[A]明顯就是個Free Monad。它的結構化表達方式分別有Now,Async,Suspend,BindSuspend,BindAsync。咱們能夠用這些結構實現flatMap函數，因此Future就是Free Monad：

def flatMap[B](f: A => Future[B]): Future[B] = this match { case Now(a) => Suspend(() => f(a)) case Suspend(thunk) => BindSuspend(thunk, f) case Async(listen) => BindAsync(listen, f) case BindSuspend(thunk, g) => Suspend(() => BindSuspend(thunk, g andThen (_ flatMap f))) case BindAsync(listen, g) => Suspend(() => BindAsync(listen, g andThen (_ flatMap f))) }

由於free structure類型支持算式／算法關注分離，咱們能夠用scalaz.Future來描述程序功能而不涉及正真運算。這樣，在上面那個例子裏若是progA,progB是Task類型的，那麼program的構建就是安全的，由於咱們最後是用Task.run來真正進行運算產生反作用的。scalaz.Task又在scalaz.Future功能基礎上再增長了異常處理等功能。

monix.Task採起了延遲運算的方式來實現算式/算法分離，下面是這個類型的基礎構建結構：

  /** [[Task]] state describing an immediate synchronous value. */
  private[eval] final case class Now[A](value: A) extends Task[A] {...} /** [[Task]] state describing an immediate synchronous value. */
  private[eval] final case class Eval[A](thunk: () => A) extends Task[A] /** Internal state, the result of [[Task.defer]] */
  private[eval] final case class Suspend[+A](thunk: () => Task[A]) extends Task[A] /** Internal [[Task]] state that is the result of applying `flatMap`. */
  private[eval] final case class FlatMap[A, B](source: Task[A], f: A => Task[B]) extends Task[B] /** Internal [[Coeval]] state that is the result of applying `map`. */
  private[eval] final case class Map[S, +A](source: Task[S], f: S => A, index: Int) extends Task[A] with (S => Task[A]) { def apply(value: S): Task[A] =
      new Now(f(value)) override def toString: String = super[Task].toString } /** Constructs a lazy [[Task]] instance whose result will * be computed asynchronously. * * Unsafe to build directly, only use if you know what you're doing. * For building `Async` instances safely, see [[create]]. */
  private[eval] final case class Async[+A](register: (Context, Callback[A]) => Unit) extends Task[A] 

下面的例子裏示範了若是用這些結構來構件monix.Task： 

object Task extends TaskInstancesLevel1 {
  /** Returns a new task that, when executed, will emit the result of
    * the given function, executed asynchronously.
    *
    * This operation is the equivalent of:
    * {{{
    *   Task.eval(f).executeAsync
    * }}}
    *
    * @param f is the callback to execute asynchronously
    */
  def apply[A](f: => A): Task[A] =
    eval(f).executeAsync

  /** Returns a `Task` that on execution is always successful, emitting
    * the given strict value.
    */
  def now[A](a: A): Task[A] =
    Task.Now(a)

  /** Lifts a value into the task context. Alias for [[now]]. */
  def pure[A](a: A): Task[A] = now(a)

  /** Returns a task that on execution is always finishing in error
    * emitting the specified exception.
    */
  def raiseError[A](ex: Throwable): Task[A] =
    Error(ex)

  /** Promote a non-strict value representing a Task to a Task of the
    * same type.
    */
  def defer[A](fa: => Task[A]): Task[A] =
    Suspend(fa _)
...}
    source match {
      case Task.Now(v) => F.pure(v)
      case Task.Error(e) => F.raiseError(e)
      case Task.Eval(thunk) => F.delay(thunk())
      case Task.Suspend(thunk) => F.suspend(to(thunk()))
      case other => suspend(other)(F)
    }

這個Suspend結構就是延遲運算的核心。monix.Task是一套新出現的解決方案，借鑑了許多scalaz.Task的概念和方法同時又加入了不少優化、附加的功能，而且github更新也很近期。使用monix.Task應該是一個正確的選擇。

首先咱們必須解決scala.Future與monix.Task之間的轉換：

 import monix.eval.Task import monix.execution.Scheduler.Implicits.global final class FutureToTask[A](x: => Future[A]) { def asTask: Task[A] = Task.deferFuture[A(x) } final class TaskToFuture[A](x: => Task[A]) { def asFuture: Future[A] = x.runAsync }

下面是一個完整的Task用例：

import scala.concurrent._ import scala.util._ import scala.concurrent.duration._ import monix.eval.Task import monix.execution._ object MonixTask extends App { import monix.execution.Scheduler.Implicits.global

  // Executing a sum, which (due to the semantics of apply) // will happen on another thread. Nothing happens on building // this instance though, this expression is pure, being // just a spec! Task by default has lazy behavior ;-)
  val task = Task { 1 + 1 } // Tasks get evaluated only on runAsync! // Callback style:
  val cancelable = task.runOnComplete { case Success(value) => println(value) case Failure(ex) => System.out.println(s"ERROR: ${ex.getMessage}") } //=> 2 // If we change our mind...
 cancelable.cancel() // Or you can convert it into a Future
  val future: CancelableFuture[Int] = task.runAsync // Printing the result asynchronously
  future.foreach(println) //=> 2
 val task = Task.now { println("Effect"); "Hello!" } //=> Effect // task: monix.eval.Task[String] = Delay(Now(Hello!))
}

下面咱們就看看各類Task的構建方法：

  /* ------ taskNow ----*/ val taskNow = Task.now { println("Effect"); "Hello!" } //=> Effect // taskNow: monix.eval.Task[String] = Delay(Now(Hello!))

  /* --------taskDelay possible another on thread ------*/ val taskDelay = Task { println("Effect"); "Hello!" } // taskDelay: monix.eval.Task[String] = Delay(Always(<function0>))
 taskDelay.runAsync.foreach(println) //=> Effect //=> Hello! // The evaluation (and thus all contained side effects) // gets triggered on each runAsync:
  taskDelay.runAsync.foreach(println) //=> Effect //=> Hello!

  /* --------taskOnce ------- */ val taskOnce = Task.evalOnce { println("Effect"); "Hello!" } // taskOnce: monix.eval.Task[String] = EvalOnce(<function0>)
 taskOnce.runAsync.foreach(println) //=> Effect //=> Hello! // Result was memoized on the first run!
  taskOnce.runAsync.foreach(println) //=> Hello!

  /* --------taskFork ------- */
  // this guarantees that our task will get executed asynchronously:
  val task = Task(Task.eval("Hello!")).executeAsync //val task = Task.fork(Task.eval("Hello!")) // The default scheduler
  import monix.execution.Scheduler.Implicits.global

  // Creating a special scheduler meant for I/O
 import monix.execution.Scheduler lazy val io = Scheduler.io(name="my-io") //Then we can manage what executes on which: // Override the default Scheduler by fork:
  val source = Task(println(s"Running on thread: ${Thread.currentThread.getName}")) val forked = source.executeOn(io,true) // val forked = Task.fork(source, io)
 source.runAsync //=> Running on thread: ForkJoinPool-1-worker-1
 forked.runAsync //=> Running on thread: my-io-4

  /* --------taskError ------- */ import scala.concurrent.TimeoutException val taskError = Task.raiseError[Int](new TimeoutException) // error: monix.eval.Task[Int] = // Delay(Error(java.util.concurrent.TimeoutException))
 taskError.runOnComplete(result => println(result)) //=> Failure(java.util.concurrent.TimeoutException)

下面是一些控制函數：

  final def doOnFinish(f: Option[Throwable] => Task[Unit]): Task[A] = final def doOnCancel(callback: Task[Unit]): Task[A] = final def onCancelRaiseError(e: Throwable): Task[A] = final def onErrorRecoverWith[B >: A](pf: PartialFunction[Throwable, Task[B]]): Task[B] = final def onErrorHandleWith[B >: A](f: Throwable => Task[B]): Task[B] = final def onErrorFallbackTo[B >: A](that: Task[B]): Task[B] = final def restartUntil(p: (A) => Boolean): Task[A] = final def onErrorRestart(maxRetries: Long): Task[A] = final def onErrorRestartIf(p: Throwable => Boolean): Task[A] = final def onErrorRestartLoop[S, B >: A](initial: S)(f: (Throwable, S, S => Task[B]) => Task[B]): Task[B] = final def onErrorHandle[U >: A](f: Throwable => U): Task[U] = final def onErrorRecover[U >: A](pf: PartialFunction[Throwable, U]): Task[U] =

Task是經過asyncRun和runSync來進行異步、同步實際運算的： 

  def runAsync(implicit s: Scheduler): CancelableFuture[A] = def runAsync(cb: Callback[A])(implicit s: Scheduler): Cancelable = def runAsyncOpt(implicit s: Scheduler, opts: Options): CancelableFuture[A] = def runAsyncOpt(cb: Callback[A])(implicit s: Scheduler, opts: Options): Cancelable = final def runSyncMaybe(implicit s: Scheduler): Either[CancelableFuture[A], A] = final def runSyncMaybeOpt(implicit s: Scheduler, opts: Options): Either[CancelableFuture[A], A] = final def runSyncUnsafe(timeout: Duration) (implicit s: Scheduler, permit: CanBlock): A = final def runSyncUnsafeOpt(timeout: Duration) (implicit s: Scheduler, opts: Options, permit: CanBlock): A = final def runOnComplete(f: Try[A] => Unit)(implicit s: Scheduler): Cancelable =

下面示範了兩個一般的Task運算方法：

  val task1 = Task {println("sum:"); 1+2}.delayExecution(1 second) println(task1.runSyncUnsafe(2 seconds)) task1.runOnComplete { case Success(r) => println(s"result: $r") case Failure(e) => println(e.getMessage) }

下面是本次示範的源代碼：

import scala.util._ import scala.concurrent.duration._ import monix.eval.Task import monix.execution._ object MonixTask extends App { import monix.execution.Scheduler.Implicits.global



  // Executing a sum, which (due to the semantics of apply) // will happen on another thread. Nothing happens on building // this instance though, this expression is pure, being // just a spec! Task by default has lazy behavior ;-)
  val task = Task { 1 + 1 } // Tasks get evaluated only on runAsync! // Callback style:
  val cancelable = task.runOnComplete { case Success(value) => println(value) case Failure(ex) => System.out.println(s"ERROR: ${ex.getMessage}") } //=> 2 // If we change our mind...
 cancelable.cancel() // Or you can convert it into a Future
  val future: CancelableFuture[Int] = task.runAsync // Printing the result asynchronously
  future.foreach(println) //=> 2

  /* ------ taskNow ----*/ val taskNow = Task.now { println("Effect"); "Hello!" } //=> Effect // taskNow: monix.eval.Task[String] = Delay(Now(Hello!))

  /* --------taskDelay possible another on thread ------*/ val taskDelay = Task { println("Effect"); "Hello!" } // taskDelay: monix.eval.Task[String] = Delay(Always(<function0>))
 taskDelay.runAsync.foreach(println) //=> Effect //=> Hello! // The evaluation (and thus all contained side effects) // gets triggered on each runAsync:
  taskDelay.runAsync.foreach(println) //=> Effect //=> Hello!

  /* --------taskOnce ------- */ val taskOnce = Task.evalOnce { println("Effect"); "Hello!" } // taskOnce: monix.eval.Task[String] = EvalOnce(<function0>)
 taskOnce.runAsync.foreach(println) //=> Effect //=> Hello! // Result was memoized on the first run!
  taskOnce.runAsync.foreach(println) //=> Hello!

  /* --------taskFork ------- */
  // this guarantees that our task will get executed asynchronously:
  val task = Task(Task.eval("Hello!")).executeAsync //val task = Task.fork(Task.eval("Hello!")) // The default scheduler
  import monix.execution.Scheduler.Implicits.global

  // Creating a special scheduler meant for I/O
 import monix.execution.Scheduler lazy val io = Scheduler.io(name="my-io") //Then we can manage what executes on which: // Override the default Scheduler by fork:
  val source = Task(println(s"Running on thread: ${Thread.currentThread.getName}")) val forked = source.executeOn(io,true) // val forked = Task.fork(source, io)
 source.runAsync //=> Running on thread: ForkJoinPool-1-worker-1
 forked.runAsync //=> Running on thread: my-io-4

  /* --------taskError ------- */ import scala.concurrent.TimeoutException val taskError = Task.raiseError[Int](new TimeoutException) // error: monix.eval.Task[Int] = // Delay(Error(java.util.concurrent.TimeoutException))
 taskError.runOnComplete(result => println(result)) //=> Failure(java.util.concurrent.TimeoutException)
 val task1 = Task {println("sum:"); 1+2}.delayExecution(1 second) println(task1.runSyncUnsafe(2 seconds)) task1.runOnComplete { case Success(r) => println(s"result: $r") case Failure(e) => println(e.getMessage) } }