Scalaz（45）－ concurrency ：Task－函数式多线程编程核心配件

    我们在上一节讨论了scalaz Future，我们说它是一个不完善的类型，最起码没有完整的异常处理机制，只能用在构建类库之类的内部环境。如果scalaz在Future类定义中增加异常处理工具的话，用户就会经常遇到Future[Throwable/A]这样的类型，那么在进行Monadic编程时就必须使用Monad Transformer来匹配类型，程序就会变得不必要的复杂。scalaz的解决方案就是把Future[Throwable/A]包嵌在Task类里，然后把所有Future都统一升格成Task。Task是个Monad, 这样，我们就可以统一方便地用Task来进行多线程函数式编程了。我们先看看Task的定义：scalaz.concurrent/Task.scala

class Task[+A](val get: Future[Throwable / A]) {

  def flatMap[B](f: A => Task[B]): Task[B] =
    new Task(get flatMap {
      case -/(e) => Future.now(-/(e))
      case /-(a) => Task.Try(f(a)) match {
        case e @ -/(_) => Future.now(e)
        case /-(task) => task.get
      }
    })

  def map[B](f: A => B): Task[B] =
    new Task(get map { _ flatMap {a => Task.Try(f(a))} })
...

Task实现了flatMap，所以是个Monad，我们可以在for-comprehension中使用Task。

Task的构建方式与Future一样：

val tnow = Task.now { println("run now ..."); 3+4 }
                                     //> run now ...
                                     //| tnow  : scalaz.concurrent.Task[Int] = scalaz.concurrent.Task@1a968a59
val tdelay = Task.delay { println("run delay ...");  3+4 }
                                    //> tdelay  : scalaz.concurrent.Task[Int] = scalaz.concurrent.Task@13deb50e
val tapply = Task { println("run apply ..."); 3+4 }
                                   //> tapply  : scalaz.concurrent.Task[Int] = scalaz.concurrent.Task@59494225

同样，now函数是即时运算的。它就是一个lifter，能把一个普通运算直接升格为Task。

针对Task有几种运算方法：

tnow.unsafePerformSync                            //> res0: Int = 7
tdelay.unsafePerformSync                          //> run delay ...
                                                  //| res1: Int = 7
tnow.unsafePerformAsync {
  case /-(a) => println(s"the result is: $a")
  case -/(e) => println(e.getMessage)
}                                                 //> the result is: 7
tdelay.unsafePerformAsync {
  case /-(a) => println(s"the result is: $a")
  case -/(e) => println(e.getMessage)
}                                                 //> run delay ...
                                                  //| the result is: 7
tapply.unsafePerformAsync {
  case /-(a) => println(s"the result is: $a")
  case -/(e) => println(e.getMessage)
}
Thread.sleep(1000)                                //> run apply ...
                                                  //| the result is: 7

从上面的例子我们可以得出：tnow已经完成了运算，因为运算结果没有"run now ..."提示了。tdelay和tapply都是存在trampoline结构里的。但tapply存在更深一层的结构里，所以我们必须拖时间来等待tapply的运算结果。tdelay存放在Future.Suspend结构里，而tapply是存放在Future.Async结构里的，所以tdelay是一种延迟运算，而tapply就是异步运算了：

def delay[A](a: => A): Task[A] = suspend(now(a))
def suspend[A](a: => Task[A]): Task[A] = new Task(Future.suspend(
    Try(a.get) match {
      case -/(e) => Future.now(-/(e))
      case /-(f) => f
  }))
//Future.suspend:
 def suspend[A](f: => Future[A]): Future[A] = Suspend(() => f)

 def apply[A](a: => A)(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Task[A] =
    new Task(Future(Try(a))(pool))
//Future.apply
def apply[A](a: => A)(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Future[A] = Async { cb =>
    pool.submit { new Callable[Unit] { def call = cb(a).run }}
  }

好了，我们再看看Task是怎样处理异常情况的：

def eval(value: => Int) = Task { Thread.sleep(1000); value }
                                                  //> eval: (value: => Int)scalaz.concurrent.Task[Int]
eval( 3 * 7 ).onFinish {
  case None => Task { println("finished calculation successfully.") }
  case Some(e) => Task { println(s"caught error [${e.getMessage}]") }
}.unsafePerformSyncAttempt match {
  case -/(e) => println(s"calculation error [${e.getMessage}]")
  case /-(a) => println(s"the result is: $a")
}                                                 //> finished calculation successfully.
                                                  //| the result is: 21
// 异常处理
eval( 3 * 7 / 0 ).onFinish {
  case None => Task { println("finished calculation successfully.") }
  case Some(e) => Task { println(s"caught error [${e.getMessage}]") }
}.unsafePerformAsync {
  case -/(e) => println(s"calculation error [${e.getMessage}]")
  case /-(a) => println(s"the result is: $a")
}
Thread.sleep(2000)                                //> caught error [/ by zero]
                                                  //| calculation error [/ by zero]

精准异常处理例子：

import java.util.concurrent._
val timedTask = Task {Thread.sleep(2000); 3+4}   
                     //> timedTask  : scalaz.concurrent.Task[Int] = scalaz.concurrent.Task@3d921e20
timedTask.timed(1 second).handleWith {
  case e: TimeoutException => Task { println(s"calculation exceeding time limit: ${e.getMessage}") }
}.unsafePerformSync           //> calculation exceeding time limit: Timed out after 1000 milliseconds
                              //| res2: AnyVal{def getClass(): Class[_ >: Int with Unit <: AnyVal]} = ()

再看一些多线程编程例子：

val tasks = (1 |-> 5).map(n => Task{ Thread.sleep(100); n })
                                //> tasks  : List[scalaz.concurrent.Task[Int]] = List(scalaz.concurrent.Task@61
                                //| 8b19ad, scalaz.concurrent.Task@2d3379b4, scalaz.concurrent.Task@30c15d8b, s
                                //| calaz.concurrent.Task@5e0e82ae, scalaz.concurrent.Task@6771beb3)
//并行运算list of tasks
Task.gatherUnordered(tasks).unsafePerformSync     //> res3: List[Int] = List(1, 2, 3, 4, 5)
val sb = new StringBuffer                         //> sb  : StringBuffer = 
val t1 = Task.fork { Thread.sleep(100); sb.append("a"); Task.now("a")}
                                 //> t1  : scalaz.concurrent.Task[String] = scalaz.concurrent.Task@6200f9cb
val t2 = Task.fork { Thread.sleep(800); sb.append("b"); Task.now("b")}
                                 //> t2  : scalaz.concurrent.Task[String] = scalaz.concurrent.Task@2002fc1d
val t3 = Task.fork { Thread.sleep(200); sb.append("c"); Task.now("c")}
                                 //> t3  : scalaz.concurrent.Task[String] = scalaz.concurrent.Task@69453e37
val t4 = Task.fork { Thread.sleep(100); sb.append("d"); Task.now("d")}
                                 //> t4  : scalaz.concurrent.Task[String] = scalaz.concurrent.Task@6f4a47c7
val t5 = Task.fork { Thread.sleep(400); sb.append("e"); Task.now("e")}
                                 //> t5  : scalaz.concurrent.Task[String] = scalaz.concurrent.Task@ae13544
val t6 = Task.fork { Thread.sleep(100); sb.append("f"); Task.now("f")}
                                 //> t6  : scalaz.concurrent.Task[String] = scalaz.concurrent.Task@3d34d211
val r = Nondeterminism[Task].nmap6(t1,t2,t3,t4,t5,t6)(List(_,_,_,_,_,_))
                                 //> r  : scalaz.concurrent.Task[List[String]] = scalaz.concurrent.Task@394df057 
r.unsafePerformSync              //> res4: List[String] = List(a, b, c, d, e, f)

看个耗时算法的并行运算吧：

def seqFib(n: Int): Task[Int] =  n match {
  case 0 | 1 => Task now  1
  case n => {
    for {
      x <- seqFib(n-1)
      y <- seqFib(n-2)
    } yield x + y
  }
 }                                                //> seqFib: (n: Int)scalaz.concurrent.Task[Int]
 //并行算法
 def parFib(n: Int): Task[Int] = n match {
    case 0 | 1 => Task now 1
    case n => {
       val ND = Nondeterminism[Task]
       for {
         pair <- ND.both(parFib(n-1), parFib(n-2))
         (x,y) = pair
       } yield x + y
    }
 }                                                //> parFib: (n: Int)scalaz.concurrent.Task[Int]
 def msFib(n: Int, fib: Int => Task[Int]) = for {
   b <- Task now { System.currentTimeMillis() }
   a <- fib(n)
   e <- Task now { System.currentTimeMillis() }
 } yield (a, (e-b))                               //> msFib: (n: Int, fib: Int => scalaz.concurrent.Task[Int])scalaz.concurrent.T
                                                  //| ask[(Int, Long)]
 
 msFib(20, parFib).unsafePerformSync              //> res3: (Int, Long) = (10946,373)
 msFib(20, seqFib).unsafePerformSync              //> res4: (Int, Long) = (10946,17)

哎呀！奇怪了，为什么并行算法要慢很多呢？这个问题暂且放一放，以后再研究。当然，如果有读者能给出个解释就太感激了。
Task的线程池是如何分配的呢？看看Task.apply和Task.fork：

 /** Create a `Task` that will evaluate `a` using the given `ExecutorService`. */
  def apply[A](a: => A)(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Task[A] =
    new Task(Future(Try(a))(pool))
def fork[A](a: => Task[A])(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Task[A] =
    apply(a).join
//Future.apply
/** Create a `Future` that will evaluate `a` using the given `ExecutorService`. */
  def apply[A](a: => A)(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Future[A] = Async { cb =>
    pool.submit { new Callable[Unit] { def call = cb(a).run }}
 

这两个函数都包括了一个隐式参数implicit pool: ExecutorService。默认值是Strategy.DefultExecutorService。我们可以这样指定线程池：

 Task {longProcess}(myExecutorService)
 Task.fork { Task {longProcess} }(myExecutorService)

下面是一个动态指定线程池的例子：

import java.util.concurrent.{ExecutorService,Executors}
type Delegated[A] = Kleisli[Task,ExecutorService,A]
def delegate: Delegated[ExecutorService] = Kleisli(e => Task.now(e))
                            //> delegate: => demo.ws.task.Delegated[java.util.concurrent.ExecutorService]
implicit def delegateTaskToPool[A](ta: Task[A]): Delegated[A] = Kleisli(x => ta)
                            //> delegateTaskToPool: [A](ta: scalaz.concurrent.Task[A])demo.ws.task.Delegated[A]
val tPrg = for {
  p <- delegate
  b <- Task("x")(p)
  c <- Task("y")(p)
} yield c                   //> tPrg  : scalaz.Kleisli[scalaz.concurrent.Task,java.util.concurrent.Executor
                            //| Service,String] = Kleisli(<function1>)
tPrg.run(Executors.newFixedThreadPool(3)).unsafePerformSync
                            //> res3: String = y

当然，Task和scala Future之间是可以相互转换的：

import scala.concurrent.{Future => sFuture}
import scala.util.{Success,Failure}
import scala.concurrent.ExecutionContext
def futureToTask[A](fut: sFuture[A])(implicit ec: ExecutionContext): Task[A] =
  Task.async {
    cb =>
      fut.onComplete {
        case Success(a) => cb(a.right)
        case Failure(e) => cb(e.left)
      }
  }
def taskToFuture[A](ta: Task[A]): sFuture[A] = {
  val prom = scala.concurrent.Promise[A]
  ta.unsafePerformAsync {
    case -/(e) => prom.failure(e)
    case /-(a) => prom.success(a)
  }
  prom.future
}

与Future不同的是：Task增加了异常处理机制。