标签：编程 scala functional programmi
在前面的几节讨论里我们终于得出了一个概括又通用的IO Process类型Process[F[_],O]。这个类型同时可以代表数据源（Source）和数据终端（Sink）。在这节讨论里我们将针对Process[F,O]的特性通过一些应用实例来示范它的组合性（composibility）和由数据源到接收终端IO全过程的功能完整性。
我们已经在前面的讨论中对IO Process的各种函数组合进行了调研和尝试，现在我们先探讨一下数据源设计方案：为了实现资源使用的安全性和IO程序的可组合性，我们必须保证无论在完成资源使用或出现异常时数据源都能得到释放，同时所有副作用的产生都必须延后直至Interpreter开始运算IO程序时。
我们先试试一个读取文件字符内容的组件：
import java.io.{BufferedReader, FileReader}
def readFile(fileName: String): Process[IO,String] =
await[IO,BufferedReader,String](IO{new BufferedReader(new FileReader(fileName))}){
case Left(err) => Halt(err)
case Right(r) => {
lazy val next: Process[IO,String] = await(IO{r.readLine}) {
case Left(err2) => await(IO{r.close}){_ => Halt[IO,String](err2)}()
case Right(line) => emit(line, next)
}()
next
}
}()
注意以下几个问题：首先是所有的IO动作都是通过await函数来实现的，再就是所有产生副作用的语句都被包嵌在IO{}里，这是典型的延迟运算。我们先来看看这个await函数：
def await[F[_],A,O](req: F[A])(
rcvfn: Either[Throwable,A] => Process[F,O] = (a: Either[Throwable,A]) => Halt[F,O](End))
(fallback: Process[F,O] = Halt[F,O](End),
onError: Process[F,O] = Halt[F,O](End)): Process[F,O] = Await(req,rcvfn,fallback,onError)
req是个F[A]，在例子里就是IO[A]。把这个函数精简表述就是: await(IO(iorequest))(ioresult => Process)。从这个函数的款式可以看出：在调用await函数的时候并不会产生副作用，因为产生副作用的语句是包嵌在IO｛｝里面的。我们需要一个interpreter来运算这个IO{...}产生副作用。await的另一个输入参数是 iorequest => Process：iorequest是运算iorequest返回的结果：可以是A即String或者是个异常Exception。所以我们可以用PartialFunction来表达：
｛ case Left(err) => Process[IO,???]; case Right(a) => Process[IO,???] }
我们用PartialFunction来描述运算iorequest后返回正常结果或者异常所对应的处理办法。
上面的例子readFile就是用这个await函数来打开文件： IO {new BufferedReader(new FileReader(fileName))}
读取一行：IO {r.readLine}，如果读取成功则发送emit出去: case Right(line) => emit(line,next),
如果出现异常则关闭文件：case Left(err) => IO {r.close}，注意我们会使用异常End来代表正常完成读取：
case object End extends Exception case object Kill extends Exception
def collect[O](src: Process[IO,O]): IndexedSeq[O] = {
val E = java.util.concurrent.Executors.newFixedThreadPool(4)
def go(cur: Process[IO,O], acc: IndexedSeq[O]): IndexedSeq[O] = cur match {
case Halt(e) => acc
case Emit(os,ns) => go(ns, acc ++ os)
case Halt(err) => throw err
case Await(rq,rf,fb,fl) =>
val next =
try rf(Right(unsafePerformIO(rq)(E)))
catch { case err: Throwable => rf(Left(err)) }
go(next, acc)
}
try go(src, IndexedSeq())
finally E.shutdown
}
首先要注意的是这句：unsafePerformIO(rq)(E)。它会真正产生副作用。当Process[IO,O] src当前状态是Await的时候就会进行IO运算。运算IO产生的结果作为Await的rf函数输入参数，正如我们上面描述的一样。所以，运算IO{iorequest}就是构建一个Await结构把iorequest和转换状态函数rf放进去就像这样：
await(iorequest)(rf)(fb,fl) = Await(ioreques,rf,fb,fl),
然后返回到collect，collect看到src状态是Await就会运算iorequest然后再运行rf。
我们的下一个问题是如何把文件里的内容一行一行读入而不是一次性预先全部搬进内存，这样我们可以读一行，处理一行，占用最少内存。我们再仔细看看readFile的这个部分：
def readFile(fileName: String): Process[IO,String] =
await[IO,BufferedReader,String](IO{new BufferedReader(new FileReader(fileName))}){
case Left(err) => Halt(err)
case Right(r) => {
lazy val next: Process[IO,String] = await(IO{r.readLine}) {
case Left(err2) => Halt[IO,String](err2) //await(IO{r.close}){_ => Halt[IO,String](err2)}()
case Right(line) => emit(line, next)
}()
next
}
}()
//状态进位，输出Process[F,O2]
final def drain[O2]: Process[F,O2] = this match {
case Halt(e) => Halt(e) //终止
case Emit(os,ns) => ns.drain //运算下一状态ns，输出
case Await(rq,rf,fb,cl) => Await(rq, rf andThen (_.drain)) //仍旧输出Await
}
readFile("farenheit.txt").filter(line => !line.startsWith("#").map(line => line.toUpperCase).drain
那么我们就会得到读取一行字符；过滤起始为#的行；转成大写字符；返回再读一行交替循环这样的效果了。
很明显readFile实在太有针对性了。函数类型款式变的复杂可读性低。我们需要一种更概括的形式来实现泛函编程语言的简练而流畅表达形式。
我们首先应该把IO运算方式重新定义一下。用await函数显得太复杂：
//await 的精简表达形式
def eval[F[_],A](fa: F[A]): Process[F,A] = //运算F[A]
await[F,A,A](fa){
case Left(err) => Halt(err)
case Right(a) => emit(a, Halt(End))
}()
def evalIO[A](ioa: IO[A]) = eval[IO,A](ioa) //运算IO[A]
//确定终结的运算
def eval_[F[_],A,B](fa: F[A]): Process[F,B] = eval[F,A](fa).drain[B] //运算F[A]直到终止
再来一个通用安全的IO资源使用组件函数：
def resource[R,O]( //通用IO程序运算函数
acquire: IO[R])( //获取IO资源。open file
use: R => Process[IO,O])( //IO运算函数 readLine
release: R => Process[IO,O]): Process[IO,O] = //释放资源函数 close file
eval(acquire) flatMap { r => use(r).onComplete(release(r)) }
def resource_[R,O]( //与resource一样，只是运算realease直至终止
acquire: IO[R])( //获取IO资源。open file
use: R => Process[IO,O])( //IO运算函数 readLine
release: R => IO[Unit]): Process[IO,O] = //释放资源函数 close file
resource(acquire)(use)(release andThen (eval_[IO,Unit,O]))
def lines(fileName: String): Process[IO,String] = //从fileName里读取
resource
{IO {io.Source.fromFile(fileName)}} //占用资源
{src => //使用资源。逐行读取
lazy val iter = src.getLines
def nextLine = if (iter.hasNext) Some(iter.next) else None //下一行
lazy val getLines: Process[IO,String] = //读取
eval(IO{nextLine}) flatMap { //运算IO
case None => Halt(End) //无法继续读取：完成或者异常
case Some(line) => emit(line, getLines) //读取然后发送
}
getLines
}
{src => eval_ (IO{src.close})} //释放资源
这样易读易解多了。
现在我们应该可以很简练但又不失清楚详尽地描述一段IO程序：
打开文件fahrenheit.txt
读取一行字符
过滤空行或者以#开始的字行，可通过的字行代表亨氏温度数
把亨氏温度转换成摄氏温度数
这里面的温度转换函数如下：
def fahrenheitToCelsius(f: Double): Double =
(f - 32) * 5.0/9.0
lines("fahrenheit.txt").
filter(line => !line.startsWith("#") && !line.trim.isEmpty).
map(line => fahrenheitToCelsius(line.toDouble).toString).
drain
现在到了了解IO过程的另一端：Sink的时候了。我们如果需要通过Process来实现输出功能的话，也就是把Source[O]的这个O发送输出到一个Sink。实际上我们也可以用Process来表达Sink，先看一个简单版本的Sink如下：
type SimpleSink[F[_],O] = Process[F,O => F[Unit]]
def simpleWriteFile(fileName: String, append: Boolean = false) : SimpleSink[IO, String] =
resource[FileWriter, String => IO[Unit]]
{IO {new FileWriter(fileName,append)}} //acquire
{w => IO{(s:String) => IO{w.write(s)}}} //use
{w => IO{w.close}} //release
type Sink[F[_],O] = Process[F, O => Process[F,Unit]]
import java.io.FileWriter
def fileW(file: String, append: Boolean = false): Sink[IO,String] =
resource[FileWriter, String => Process[IO,Unit]]
{ IO { new FileWriter(file, append) }}
{ w => stepWrite { (s: String) => eval[IO,Unit](IO(w.write(s))) }} //重复循环逐行写
{ w => eval_(IO(w.close)) }
/* 一个无穷循环恒量stream. */
def stepWrite[A](a: A): Process[IO,A] =
eval(IO(a)).flatMap { a => Emit(a, stepWrite(a)) } 通过Emit的下一状态重复运算IO(a)
下一步是把Sink和Process对接起来。我们可以用以下的to组件来连接：
def to[O2](sink: Sink[F,O]): Process[F,Unit] =
join { (this zipWith sink)((o,f) => f(o)) }
def join[F[_],A](p: Process[F,Process[F,A]]): Process[F,A] =
p.flatMap(pa => pa)
val converter: Process[IO,Unit] =
lines("fahrenheit.txt"). //读取
filter(line => !line.startsWith("#") && !line.trim.isEmpty). //过滤
map(line => fahrenheitToCelsius(line.toDouble).toString). //温度转换
pipe(intersperse("\n")). //加end of line
to(fileW("celsius.txt")). //写入
drain //继续循环
type Channel[F[_],I,O] = Process[F, I => Process[F,O]]
我们用Channel来描述一个数据库查询：
import java.sql.{Connection, PreparedStatement, ResultSet}
def query(conn: IO[Connection]):
Channel[IO, Connection => PreparedStatement, Map[String,Any]] = //Map === Row
resource_ //I >>> Connection => PreparedStatement
{ conn } //打开connection
{ conn => constant { (q: Connection => PreparedStatement) => //循环查询
resource_
{ IO { //运行query
val rs = q(conn).executeQuery
val ncols = rs.getMetaData.getColumnCount
val cols = (1 to ncols).map(rs.getMetaData.getColumnName)
(rs, cols)
}}
{ case (rs, cols) => //读取纪录Row
def step =
if (!rs.next) None
else Some(cols.map(c => (c, rs.getObject(c): Any)).toMap)
lazy val rows: Process[IO,Map[String,Any]] = //循环读取
eval(IO(step)).flatMap {
case None => Halt(End)
case Some(row) => Emit(row, rows) //循环运算rows函数
}
rows
}
{ p => IO { p._1.close } } // close the ResultSet
}}
{ c => IO(c.close) }
从一个文件里读取存放亨氏温度的文件名后进行温度转换并存放到celsius.txt中
val convertAll: Process[IO,Unit] = (for {
out <- fileW("celsius.txt").once // out的类型是String => Process[IO,Unit]
file <- lines("fahrenheits.txt") //fahrenheits.txt里保存了一串文件名
_ <- lines(file). //动态打开文件读取温度记录
map(line => fahrenheitToCelsius(line.toDouble)). //温度系统转换
flatMap(celsius => out(celsius.toString)) //输出
} yield ()) drain //继续循环
输出到多个.celsius文件：
val convertMultisink: Process[IO,Unit] = (for {
file <- lines("fahrenheits.txt") //读取文件名称
_ <- lines(file). //打开文件读取温度数据
map(line => fahrenheitToCelsius(line.toDouble)). //温度系统转换
map(_ toString).
to(fileW(file + ".celsius")) //写入文件
} yield ()) drain
val convertMultisink2: Process[IO,Unit] = (for {
file <- lines("fahrenheits.txt")
_ <- lines(file).
filter(!_.startsWith("#")). //过滤#开始字串
map(line => fahrenheitToCelsius(line.toDouble)).
filter(_ > 0). // 过滤0度以下温度
map(_ toString).
to(fileW(file + ".celsius"))
} yield ()) drain
版权声明：本文为博主原创文章，未经博主允许不得转载。
泛函编程（38）－泛函Stream IO：IO Process in action
标签：编程 scala functional programmi
原文地址：http://blog.csdn.net/tiger_xc/article/details/47948289
