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一样:

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

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

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

 1 tnow.unsafePerformSync                            //> res0: Int = 7
 2 tdelay.unsafePerformSync                          //> run delay ...
 3                                                   //| res1: Int = 7
 4 tnow.unsafePerformAsync {
 5   case \/-(a) => println(s"the result is: $a")
 6   case -\/(e) => println(e.getMessage)
 7 }                                                 //> the result is: 7
 8 tdelay.unsafePerformAsync {
 9   case \/-(a) => println(s"the result is: $a")
10   case -\/(e) => println(e.getMessage)
11 }                                                 //> run delay ...
12                                                   //| the result is: 7
13 tapply.unsafePerformAsync {
14   case \/-(a) => println(s"the result is: $a")
15   case -\/(e) => println(e.getMessage)
16 }
17 Thread.sleep(1000)                                //> run apply ...
18                                                   //| the result is: 7

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

 1 def delay[A](a: => A): Task[A] = suspend(now(a))
 2 def suspend[A](a: => Task[A]): Task[A] = new Task(Future.suspend(
 3     Try(a.get) match {
 4       case -\/(e) => Future.now(-\/(e))
 5       case \/-(f) => f
 6   }))
 7 //Future.suspend:
 8  def suspend[A](f: => Future[A]): Future[A] = Suspend(() => f)
 9 
10  def apply[A](a: => A)(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Task[A] =
11     new Task(Future(Try(a))(pool))
12 //Future.apply
13 def apply[A](a: => A)(implicit pool: ExecutorService = Strategy.DefaultExecutorService): Future[A] = Async { cb =>
14     pool.submit { new Callable[Unit] { def call = cb(a).run }}
15   }

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

 1 def eval(value: => Int) = Task { Thread.sleep(1000); value }
 2                                                   //> eval: (value: => Int)scalaz.concurrent.Task[Int]
 3 eval( 3 * 7 ).onFinish {
 4   case None => Task { println("finished calculation successfully.") }
 5   case Some(e) => Task { println(s"caught error [${e.getMessage}]") }
 6 }.unsafePerformSyncAttempt match {
 7   case -\/(e) => println(s"calculation error [${e.getMessage}]")
 8   case \/-(a) => println(s"the result is: $a")
 9 }                                                 //> finished calculation successfully.
10                                                   //| the result is: 21
11 // 异常处理
12 eval( 3 * 7 / 0 ).onFinish {
13   case None => Task { println("finished calculation successfully.") }
14   case Some(e) => Task { println(s"caught error [${e.getMessage}]") }
15 }.unsafePerformAsync {
16   case -\/(e) => println(s"calculation error [${e.getMessage}]")
17   case \/-(a) => println(s"the result is: $a")
18 }
19 Thread.sleep(2000)                                //> caught error [/ by zero]
20                                                   //| calculation error [/ by zero]

精准异常处理例子:

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

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

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

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

 1 def seqFib(n: Int): Task[Int] =  n match {
 2   case 0 | 1 => Task now  1
 3   case n => {
 4     for {
 5       x <- seqFib(n-1)
 6       y <- seqFib(n-2)
 7     } yield x + y
 8   }
 9  }                                                //> seqFib: (n: Int)scalaz.concurrent.Task[Int]
10  //并行算法
11  def parFib(n: Int): Task[Int] = n match {
12     case 0 | 1 => Task now 1
13     case n => {
14        val ND = Nondeterminism[Task]
15        for {
16          pair <- ND.both(parFib(n-1), parFib(n-2))
17          (x,y) = pair
18        } yield x + y
19     }
20  }                                                //> parFib: (n: Int)scalaz.concurrent.Task[Int]
21  def msFib(n: Int, fib: Int => Task[Int]) = for {
22    b <- Task now { System.currentTimeMillis() }
23    a <- fib(n)
24    e <- Task now { System.currentTimeMillis() }
25  } yield (a, (e-b))                               //> msFib: (n: Int, fib: Int => scalaz.concurrent.Task[Int])scalaz.concurrent.T
26                                                   //| ask[(Int, Long)]
27  
28  msFib(20, parFib).unsafePerformSync              //> res3: (Int, Long) = (10946,373)
29  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。我们可以这样指定线程池:

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

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

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

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

 1 import scala.concurrent.{Future => sFuture}
 2 import scala.util.{Success,Failure}
 3 import scala.concurrent.ExecutionContext
 4 def futureToTask[A](fut: sFuture[A])(implicit ec: ExecutionContext): Task[A] =
 5   Task.async {
 6     cb =>
 7       fut.onComplete {
 8         case Success(a) => cb(a.right)
 9         case Failure(e) => cb(e.left)
10       }
11   }
12 def taskToFuture[A](ta: Task[A]): sFuture[A] = {
13   val prom = scala.concurrent.Promise[A]
14   ta.unsafePerformAsync {
15     case -\/(e) => prom.failure(e)
16     case \/-(a) => prom.success(a)
17   }
18   prom.future
19 }

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

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏IT技术精选文摘

Java并发包类总览

并发容器 这些容器的关键方法大部分都实现了线程安全的功能,却不使用同步关键字(synchronized)。值得注意的是Queue接口本身定义的几个常用方法的区...

18010
来自专栏禁心尽力

多线程编程:阻塞、并发队列的使用总结

最近,一直在跟设计的任务调度模块周旋,目前终于完成了第一阶段的调试。今天,我想借助博客园平台把最近在设计过程中,使用队列和集合的一些基础知识给大家总结一下,方便...

1855
来自专栏函数式编程语言及工具

SDP(13): Scala.Future - far from completion,绝不能用来做甩手掌柜

  在前面几篇关于数据库引擎的讨论里很多的运算函数都返回了scala.Future类型的结果,因为我以为这样就可以很方便的实现了non-blocking效果。无...

3336
来自专栏栗霖积跬步之旅

java多线程编程核心技术——第七章补漏拾遗

本章主要知识点:   1)线程组的使用   2)如何切换线程状态   3)SimpleDateFormat类与多线程的解决方法   4)如何处理线程异常。 这本...

1649
来自专栏爱撒谎的男孩

并发编程之线程管理

783
来自专栏Web项目聚集地

通俗易懂讲解Java线程安全

一个非科班的技术男,自学半年找到了份不错的程序员工作,运营维护订阅号「一个程序员的成长」希望做一个专注于Java领域的公众号,喜欢本文章可以搜索关注。

772
来自专栏轮子工厂

5个步骤,教你瞬间明白线程和线程安全

记得今年3月份刚来杭州面试的时候,有一家公司的技术总监问了我这样一个问题:你来说说有哪些线程安全的类?我心里一想,这我早都背好了,稀里哗啦说了一大堆。

744
来自专栏程序之美

异步任务执行的设计模式

602
来自专栏逸鹏说道

C# 温故而知新: 线程篇(三)下

结果计时器会一直滚动,因为a对象被锁住,除非完成Thread.Sleep(3000000)后才能进入到a共享区 ? 由于以上的问题,微软还是建议我们使用一个私有...

3566
来自专栏MasiMaro 的技术博文

老版VC++线程池

在一般的设计中,当需要一个线程时,就创建一个,但是当线程过多时可能会影响系统的整体效率,这个性能的下降主要体现在:当线程过多时在线程间来回切换需要花费时间,而频...

973

扫码关注云+社区