Блог

Следующий код должен выполнять putStrLn эффект параллельно из-за mapMPar :

 val runtime = zio.Runtime.default
val foo = ZIO.sleep(5.second) *> ZIO("foo")
val bar = ZIO("bar")

val k = ZStream.fromEffect(foo)    ZStream.fromEffect(bar)
val r = k.mapMPar(3)(x => console.putStrLn(s"Processing `${x}`"))

runtime.unsafeRun(r.runDrain)
  

Но на самом деле это всегда происходит foo раньше bar , несмотря ни на что. Я что-то пропустил или это ошибка?

Я думаю, что ваш пример просто не делает того, что вы ожидаете от него. fromEffect создает поток, который в основном говорит «У меня есть эффект, который в конечном итоге сгенерирует один элемент», затем первый поток ожидает 5 секунд, прежде чем создать этот элемент. Из-за природы потока оператор or concat является ленивым, что означает, что он не может начать обработку, пока все элементы не будут использованы из первого потока (что не может произойти в течение 5 секунд). В результате ваш поток действительно выглядит следующим образом:

 --5s--(foo)(bar)|
  

вместо того, что, я полагаю, вы думаете, что это должно понравиться:

 (bar)--5s--(foo)|
  

Возможно, лучший способ подумать об этом заключается в том, что для большей части потока у вас есть однополосная магистраль, одновременно может перемещаться только один элемент, а все последующие элементы блокируются элементами в начале строки. Как только вы нажмете на этот Par блок, вы открываете доступ к нескольким полосам движения, что означает, что более быстрые движущиеся объекты потенциально могут обогнать.

Таким образом, я могу добиться желаемого поведения, выполнив вместо этого что-то вроде этого:

 val k = ZStream("foo", "bar")
val r = k.mapMPar(3)(x => putStrLn(s"$x:enter") *> (ZIO.sleep(5.second) *> putStrLn(s"Processing `${x}`")) <* putStrLn(s"$x:exit"))

r.runDrain
  

Или написан немного более компактно:

 ZStream("foo", "bar").mapMPar(3)(x => for {
  _ <- putStrLn(s"$x:enter")
  _ <- ZIO.sleep(5.seconds) *> putStrLn(s"Processing `$x`")
  _ <- putStrLn(s"$x:exit")
} yield ()).runDrain