scalaz-stream-fs2是一种函数式的数据流编程工具。fs2的类型款式是：Stream[F[_],O]，F[_]代表一种运算模式，O代表Stream数据元素的类型。实际上F就是一种延迟运算机制：F中间包含的类型如F[A]的A是一个可能会产生副作用不纯代码（impure code）的运算结果类型，我们必须用F对A运算的延迟机制才能实现编程过程中的函数组合（compositionality），这是函数式编程的标准做法。如果为一个Stream装备了F[A]，就代表这个Stream会在处理数据元素O的过程中对O施用运算A，如果这个运算A会与外界交互（interact with outside world）如：文件、数据库、网络等的读写操作，那么这个Stream有数据元素I/O功能的需求。我们可以通过fs2 Stream的状态机器特性（state machine）及F[A]与外界交互功能来编写完整的数据处理（data processing）程序。如果能够在数据库程序编程中善用fs2的多线程运算模式来实现对数据库存取的并行运算，将会大大提高数据处理的效率。我们将在本篇着重讨论fs2在实现I/O程序中的有关方式方法。
首先，我们需要以整体Stream为程序运算框架，把与外界交互的运算A串联起来，然后通过Stream的节点来代表程序状态。我们首先需要某种方式把F[A]与Stream[F,A]关联起来，也就是我们所说的把一个F[A]升格成Stream[F,A]。fs2提供了Stream.eval函数，我们看看它的类型款式：
def eval[F[_], A](fa: F[A]): Stream[F, A] = attemptEval(fa) flatMap { _ fold(fail, emit) }
很明显，提供一个F[A]，eval返回Stream[F,A]。这个返回结果Stream[F,A]的元素A是通过运算F[A]获取的：在一个数据库程序应用场景里这个A可能是个数据库连接（connection），那么F[A]就是一个连接数据库的操作函数，返回的A是个连接connection。这次我们来模拟一个对数据库表进行新纪录存储的场景。一般来说我们会按以下几个固定步骤进行：
1、连接数据库，获取connection连接
2、产生新数据（在其它场景里可能是读取数据然后更新）。这可能是一个循环的操作
3、将数据写入数据库
这三个步骤可以用Stream的三种状态来表示：一个源头（source）、传转（pipe transducer）、终点（sink)。
我们先示范如何构建源头：这是一种占用资源的操作，会产生副作用，所以我们必须用延迟运算方式来编程：
//用Map模拟数据库表
import scala.collection.mutable.Map
type DataStore = Map[Long, String]
val dataStore: DataStore = Map() //> dataStore : fs2Eval.DataStore = Map()
case class Connection(id: String, store: DataStore)
def src(producer: String): Stream[Task,Connection] =
Stream.eval(Task.delay { Connection(producer,dataStore)})
//> src: (producer: String)fs2.Stream[fs2.Task,fs2Eval.Connection]
这个示范用了一个mutable map类型来模拟会产生副作用的数据库表。我们把具体产生数据的源头用Connection.id传下去便于在并行运算示范里进行跟踪。在这个环节里我们模拟了连接数据库dataStore操作。
产生数据是在内存里进行的，不会使用到connection，但我们依然需要把这个connection传递到下个环节：
case class Row(conn: Connection, key: Long, value: String)
val recId = new java.util.concurrent.atomic.AtomicLong()
//> recId : java.util.concurrent.atomic.AtomicLong = 1
def createData(conn: Connection): Row =
Row(conn, recId.incrementAndGet, s"Producer $conn.id: at ${System.currentTimeMillis}")
//> createData: (conn: fs2Eval.Connection)fs2Eval.Row
val trans: Pipe[Task,Connection,Row] = _.map {conn => createData(conn)}
//> trans : fs2.Pipe[fs2.Task,fs2Eval.Connection,fs2Eval.Row] = <function1>
trans是个Pipe。我们可以用through把它连接到src。
向数据库读写都会产生副作用。下一个环节我们模拟把trans传递过来的Row写入数据库。这里我们需要用延迟运算机制：
def log: Pipe[Task, Row, Row] = _.evalMap { r =>
Task.delay {println(s"saving row pid:${r.conn.id}, rid:${r.key}"); r}}
def saveRow(row: Row) = row.conn.store += (row.key -> row.value)
val snk: Sink[Task,Row] = _.evalMap { r =>
Task.delay { saveRow(r); () } }
增加了个跟踪函数log。从上面的代码可以看出：实际上Sink就是Pipe，只不过返回了()。
我们试试把这几个步骤连接起来运算一下：
val sprg = src("").through(trans).repeat.take().through(log).to(snk)
//> sprg : fs2.Stream[fs2.Task,Unit] = evalScope(Scope(Bind(Eval(Snapshot),<function1>))).flatMap(<function1>).flatMap(<function1>).flatMap(<function1>).flatMap(<function1>)
sprg.run.unsafeRun //> saving row pid:001, rid:2
//| saving row pid:001, rid:3
//| saving row pid:001, rid:4
println(dataStore) //> Map(2 -> Connection(001,Map()).id: at 1472605736214, 4 -> Connection(001,Map(2 -> Connection(001,Map()).id: at 1472605736214, 3 -> Connection(001,Map(2 -> Connection(001,Map()).id: at 1472605736214)).id: at 1472605736245)).id : at 1472605736248, 3 -> Connection(001,Map(2 -> Connection(001,Map()).id: at 1472605736214)).id: at 1472605736245)
我们看到mutable map dataStore内容有变化了。
如果我们把以上的例子用并行运算方式来实现的话，应该如何调整？为方便观察结果，我们先在几个环节增加一些时间延迟：
implicit val strategy = Strategy.fromFixedDaemonPool()
implicit val scheduler = Scheduler.fromFixedDaemonPool()
def src(producer: String): Stream[Task,Connection] =
Stream.eval(Task.delay { Connection(producer,dataStore)}
.schedule(.seconds)) val trans: Pipe[Task,Connection,Row] = _.evalMap {conn =>
Task.delay{createData(conn)}.schedule(.second)}
下面我们把一些类型调整成Stream[Task,Stream[Row]]，然后把concurrent.join函数掺进去：
val srcs = concurrent.join()(Stream(src(""),src(""),src(""),src("")))
//> srcs : fs2.Stream[fs2.Task,fs2Eval.Connection] = attemptEval(Task).flatMap
<function1>).flatMap(<function1>)
val recs: Pipe[Task,Connection,Row] = src => {
concurrent.join()(src.map { conn =>
Stream.repeatEval(Task {createData(conn)}.schedule(.second)) })
} //> recs : fs2.Pipe[fs2.Task,fs2Eval.Connection,fs2Eval.Row] = <function1>
def saveRows(row: Row) = { row.conn.store += (row.key -> row.value); row}
//> saveRows: (row: fs2Eval.Row)fs2Eval.Row
val snks: Pipe[Task,Row,Row] = rs => {
concurrent.join()(rs.map { r =>
Stream.eval(Task {saveRows(r)}.schedule(.second)) })
} //> snks : fs2.Pipe[fs2.Task,fs2Eval.Row,fs2Eval.Row] = <function1>
我们试着把它们连接起来进行运算：
val par = srcs.through(recs).take().through(log("before")).through(chnn).through(log("after"))
//> par : fs2.Stream[fs2.Task,fs2Eval.Row] = attemptEval(Task).flatMap(<function1>).flatMap(<function1>).flatMap(<function1>)
par.run.unsafeRun //> before saving pid:001, rid:3
//| before saving pid:003, rid:2
//| before saving pid:002, rid:4
//| before saving pid:001, rid:5
//| after saving pid:001, rid:3
//| after saving pid:003, rid:2
//| before saving pid:003, rid:6
//| after saving pid:002, rid:4
//| before saving pid:002, rid:7
//| after saving pid:001, rid:5
//| before saving pid:001, rid:8
//| before saving pid:003, rid:9
//| after saving pid:003, rid:6
//| after saving pid:002, rid:7
//| before saving pid:002, rid:10
//| before saving pid:004, rid:11
//| after saving pid:001, rid:8
//| after saving pid:003, rid:9
//| after saving pid:002, rid:10
//| after saving pid:004, rid:11
从跟踪函数显示可以看出before,after是交叉发生的，这就代表已经实现了并行运算。
下面是本篇示范源代码：
import fs2._
import scala.concurrent.duration._
object fs2Eval { //用Map模拟数据库表
import scala.collection.mutable.Map
type DataStore = Map[Long, String]
val dataStore: DataStore = Map()
case class Connection(id: String, store: DataStore)
implicit val strategy = Strategy.fromFixedDaemonPool()
implicit val scheduler = Scheduler.fromFixedDaemonPool()
def src(producer: String): Stream[Task,Connection] =
Stream.eval(Task.delay { Connection(producer,dataStore)}
.schedule(.seconds))
case class Row(conn: Connection, key: Long, value: String)
val recId = new java.util.concurrent.atomic.AtomicLong()
def createData(conn: Connection): Row =
Row(conn, recId.incrementAndGet, s"$conn.id: at ${System.currentTimeMillis}")
val trans: Pipe[Task,Connection,Row] = _.evalMap {conn =>
Task.delay{createData(conn)}.schedule(.second)} def log(pfx: String): Pipe[Task, Row, Row] = _.evalMap { r =>
Task.delay {println(s"$pfx saving pid:${r.conn.id}, rid:${r.key}"); r}}
def saveRow(row: Row) = row.conn.store += (row.key -> row.value) val snk: Sink[Task,Row] = _.evalMap { r =>
Task.delay { saveRow(r); () } } val sprg = src("").through(trans).repeat.take().through(log("")).to(snk)
//sprg.run.unsafeRun
//println(dataStore) val srcs = concurrent.join()(Stream(src(""),src(""),src(""),src("")))
val recs: Pipe[Task,Connection,Row] = src => {
concurrent.join()(src.map { conn =>
Stream.repeatEval(Task {createData(conn)}.schedule(.second)) })
} def saveRows(row: Row) = { row.conn.store += (row.key -> row.value); row}
val chnn: Pipe[Task,Row,Row] = rs => {
concurrent.join()(rs.map { r =>
Stream.eval(Task {saveRows(r)}.schedule(.second)) })
} val par = srcs.through(recs).repeat.take().through(log("before")).through(chnn).through(log("after"))
par.run.unsafeRun