将 scalaz-stream 进程拆分为两个子流答案

【问题标题】：Splitting a scalaz-stream process into two child streams将 scalaz-stream 进程拆分为两个子流
【发布时间】：2015-02-15 19:25:57
【问题描述】：

使用 scalaz-stream 是否可以拆分/分叉然后重新加入流？

例如，假设我有以下功能

val streamOfNumbers : Process[Task,Int] = Process.emitAll(1 to 10)

val sumOfEvenNumbers = streamOfNumbers.filter(isEven).fold(0)(add)
val sumOfOddNumbers  = streamOfNumbers.filter(isOdd).fold(0)(add)

zip( sumOfEven, sumOfOdd ).to( someEffectfulFunction )

使用 scalaz-stream，在此示例中，结果将如您所料 - 将 1 到 10 的数字元组传递到接收器。

但是，如果我们将streamOfNumbers 替换为需要 IO 的东西，它实际上会执行两次 IO 操作。

使用 Topic 我可以创建一个 pub/sub 进程来正确复制流中的元素，但它不会缓冲 - 它只是尽可能快地使用整个源，而不管速度接收器消耗它。

我可以将它包装在一个有界队列中，但最终结果感觉比它需要的复杂得多。

有没有一种更简单的方法可以在 scalaz-stream 中拆分流，而无需从源中重复 IO 操作？

【问题讨论】：

标签： scala stream scalaz scalaz-stream

【解决方案1】：

您也许仍然可以使用主题，并确保在您推送主题之前子进程将订阅。

但是请注意，这个解决方案没有任何限制，即如果你推得太快，你可能会遇到 OOM 错误。

def split[A](source:Process[Task,A]): Process[Task,(Process[Task,A], Proces[Task,A])]] = {
  val topic = async.topic[A]

  val sub1 = topic.subscribe
  val sub2 = topic.subscribe

  merge.mergeN(Process(emit(sub1->sub2),(source to topic.publish).drain))
}

【讨论】：

对于动态分割数，返回类型 Process[Task, Array[Process[Task,A]]] 是否有意义？另外，我如何修改它以使用 Queue 而不是 Topic 并为每个子进程提供唯一的流元素？感谢您的宝贵时间

【解决方案2】：

还要用“拆分”要求来澄清先前的答案 delas。您的特定问题的解决方案可能不需要拆分流：

val streamOfNumbers : Process[Task,Int] = Process.emitAll(1 to 10)
val oddOrEven: Process[Task,Int\/Int] = streamOfNumbers.map {
   case even if even % 2 == 0 => right(even)
   case odd => left(odd)
} 
val summed = oddOrEven.pipeW(sump1).pipeO(sump1)

val evenSink: Sink[Task,Int] = ???
val oddSink: Sink[Task,Int] = ???

summed
.drainW(evenSink)
.to(oddSink)

【讨论】：

我在您的回答中看到了一个通用组合器：def partition[A](p: A => Boolean): Process1[A, A \/ A] = process1.lift(a => if (p(a)) right(a) else left(a))。应该加吗？
Frank 我认为对于这些用例，我们缺少一种更有效的partialTee。但是，是的，你的提议对我来说似乎很好。我想看看像 observeIf[A,B](pf:PartialFunction[A,B])(Sink[T,B]) 这样的东西。
谢谢。我的真实场景要复杂一些，因此无法以这种方式简化，但这给了我更多线程可以解决问题。

【解决方案3】：

我同样需要这个功能。我的情况有点棘手，不允许我以这种方式解决它。

感谢 Daniel Spiewak 在this thread 中的回复，我能够让以下工作正常进行。我通过添加onHalt 改进了他的解决方案，因此一旦Process 完成，我的应用程序就会退出。

def split[A](p: Process[Task, A], limit: Int = 10): Process[Task, (Process[Task, A], Process[Task, A])] = {
  val left = async.boundedQueue[A](limit)
  val right = async.boundedQueue[A](limit)

  val enqueue = p.observe(left.enqueue).observe(right.enqueue).drain.onHalt { cause =>
    Process.await(Task.gatherUnordered(Seq(left.close, right.close))){ _ => Halt(cause) }
  }
  val dequeue = Process((left.dequeue, right.dequeue))

  enqueue merge dequeue
}

【讨论】：