通过维护我自己的内部缓冲区以某种方式切换到基于“推”的模型,这样我就不会拉两次?
是的。例如,您可以使用 fs2 中的队列:
def splitStream[F[_], A](stream: Stream[F, A], split: A => B): F[(Stream[F, A], Stream[F, B])] =
for {
q <- Queue.noneTerminated[F, A]
} yield (stream.evalTap(a => q.enqueue1(Some(a)).onFinalize(q.enqueue1(None)), q.dequeue.map(split))
当然,这里的问题是,如果调用者忽略任何一个流,另一个将死锁并且永远不会发出任何东西。这通常是您在尝试将一个流分成多个流时遇到的问题,并且无论何时订阅,都保证出现在每个子流中的值。
我通常采用的解决方案是结合更大的操作并使用诸如broadcast
or之类的运算符parJoin
:
def splitAndRun[F[_]: Concurrent, A](
base: Stream[F, A],
runSeveralThings: List[Stream[F, A] => Stream[F, Unit]]
): F[Unit] =
base.broadcastTo(run: _*).compile.drain
在这里,您知道您将拥有多少消费者,因此首先不会有忽略的流。