Scalaz（41）－ Free ：IO Monad－Free特定版本的FP语法

用户1150956

发布于 2018-01-05 10:34:44

1.5K0

发布于 2018-01-05 10:34:44

我们不断地重申FP强调代码无副作用，这样才能实现编程纯代码。像通过键盘显示器进行交流、读写文件、数据库等这些IO操作都会产生副作用。那么我们是不是为了实现纯代码而放弃IO操作呢？没有IO的程序就是一段烧CPU的代码，没有任何意义，所以任何类型的程序都必须具备IO功能，而在FP模式中对IO操作有特别的控制方式：具体实现是通过把代码中产生副作用的部分抽离出来延后运算（在所有纯代码运算之后）。scalaz的IO Monad就是处理副作用代码延后运算的一种数据结构。我先举个简单的例子来示范如何通过一种数据结构来实现对副作用代码的延迟运算：人机交互是一种典型的IO，有键盘输入，又有显示屏输出。println,readLine都会产生副作用，我们必须用一种数据类型来实现副作用代码抽离及延后运算，这种类型就是IO。我们先看看这个例子：我们希望实现人机交互如下：

1 def ask(prompt: String): String = { 
2    println(prompt)
3    readLine
4  }
5 def tell(msg: String): Unit = println(msg)
6 for {
7   name <- ask("what's your name?")
8   _ <- tell(s"I'm $name")
9 } yield()

ask和tell分别返回String和Unit，它们都是副作用即时产生的结果。ask和tell都是非纯函数。我们可以设计一个类型来实现副作用代码抽离：

 1 trait MyIO[+A] {self =>
 2   def run: A
 3   def map[B](f: A => B): MyIO[B] =
 4     new MyIO[B] {
 5        def run = f(self.run)
 6     }
 7   def flatMap[B](f: A => MyIO[B]): MyIO[B] =
 8     new MyIO[B] {
 9        def run = f(self.run).run
10     }
11 }
12 object MyIO {
13   def apply[A](a: A) = new MyIO[A] { def run = a }
14   implicit val ioMonad = new Monad[MyIO] {
15      def point[A](a: => A) = new MyIO[A] { def run = a }
16      def bind[A,B](ma: MyIO[A])(f: A => MyIO[B]): MyIO[B] =
17        ma flatMap f
18   }
19 }

现在我们可以把ask和tell函数的返回类型改成MyIO：

 1 import MyIO._
 2 def ask(prompt: String): MyIO[String] = 
 3   MyIO {
 4        println(prompt)
 5        readLine
 6   }
 7 def tell(msg: String): MyIO[Unit] =
 8   MyIO {
 9     println(msg)
10   }

MyIO是个Monad，我们可以在for-comprehension里用行令方式编程了：

1 val org: MyIO[Unit] = for {
2   first <- ask("What's your first name?")
3   last <- ask("What's your last name?")
4   _ <- tell(s"Hello $first $last!")
5 } yield()

注意，在这个阶段我们只完成了对一个程序功能的描述，实际运算是在MyIO.run那里：

 1 object MyIOApp extends App {
 2   import MyIOFunctions._
 3   pr.run
 4 }
 5 //运算结果：
 6 What's your first name?
 7 Tiger
 8 What's your last name?
 9 Chan
10 Hello Tiger Chan!

run是MyIO类型的interpreter。现在我们已经实现了程序描述（算式）和运算（算法）的关注分离。而且我们可以随便使用ask和tell而不进行运算，延迟至调用run对MyIO类型进行运算。以上只是一种简单的示范IO类型，我先把完整的源代码提供如下：

 1 package demo.app
 2 import scalaz._
 3 import Scalaz._
 4 
 5 trait MyIO[+A] {self =>
 6   def run: A
 7   def map[B](f: A => B): MyIO[B] =
 8     new MyIO[B] {
 9        def run = f(self.run)
10     }
11   def flatMap[B](f: A => MyIO[B]): MyIO[B] =
12     new MyIO[B] {
13        def run = f(self.run).run
14     }
15 }
16 object MyIO {
17   def apply[A](a: A) = new MyIO[A] { def run = a }
18   implicit val ioMonad = new Monad[MyIO] {
19      def point[A](a: => A) = new MyIO[A] { def run = a }
20      def bind[A,B](ma: MyIO[A])(f: A => MyIO[B]): MyIO[B] =
21        ma flatMap f
22   }
23 }
24 object MyIOFunctions {
25 import MyIO._
26 def ask(prompt: String): MyIO[String] = 
27   MyIO {
28        println(prompt)
29        readLine
30   }
31 def tell(msg: String): MyIO[Unit] =
32   MyIO {
33     println(msg)
34   }
35 val prg: MyIO[Unit] = for {
36   first <- ask("What's your first name?")
37   last <- ask("What's your last name?")
38   _ <- tell(s"Hello $first $last!")
39 } yield()
40 
41 
42 }
43 object MyIOApp extends App {
44   import MyIOFunctions._
45   prg.run
46 }

scalaz的IO Monad当然复杂的多。我们看看scalaz的IO Monad是怎样的：effect/IO.scala

sealed abstract class IO[A] {
  private[effect] def apply(rw: Tower[IvoryTower]): Trampoline[(Tower[IvoryTower], A)]
...
/** Continues this action with the given function. */
  def map[B](f: A => B): IO[B] = io(rw =>
    apply(rw) map {
      case (nw, a) => (nw, f(a))
    })

  /** Continues this action with the given action. */
  def flatMap[B](f: A => IO[B]): IO[B] = io(rw =>
    apply(rw) flatMap {
      case (nw, a) => f(a)(nw)
    })
...
 /** Construct an IO action from a world-transition function. */
  def io[A](f: Tower[IvoryTower] => Trampoline[(Tower[IvoryTower], A)]): IO[A] =
    new IO[A] {
      private[effect] def apply(rw: Tower[IvoryTower]) = Free(() => f(rw))
    }

可以看得出io[A](...)就是IO的构建器（constructor）。IO[A]类型是Free[Function0,A] = Free(() => f(rw))，最终就是一个Trampoline，这个可以从下面的apply看得出来：

object IO extends IOInstances {
  def apply[A](a: => A): IO[A] =
    io(rw => return_(rw -> a))
...

上面的Tower[IvoryTower]是状态切换函数的输入类型，不参与实际运算（不在任何运算中调用如： rw -> a），起一种状态标签作用（state tag）。主要提供给编译器（compiler）做类型安全用。实际上这个状态切换函数就是一个延迟运算值 => A。io把这个值挂到Free的suspend状态：

/** Suspend the given computation in a single step. */
  def return_[S[_], A](value: => A)(implicit S: Applicative[S]): Free[S, A] =
    liftF[S, A](S.point(value))

再看看IO的运算方式：

sealed abstract class IO[A] {
  private[effect] def apply(rw: Tower[IvoryTower]): Trampoline[(Tower[IvoryTower], A)]

  import IO._

  /**
   * Runs I/O and performs side-effects. An unsafe operation.
   * Do not call until the end of the universe.
   */
  def unsafePerformIO(): A = apply(ivoryTower).run._2

先用apply建Trampoline，再运行Free.run（Trampoline[A]=Free[Function0,A]）。注意，我们并没有采用这个Tower[IvoryTower]。再者，函数unsafePerformIO是通过private函数apply先构建了Trampoline后再进行运算的。换言之IO Monad的用户是无法自定义算法（interpreter）的。我们前面曾经把Free描述成可以自定义F[A]编程语言的数据结构，那么IO[A]就是一种固定的FP编程语言，它只有unsafePerformIO一种算法（interpreter）。

IO Monad可以使我们更方便地在IO这个壳子里进行我们熟悉的行令编程（imperative programming）,因为我们只需要把行令程序直接放进IO里就行了。看看下面这些例子：

1 val hello = print("hello ").point[IO]             //> hello  : scalaz.effect.IO[Unit] = scalaz.effect.IO$$anon$6@145eaa29
2 val world = IO (print("world,"))                  //> world  : scalaz.effect.IO[Unit] = scalaz.effect.IO$$anon$6@57c758ac
3 val howareyou = io {rw => return_(rw -> println("how are you!"))}
4                                                   //> howareyou  : scalaz.effect.IO[Unit] = scalaz.effect.IO$$anon$6@a9cd3b1
5 val greeting = hello |+| world |+| howareyou      //> greeting  : scalaz.effect.IO[Unit] = scalaz.effect.IO$$anon$6@481a996b
6 greeting.unsafePerformIO                          //> hello world,how are you!

这个例子示范了用三种方式把副作用语句print升格成IO。不要被IO[A]的IO字面误导了，IO[A]的这个A不一定是副作用命令，任何行令编程使用的语句都可以放人IO[_]，包括变量申明、赋值、文件读写等。所以我们说IO Monad就是在FP模式中进行行令编程的通用方式。可以想象我们可能会在IO这个壳子内进行我们熟悉的程序编写。那么IO Monad到底能不能符合在FP环境内的行令编程要求呢？我们可以用几个例子来对流程控制（flow control），跟踪记录（logging）即异常处理（exception handling）等方面进行测试示范：

 1 import scalaz._
 2 import Scalaz._
 3 import effect._
 4 import IO._
 5 import Free._
 6 import scala.language.higherKinds
 7 import scala.language.implicitConversions
 8 
 9 object IOPrg {
10   def div(dvdn: Int, dvsor: Int): IO[Int] = 
11     IO(dvdn / dvsor)
12   val ioprg: IO[Int] = for {
13     _ <- putLn("enter dividend:")
14     dvdn <- readLn
15     _ <- putLn("enter divisor:")
16     dvsor <- readLn
17     quot <- div(dvdn.toInt, dvsor.toInt)
18     _ <- putLn(s"the result:$quot")
19   } yield quot
20 }
21 
22 object IOMonadDemo extends App {
23   import IOPrg._
24   ioprg.unsafePerformIO()
25 }

"enter dividend:"
10
"enter divisor:"
5
"the result:2"

ioprg是一段包含了带副作用语句的程序。所有副作用延迟到unsafePerformIO()才正真产生。现在我们先示范流程控制，试着提早跳出这个for-loop。这不就是option在for-comprehension的作用吗。我们需要在IO[A]这种Monad之上增加Option的作用，可以用Monad Transformer来实现：

 1   implicit def ioToOptionT[A](io: IO[A]): OptionT[IO,A] = io.liftM[OptionT]
 2   val optionIOprg: OptionT[IO,Int] = for {
 3     _ <- putLn("enter dividend:")
 4     dvdn <- readLn
 5     _ <- putLn("enter divisor:")
 6     dvsor <- readLn
 7     quot <- div(dvdn.toInt, dvsor.toInt)
 8     _ <- putLn(s"the result:$quot")
 9   } yield quit
10 ...
11 object IOMonadDemo extends App {
12   import IOPrg._
13 //  ioprg.unsafePerformIO()
14   optionIOprg.run.unsafePerformIO()
15 }
16 ...
17 "enter dividend:"
18 10
19 "enter divisor:"
20 5
21 "the result:2"

我们把结果类型变成Option[IO,Int]后并没有改变程序的功能。现在试试中途终止：

 1   val optionIOprg: OptionT[IO,Int] = for {
 2     _ <- putLn("enter dividend:").liftM[OptionT]
 3     dvdn <- readLn.liftM[OptionT]
 4     _ <- putLn("enter divisor:").liftM[OptionT]
 5     dvsor <- readLn.liftM[OptionT]
 6     a <- if (dvsor.toInt == 0 ) OptionT(IO(None: Option[String])) else IO(0).liftM[OptionT]
 7     quot <- div(dvdn.toInt, dvsor.toInt).liftM[OptionT]
 8     _ <- putLn(s"the result:$quot").liftM[OptionT]
 9   } yield quit
10 ...
11 "enter dividend:"
12 10
13 "enter divisor:"
14 5
15 "the result:2"
16 
17 Process finished with exit code 0
18 ...
19 "enter dividend:"
20 10
21 "enter divisor:"
22 0
23 
24 Process finished with exit code 0

不错，的确在录入0时可以中途退出。不过现在for-comprehension里的每句都需要用liftM[OptionT]来升格。无法像前面统统用implicit def ioToOptionT来自动升格，这是因为把None升格时会产生类型推导问题，这个就放在以后再研究了。

同样如果我们希望把用户的输入记录下来，我们可以用Writer的功能来实现。下一个例子就是Writer-IO Monad Transformer示范了：

 1   type WriterTIO[F[_],A] = WriterT[F,List[String],A]
 2   val writerIOprg: WriterT[IO,List[String],Int] = for {
 3     _ <- putLn("enter dividend:").liftM[WriterTIO]
 4     dvdn <- readLn.liftM[WriterTIO]
 5     _ <- WriterT.writerT((List(s"received dividend $dvdn"),dvdn).point[IO])
 6     _ <- putLn("enter divisor:").liftM[WriterTIO]
 7     dvsor <- readLn.liftM[WriterTIO]
 8     _ <- WriterT.writerT(IO(List(s"received divisor $dvsor, ready to divide ..."),dvdn))
 9     quot <- div(dvdn.toInt, dvsor.toInt).liftM[WriterTIO]
10     _ <- putLn(s"the result:$quot").liftM[WriterTIO]
11   } yield quit
12 ...
13 object IOMonadDemo extends App {
14   import IOMonadPrg._
15   //  ioprg.unsafePerformIO()
16   //optionIOprg.run.unsafePerformIO()
17   println(writerIOprg.run.unsafePerformIO())
18 }
19 ...
20 "enter dividend:"
21 10
22 "enter divisor:"
23 5
24 "the result:2"
25 (List(received dividend 10, received divisor 5, ready to divide ...),2)
26 
27 Process finished with exit code 0

用WriterT可以达到logging目的。当然，我们可以同时拥有Option和Writer的作用，这时的Monad Transformer就是三层的了，我们在前面的这篇讨论也做过示范。

最后看个异常处理示范：

 1   type WriterTIO[F[_],A] = WriterT[F,List[String],A]
 2   val writerIOprg: WriterT[IO,List[String],Int] = for {
 3     _ <- putLn("enter dividend:").liftM[WriterTIO]
 4     dvdn <- readLn.liftM[WriterTIO]
 5     _ <- WriterT.writerT((List(s"received dividend $dvdn;"),dvdn).point[IO])
 6     _ <- putLn("enter divisor:").liftM[WriterTIO]
 7     dvsor <- readLn.liftM[WriterTIO]
 8     _ <- WriterT.writerT(IO(List(s"received divisor $dvsor, ready to divide ..."),dvdn))
 9     quot <- div(dvdn.toInt, dvsor.toInt).except(e => IO({println(e.getMessage());-99})).liftM[WriterTIO]
10     _ <- if (quot < 0) WriterT.writerT((List(s"divide by zero Error!!!"),-99).point[IO]) else putLn(s"the result:$quot").liftM[WriterTIO]
11   } yield (quot)
12 ...
13 object IOMonadDemo extends App {
14   import IOMonadPrg._
15   //  ioprg.unsafePerformIO()
16   //optionIOprg.run.unsafePerformIO()
17   println(writerIOprg.run.unsafePerformIO())
18 ...
19 "enter dividend:"
20 3
21 "enter divisor:"
22 0
23 / by zero
24 (List(received dividend 3;, received divisor 0, ready to divide ..., divide by zero Error!!!),-99)
25 
26 Process finished with exit code 0

以上例子调用了scalaz IO Monad typeclass 提供的except方法，scalaz还提供了其它的异常处理函数：

 /** Executes the handler if an exception is raised. */
  def except(handler: Throwable => IO[A]): IO[A] =
    io(rw => try { Free.pure(this(rw).run) } catch { case e: Throwable => handler(e)(rw) })

  /**
   * Executes the handler for exceptions that are raised and match the given predicate.
   * Other exceptions are rethrown.
   */
  def catchSome[B](p: Throwable => Option[B], handler: B => IO[A]): IO[A] =
    except(e => p(e) match {
      case Some(z) => handler(z)
      case None => throw e
    })

  /**
   * Returns a disjunction result which is right if no exception was raised, or left if an
   * exception was raised.
   */
  def catchLeft: IO[Throwable \/ A] =
    map(\/.right[Throwable, A]) except (t => IO(-\/(t)))

  /**Like "catchLeft" but takes a predicate to select which exceptions are caught. */
  def catchSomeLeft[B](p: Throwable => Option[B]): IO[B \/ A] =
    catchLeft map (_.leftMap(e => p(e).getOrElse(throw e)))

  /**Like "finally", but only performs the final action if there was an exception. */
  def onException[B](action: IO[B]): IO[A] = this except (e => for {
    _ <- action
    a <- (throw e): IO[A]
  } yield a)

以下是这次讨论的完整示范源代码：

 1 package demo.app
 2 
 3 import scalaz._
 4 import Scalaz._
 5 import effect._
 6 import IO._
 7 import Free._
 8 import scala.language.higherKinds
 9 import scala.language.implicitConversions
10 
11 object IOMonadPrg {
12   def div(dvdn: Int, dvsor: Int): IO[Int] =
13     IO(dvdn / dvsor)
14   val ioprg: IO[Int] = for {
15     _ <- putLn("enter dividend:")
16     dvdn <- readLn
17     _ <- putLn("enter divisor:")
18     dvsor <- readLn
19     quot <- div(dvdn.toInt, dvsor.toInt)
20     _ <- putLn(s"the result:$quot")
21   } yield quot
22   //implicit def ioToOptionT[A](io: IO[A]): OptionT[IO,A] = io.liftM[OptionT]
23   val optionIOprg: OptionT[IO,Int] = for {
24     _ <- putLn("enter dividend:").liftM[OptionT]
25     dvdn <- readLn.liftM[OptionT]
26     _ <- putLn("enter divisor:").liftM[OptionT]
27     dvsor <- readLn.liftM[OptionT]
28     a <- if (dvsor.toInt == 0 ) OptionT(IO(None: Option[String])) else IO(0).liftM[OptionT]
29     quot <- div(dvdn.toInt, dvsor.toInt).liftM[OptionT]
30     _ <- putLn(s"the result:$quot").liftM[OptionT]
31   } yield quot
32   type WriterTIO[F[_],A] = WriterT[F,List[String],A]
33   val writerIOprg: WriterT[IO,List[String],Int] = for {
34     _ <- putLn("enter dividend:").liftM[WriterTIO]
35     dvdn <- readLn.liftM[WriterTIO]
36     _ <- WriterT.writerT((List(s"received dividend $dvdn;"),dvdn).point[IO])
37     _ <- putLn("enter divisor:").liftM[WriterTIO]
38     dvsor <- readLn.liftM[WriterTIO]
39     _ <- WriterT.writerT(IO(List(s"received divisor $dvsor, ready to divide ..."),dvdn))
40     quot <- div(dvdn.toInt, dvsor.toInt).except(e => IO({println(e.getMessage());-99})).liftM[WriterTIO]
41     _ <- if (quot < 0) WriterT.writerT((List(s"divide by zero Error!!!"),-99).point[IO]) else putLn(s"the result:$quot").liftM[WriterTIO]
42   } yield (quot)
43 
44 
45 }
46 
47 object IOMonadDemo extends App {
48   import IOMonadPrg._
49   //  ioprg.unsafePerformIO()
50   //optionIOprg.run.unsafePerformIO()
51   println(writerIOprg.run.unsafePerformIO())
52 }

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2016-05-10 ，如有侵权请联系 cloudcommunity@tencent.com 删除

scala