FunDA(7)- Reactive Streams to fs2 Pull Streams

    Reactive-Stream不只是简单的push-model-stream, 它还带有“拖式”(pull-model)性质。这是因为在Iteratee模式里虽然理论上由Enumerator负责主动推送数据,实现了push-model功能。但实际上Iteratee也会根据自身情况,通过提供callback函数通知Enumerator可以开始推送数据,这从某种程度上也算是一种pull-model。换句话讲Reactive-Streams是通过push-pull-model来实现上下游Enumerator和Iteratee之间互动的。我们先看个简单的Iteratee例子:

def showElements: Iteratee[Int,Unit] = Cont {
  case Input.El(e) =>
     println(s"EL($e)")
     showElements
  case Input.Empty => showElements
  case Input.EOF =>
     println("EOF")
     Done((),Input.EOF)
}                                                 //> showElements: => play.api.libs.iteratee.Iteratee[Int,Unit]
val enumNumbers = Enumerator(1,2,3,4,5)           //> enumNumbers  : play.api.libs.iteratee.Enumerator[Int] = play.api.libs.iteratee.Enumerator$$anon$19@47f6473

enumNumbers |>> showElements                      //> EL(1)
                                                  //| EL(2)
                                                  //| EL(3)
                                                  //| EL(4)
                                                  //| EL(5)
                                                  //| res0: scala.concurrent.Future[play.api.libs.iteratee.Iteratee[Int,Unit]] = Success(Cont(<function1>))

我们看到:enumNumbers |>> showElements立刻启动了运算。但并没有实际完成数据发送,因为showElements并没有收到Input.EOF。首先,我们必须用Iteratee.run来完成运算:

val it = Iteratee.flatten(enum |>> consumeAll).run//> El(1)
                                                  //| El(2)
                                                  //| El(3)
                                                  //| El(4)
                                                  //| El(5)
                                                  //| El(6)
                                                  //| El(7)
                                                  //| El(8)
                                                  //| EOF
                                                  //| it  : scala.concurrent.Future[Int] = Success(99)

这个run函数是这样定义的:

/**
   * Extracts the computed result of the Iteratee pushing an Input.EOF if necessary
   * Extracts the computed result of the Iteratee, pushing an Input.EOF first
   * if the Iteratee is in the [[play.api.libs.iteratee.Cont]] state.
   * In case of error, an exception may be thrown synchronously or may
   * be used to complete the returned Promise; this indeterminate behavior
   * is inherited from fold().
   *
   *  @return a [[scala.concurrent.Future]] of the eventually computed result
   */
  def run: Future[A] = fold({
    case Step.Done(a, _) => Future.successful(a)
    case Step.Cont(k) => k(Input.EOF).fold({
      case Step.Done(a1, _) => Future.successful(a1)
      case Step.Cont(_) => sys.error("diverging iteratee after Input.EOF")
      case Step.Error(msg, e) => sys.error(msg)
    })(dec)
    case Step.Error(msg, e) => sys.error(msg)
  })(dec)

再一个问题是:enumNumbers |>> showElements是个封闭的运算,我们无法逐部分截取数据流,只能取得整个运算结果。也就是说如果我们希望把一个Enumerator产生的数据引导到fs2 Stream的话,只能在所有数据都读入内存后才能实现了。这样就违背了使用Reactive-Streams的意愿。那我们应该怎么办?一个可行的方法是使用一个存储数据结构,用两个线程,一个线程里Iteratee把当前数据存入数据结构,另一个线程里fs2把数据取出来。fs2.async.mutable包提供了个Queue类型,我们可以用这个Queue结构来作为Iteratee与fs2之间的管道:Iteratee从一头把数据压进去(enqueue),fs2从另一头把数据取出来(dequeue)。

我们先设计enqueue部分,这部分是在Iteratee里进行的:

def enqueueTofs2(q: async.mutable.Queue[Task,Option[Int]]): Iteratee[Int,Unit] = Cont {
   case Input.EOF =>
       q.enqueue1(None).unsafeRun
       Done((),Input.EOF)
   case Input.Empty => enqueueTofs2(q)
   case Input.El(e) =>
       q.enqueue1(Some(e)).unsafeRun
       enqueueTofs2(q)
}    //> enqueueTofs2: (q: fs2.async.mutable.Queue[fs2.Task,Option[Int]])play.api.libs.iteratee.Iteratee[Int,Unit]

先分析一下这个Iteratee:我们直接把enqueueTofs2放入Cont状态,也就是等待接受数据状态。当收到数据时运行q.enqueue1把数据塞入q,然后不断循环运行至收到Input.EOF。注意:q.enqueue1(Some(e)).unsafeRun是个同步运算,在未成功完成数据enqueue1的情况下会一直占用线程。所以,q另一端的dequeue部分必须是在另一个线程里运行,否则会造成整个程序的死锁。fs2的Queue类型款式是:Queue[F,A],所以我们必须用Stream.eval来对这个Queue进行函数式的操作:

val fs2Stream: Stream[Task,Int] = Stream.eval(async.boundedQueue[Task,Option[Int]](2)).flatMap { q =>
    //run Enumerator-Iteratee and enqueue data in thread 1
    //dequeue data and en-stream in thread 2(current thread)
  }

因为Stream.eval运算结果是Stream[Task,Int],所以我们可以得出这个flatMap内的函数款式 Queue[Task,Option[Int]] => Stream[Task,Int]。下面我们先考虑如何实现数据enqueue部分:这部分是通过Iteratee的运算过程产生的。我们提到过这部分必须在另一个线程里运行,所以可以用Task来选定另一线程如下:

    Task { Iteratee.flatten(enumerator |>> pushData(q)).run }.unsafeRunAsyncFuture()

现在这个Task就在后面另一个线程里自己去运算了。但它的运行进展则会依赖于另一个线程中dequeue数据的进展。我们先看看fs2提供的两个函数款式:

/** Repeatedly calls `dequeue1` forever. */
  def dequeue: Stream[F, A] = Stream.bracket(cancellableDequeue1)(d => Stream.eval(d._1), d => d._2).repeat

/**
   * Halts the input stream at the first `None`.
   *
   * @example {{{
   * scala> Stream[Pure, Option[Int]](Some(1), Some(2), None, Some(3), None).unNoneTerminate.toList
   * res0: List[Int] = List(1, 2)
   * }}}
   */
  def unNoneTerminate[F[_],I]: Pipe[F,Option[I],I] =
    _ repeatPull { _.receive {
      case (hd, tl) =>
        val out = Chunk.indexedSeq(hd.toVector.takeWhile { _.isDefined }.collect { case Some(i) => i })
        if (out.size == hd.size) Pull.output(out) as tl
        else if (out.isEmpty) Pull.done
        else Pull.output(out) >> Pull.done
    }}

刚好,dequeue产生Stream[F,A]。而unNoneTerminate可以根据Stream(None)来终止运算。现在我们可以把这个Reactive-Streams到fs2-pull-streams转换过程这样来定义:

implicit val strat = Strategy.fromFixedDaemonPool(4)
                                                  //> strat  : fs2.Strategy = Strategy
val fs2Stream: Stream[Task,Int] = Stream.eval(async.boundedQueue[Task,Option[Int]](2)).flatMap { q =>
  Task(Iteratee.flatten(enumNumbers |>> enqueueTofs2(q)).run).unsafeRunAsyncFuture
  pipe.unNoneTerminate(q.dequeue)
}   //> fs2Stream  : fs2.Stream[fs2.Task,Int] = attemptEval(Task).flatMap(<function1>).flatMap(<function1>)

现在这个stream应该已经变成fs2.Stream[Task,Int]了。我们可以用前面的log函数来试运行一下:

def log[A](prompt: String): Pipe[Task,A,A] =
    _.evalMap {row => Task.delay{ println(s"$prompt> $row"); row }}
                                                  //> log: [A](prompt: String)fs2.Pipe[fs2.Task,A,A]
    
fs2Stream.through(log("")).run.unsafeRun          //> > 1
                                                  //| > 2
                                                  //| > 3
                                                  //| > 4
                                                  //| > 5

我们成功的把Iteratee的Reactive-Stream转化成fs2的Pull-Model-Stream。

下面是这次讨论的源代码:

import play.api.libs.iteratee._
import scala.concurrent._
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.collection.mutable._
import fs2._
object iteratees {
def showElements: Iteratee[Int,Unit] = Cont {
  case Input.El(e) =>
     println(s"EL($e)")
     showElements
  case Input.Empty => showElements
  case Input.EOF =>
     println("EOF")
     Done((),Input.EOF)
}
val enumNumbers = Enumerator(1,2,3,4,5)

enumNumbers |>> showElements

Iteratee.flatten(enumNumbers |>> showElements).run


def enqueueTofs2(q: async.mutable.Queue[Task,Option[Int]]): Iteratee[Int,Unit] = Cont {
   case Input.EOF =>
       q.enqueue1(None).unsafeRun
       Done((),Input.EOF)
   case Input.Empty => enqueueTofs2(q)
   case Input.El(e) =>
       q.enqueue1(Some(e)).unsafeRun
       enqueueTofs2(q)
}
implicit val strat = Strategy.fromFixedDaemonPool(4)
val fs2Stream: Stream[Task,Int] = Stream.eval(async.boundedQueue[Task,Option[Int]](2)).flatMap { q =>
  Task(Iteratee.flatten(enumNumbers |>> enqueueTofs2(q)).run).unsafeRunAsyncFuture
  pipe.unNoneTerminate(q.dequeue)
}

def log[A](prompt: String): Pipe[Task,A,A] =
    _.evalMap {row => Task.delay{ println(s"$prompt> $row"); row }}
    
fs2Stream.through(log("")).run.unsafeRun
 
}

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Phoenix的Android之旅

两个数值相等的Integer不一定相等,为什么

昨天说到两个值是128的 Integer 对象 用 == 来比较的话结果是 false, 今天解释下为什么

903
来自专栏后端之路

Dubbo异常处理

小伙伴在使用dubbo过程中推出了疑问,部分自定义异常不见了,变成了RuntimeException。并且message超长。 如下图 我们来查看一下Dubb...

3065
来自专栏函数式编程语言及工具

Scalaz(53)- scalaz-stream: 程序运算器-application scenario

    从上面多篇的讨论中我们了解到scalaz-stream代表一串连续无穷的数据或者程序。对这个数据流的处理过程就是一个状态机器(state machine...

1789
来自专栏编舟记

命令式到函数式编程

应用场景:当我们用到 if-elseif-else 的时候,可以考虑使用 Optional 语义。 举例说明:

622
来自专栏wannshan(javaer,RPC)

dubbo 缓存的使用和实现解析

dubbo缓存主要实现,对方法调用结果的缓存。 在服务消费方和提供方都可以配置使用缓存。 以消费方为例,可以配置全局缓存策略,这样所有服务引用都启动缓存 ...

2696
来自专栏spring源码深度学习

java基础io流——配角也风流(不求甚解)

PrintStream是OutputStream的子类,PrintWriter是Writer的子类,两者处于对等的位置上,所以它们的API是非常相似的。

622
来自专栏函数式编程语言及工具

FunDA(5)- Reactive Streams:Play with Iteratees

    FunDA的设计目标就是把后台数据库中的数据搬到内存里,然后进行包括并行运算的数据处理,最后可能再对后台数据库进行更新。如果需要把数据搬到内存的话,那我...

18810
来自专栏函数式编程语言及工具

Scalaz(47)- scalaz-stream: 深入了解-Source

   scalaz-stream库的主要设计目标是实现函数式的I/O编程(functional I/O)。这样用户就能使用功能单一的基础I/O函数组合成为功能完...

2625
来自专栏玩转JavaEE

Spring Cloud中Hystrix的服务降级与异常处理

上篇文章我们看了自定义Hystrix请求命令的问题,使小伙伴们对Hystrix的使用有了进一步的了解,之前两篇文章都有涉及到一个叫做fallbackMethod...

3634
来自专栏wannshan(javaer,RPC)

dubbo通信消息解析过程分析(1)

由于rpc底层涉及网络编程接口,线程模型,网络数据结构,服务协议,细到字节的处理。牵涉内容较多,今天就先从一个点说起。 说说,dubbo通过netty框架做传...

3496

扫码关注云+社区