Scalaz(49)- scalaz-stream: 深入了解-Sink/Channel

   一个完整的scalaz-stream有以下几个部分组成:Source -> Transducer -> Sink,用直白文字来描述就是:“输入 -> 传换 -> 输出”。我们已经在前面几篇讨论中介绍了Source和Transducer的基本情况,这篇讨论我们探讨一下Sink。scalaz-stream最基本的功能就是从Source接收一串元素,经过处理然后输出。毕竟我们从外部获取了数据、处理完毕后总不能就留在原地内存,还是要把数据输出到对当前运算中程序来说的一个外部系统。Sink就等于是这个外部系统的输入接口。与往常一样,我们先来看看Sink的类型款式:

 /**
   * An effectful sink, to which we can send values. Modeled
   * as a source of effectful functions.
   */
  type Sink[+F[_],-O] = Process[F, O => F[Unit]]

从类型款式看来Sink就是一种Process。只不过它包含的元素是一串函数(O => F[Unit])。这里的O是数据元素,F[Unit]的意思是对O进行F运算后不返回任何结果,如:Task{ println(...)}。我们先构造一个简单的Sink:

1 val sinkStdout: Sink[Task,String] =
2    Process.repeatEval {
3       Task.delay { (s: String) =>
4         Task.delay { println(s) }}}               //> sinkStdout  : scalaz.stream.Sink[scalaz.concurrent.Task,String] = Append(Await(scalaz.concurrent.Task@702b8b12,<function1>,<function1>),Vector(<function1>))
5    
6  val sinkStdout2: Sink[Task,String] =
7    Process.constant { (s: String) =>
8    Task.delay { println(s) }}                     //> sinkStdout2  : scalaz.stream.Sink[scalaz.concurrent.Task,String] = Append(Emit(Vector(<function1>)),Vector(<function1>))

我们应该怎样把数据传给Sink呢?首先我们可以用tee.zip:

1  (range(1,6) zip sinkStdout).flatMap {
2    case (i,f) => eval (f(i.toString))
3  }.run.run                                        //> 1
4                                                   //| 2
5                                                   //| 3
6                                                   //| 4
7                                                   //| 5

实际上scalaz-stream提供了to函数来支持Sink连接。to还是通过tee.zip来实现的:

  /** Feed this `Process` through the given effectful `Channel`. */
    def through[F2[x]>:F[x],O2](f: Channel[F2,O,O2]): Process[F2,O2] =
      self.zipWith(f)((o,f) => f(o)).eval onHalt { _.asHalt }     
  /** Attaches `Sink` to this  `Process`  */
    def to[F2[x]>:F[x]](f: Sink[F2,O]): Process[F2,Unit] =
      through(f)

我们用to来重复示范上面的例子:

1 (range(1,6).map(_.toString) to sinkStdout).run.run
2                                                   //> 1
3                                                   //| 2
4                                                   //| 3
5                                                   //| 4
6                                                   //| 5

可以说用to表述更简洁。如果我们需要把数据发送到多个外部系统,那我们就必须连接多个Sink了,可以用zip来连接多个Sink:

 1  (range(1,6) zip sinkStdout zip sinkStdout2).flatMap {
 2    case (((i,f),f2)) => for {
 3      _ <- eval(f(i.toString))
 4      _ <- eval(f2(i.toString))
 5    } yield ()
 6  }.run.run                                        //> 1
 7                                                   //| 1
 8                                                   //| 2
 9                                                   //| 2
10                                                   //| 3
11                                                   //| 3
12                                                   //| 4
13                                                   //| 4
14                                                   //| 5
15                                                   //| 5

scalaz-stream提供的observe函数可以像一个分流器一样安插在数据流中间复制一份数据发送到一个Sink而不影响正在流动的数据:

 1 (range(1,4).map(_.toString) observe sinkStdout observe sinkStdout2 to sinkStdout)
 2 .run.run                                          //> 1
 3                                                   //| 1
 4                                                   //| 1
 5                                                   //| 2
 6                                                   //| 2
 7                                                   //| 2
 8                                                   //| 3
 9                                                   //| 3
10                                                   //| 3

以上例子相当于连接了3个Sink。observe通常被用来跟踪流中数据,因为它不会影响数据流的正常运算。我们也可以把多个Sink zip成一个多功能的Sink。与上面例子不同的是它只有一个输出口:

 1 import scalaz._
 2 import Scalaz._
 3 import scalaz.stream._
 4 import scalaz.concurrent._
 5 import scala.language.higherKinds
 6 object streamLogDemo {
 7   sealed trait Loglevel
 8   case object Info extends Loglevel
 9   case object Debug extends Loglevel
10   case object Warning extends Loglevel
11   
12   case class Line(level: Loglevel, line: String)
13   //Sinks
14   val outInfo = io.stdOutLines.contramap {(l: Line) => "Info: " + l.line}
15                                                   //> outInfo  : scalaz.stream.Channel[scalaz.concurrent.Task,Line,Unit] = Append(Emit(Vector(<function1>)),Vector(<function1>))
16   val outDebug = io.stdOutLines.contramap {(l: Line) => "Debug: " + l.line}
17                                                   //> outDebug  : scalaz.stream.Channel[scalaz.concurrent.Task,Line,Unit] = Append(Emit(Vector(<function1>)),Vector(<function1>))
18   val outWarning = io.stdOutLines.contramap {(l: Line) => "Warning: " + l.line}
19                                                   //> outWarning  : scalaz.stream.Channel[scalaz.concurrent.Task,Line,Unit] = Append(Emit(Vector(<function1>)),Vector(<function1>))
20   
21   val zippedSink = (outInfo zip outDebug zip outWarning).map {
22     case ((fInfo,fDebug), fWarning) =>
23       (l: Line) => l.level match {
24         case Info    => fInfo(l)
25         case Debug   => fDebug(l)
26         case Warning => fWarning(l)
27       }
28   }                                               //> zippedSink  : scalaz.stream.Process[[x]scalaz.concurrent.Task[x],Line => scalaz.concurrent.Task[Unit]] = Append(Halt(End),Vector(<function1>))
29   //test data
30   val lines: List[Line] = List(
31     Line(Info, "Hello"),
32     Line(Debug, "buddy"),
33     Line(Warning, "nanana"))                      //> lines  : List[Line] = List(Line(Info,Hello), Line(Debug,buddy), Line(Warning,nanana))
34   //test run
35   (Process.emitAll(lines) to zippedSink).run.run  //> Info: Hello
36                                                   //| Debug: buddy
37                                                   //| Warning: nanana
38  }

仔细观察我们可以发现上面的例子里outInfo,outDebug,outWarning的最终类型变成了Channel,io.stdOutLines的类型应该是Sink,那么Channel应该是某种Sink了。我们看看Channel的类型款式:

/**
   * An effectful sink, to which we can send values. Modeled
   * as a source of effectful functions.
   */
  type Sink[+F[_],-O] = Process[F, O => F[Unit]]

  /**
   * An effectful channel, to which we can send values and
   * get back responses. Modeled as a source of effectful
   * functions.
   */
  type Channel[+F[_],-I,O] = Process[F, I => F[O]]

从类型款式上对比Sink和Channel的不同之处只在这个与外界系统的接口函数:Channel的函数是I => F[O],它返回了F运算结果。这也很容易理解:Sink正如其名,任何东西进入Sink就如泥牛入海,无所回应。Channel会返回接收数据方的信息。Sink和Channel之间是可以相互转换的。Channel有个lift函数,能把一个A => F[B]函数升格成Channel类型:

 /** Promote an effectful function to a `Channel`. */
  def lift[F[_],A,B](f: A => F[B]): Channel[F, A, B] =
    Process constant f

实际上我如果用这个lift把A => F[Unit]这样的函数升格,就会得到一个Sink了。这个从Sink的lift函数中可以证实:

object sink {

  /** Promote an effectful function to a `Sink`. */
  def lift[F[_], A](f: A => F[Unit]): Sink[F, A] = channel lift f

}

我们用一些例子来体验一下:

 1  val ch = channel.lift((i: Int) => Task.delay { i * 3 })
 2                                                   //> ch  : scalaz.stream.Channel[scalaz.concurrent.Task,Int,Int] = Append(Emit(Vector(<function1>)),Vector(<function1>))
 3   (Process.range(1,6) through ch).runLog.run      //> res0: Vector[Int] = Vector(3, 6, 9, 12, 15)
 4   val sn = channel.lift((i: Int) => Task.delay { println(i * 3);()})
 5                                                   //> sn  : scalaz.stream.Channel[scalaz.concurrent.Task,Int,Unit] = Append(Emit(Vector(<function1>)),Vector(<function1>))
 6   (Process.range(1,6) to sn).run.run              //> 3
 7                                                   //| 6
 8                                                   //| 9
 9                                                   //| 12
10                                                   //| 15
11   val sn1 = sink.lift((i: Int) => Task.delay { println(i * 3) })
12                                                   //> sn1  : scalaz.stream.Sink[scalaz.concurrent.Task,Int] = Append(Emit(Vector(<function1>)),Vector(<function1>))
13   (Process.range(1,6) to sn1).run.run             //> 3
14                                                   //| 6
15                                                   //| 9
16                                                   //| 12
17                                                   //| 15

如果我们的系统需要与外部多个系统对接的话,我们同样可以把一个数据源发送到多个Sink和Channel:

 1 (Process.range(1,6) observe sn observe sn1 through ch).runLog.run
 2                                                   //> 3
 3                                                   //| 3
 4                                                   //| 6
 5                                                   //| 6
 6                                                   //| 9
 7                                                   //| 9
 8                                                   //| 12
 9                                                   //| 12
10                                                   //| 15
11                                                   //| 15
12                                                   //| res1: Vector[Int] = Vector(3, 6, 9, 12, 15)

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏nnngu

016 Java中的动态代理

代理 代理模式是一种常用的设计模式,其目的就是为其他对象提供一个代理以控制对某个真实对象的访问。代理类负责为委托类预处理消息,过滤消息并转发消息,以及进行消息被...

34910
来自专栏小勇DW3

承接上文

前置通知[Before advice]:在连接点前面执行,前置通知不会影响连接点的执行,除非此处抛出异常。 

722
来自专栏你不就像风一样

抛弃配置后的Spring终极教程

对于Spring,核心就是IOC容器,这个容器说白了就是把你放在里面的对象(Bean)进行统一管理,你不用考虑对象如何创建如何销毁,从这方面来说,所谓的控制反转...

792
来自专栏张善友的专栏

在Entity Framework 中执行T-sql语句

从Entity Framework  4开始在ObjectContext对象上提供了2个方法可以直接执行SQL语句:ExecuteStoreQuery<T> 和...

17910
来自专栏浪淘沙

java学习day15---UDP,reflect,proxy

1.UDP udp是一种不安全的协议 容易丢失数据 不需要建立连接 效率高 数据是以数据包的形式发送 数据 IP 端口

865
来自专栏IT可乐

Spring详解(二)------IOC控制反转

  我相信提到 Spring,很多人会脱口而出IOC(控制反转)、DI(依赖注入)、AOP等等概念,这些概念也是面试官经常问到的知识点。那么这篇博客我们就来详细...

1865
来自专栏lgp20151222

Spring中的@scope注解

但是也可以理解成,singleton是启动创建,prototype/request/session/globalsession是动态创建。

831
来自专栏CodingToDie

FastSql ORM 实现

FastSql 中 ORM 的实现 Table of Contents 原理 实现 1. 使用注解 2. 反射工具类 3. 简单的 model 4. 注解解析 ...

4116
来自专栏coolblog.xyz技术专栏

Spring AOP 源码分析 - 筛选合适的通知器

从本篇文章开始,我将会对 Spring AOP 部分的源码进行分析。本文是 Spring AOP 源码分析系列文章的第二篇,本文主要分析 Spring AOP ...

763
来自专栏xiaoheike

Hibernate save, saveOrUpdate, persist, merge, update 区别

hibernate save()方法能够保存实体到数据库,正如方法名称save这个单词所表明的意思。我们能够在事务之外调用这个方法,这也是我不喜欢使用这个方法保...

823

扫码关注云+社区