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)

转载于:https://www.cnblogs.com/tiger-xc/p/5690794.html

Scalaz（49）－ scalaz-stream: 深入了解-Sink/Channel相关推荐

Scalaz（23）－泛函数据结构： Zipper-游标定位
外面沙尘滚滚一直向北去了,意识到年关到了,码农们都回乡过年去了,而我却留在这里玩弄"拉链".不要想歪了,我说的不是裤裆拉链而是scalaz Zipper,一种泛函数据结构游标(cu ...
SpringCloud2020学习笔记13——SpringCloud Stream消息驱动
目录一.消息驱动概述 1.简介 2.官网 2.设计思想 ① 标准MQ ② 为什么用Cloud Stream ③ Stream中的消息通信方式遵循了发布-订阅模式 3.Spring Cloud Str ...
十八、Stream 流
目录 1.为什么要引入SpringCloud Stream 2.SpringCloud Stream简介 2.1.标准MQ架构图 2.2.SpringCloud Stream架构图 2.3.Sprin ...
【进阶技术】一篇文章搞掂：Spring Cloud Stream
本文总结自官方文档http://cloud.spring.io/spring-cloud-static/spring-cloud-stream/2.1.0.RC3/single/spring-clou ...
Spring Cloud Stream中文指导手册
Spring Cloud Stream中文指导手册 source 文章目录 Spring Cloud Stream中文指导手册 @[toc] Spring Cloud Stream 核心 1.简介 2 ...
OpenShift 4 - Knative教程 (6) Eventing之Channel和Subscription
<OpenShift 4.x HOL教程汇总> 说明:本文已经在OpenShift 4.10环境中验证文章目录概念操作参考概念 Knative Eventing中的Channel ...
Netty导学之NIO，Channel、Buffer、Selector详解
介绍 NIO可翻译为Non-Blocking IO非阻塞IO,也可以称其为New IO 因为其是JDK1.4新出现的. java中的流要么是输入流,要么是输出流,不可能都是,它面向流编程.而在NIO中 ...
JavaSE-23.2.1【Strem流、Stream流的使用和常见生成方式】
1 package day14.lesson2;2 3 import java.util.ArrayList;4 5 /*6 2 Strem流7 8 2.1 体验Stream流9 10 案例需求 11 ...
Spring Cloud 分布式消息—Spring Cloud Stream 自定义通道与分组分区应用
在Spring Cloud 分布式消息-Spring Cloud Stream 简介与入门一篇我们简单了介绍了Spring Cloud Stream,并且使用Spring Cloud Stream提供 ...
Flink-输出算子(Sink)使用
5.5 输出算子 5.5.1 概述调用print是返回输出类,作为最后一环sink存在该方法创建了一个PrintSinkFunction 操作,然后作为addSink方法的参数 PrintSink ...

Scalaz（49）－ scalaz-stream: 深入了解-Sink/Channel

Scalaz（49）－ scalaz-stream: 深入了解-Sink/Channel相关推荐

最新文章

热门文章