使用parMapN
,可以并行执行多个IO
,如下所示:
import cats.implicits._
import cats.effect.{ContextShift, IO}
import scala.concurrent.ExecutionContext
implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global)
val ioA = IO(for(i <- 1 to 100) { println(s"A$i"); Thread.sleep(100) })
val ioB = IO(for(i <- 1 to 100) { println(s"B$i"); Thread.sleep(100) })
val ioC = IO(for(i <- 1 to 100) { println(s"C$i"); Thread.sleep(100) })
val program = (ioA, ioB, ioC).parMapN { (_, _, _) => () }
program.unsafeRunSync()
示例输出:
A1
C1
B1
A2
C2
B2
A3
C3
B3
A4
C4
B4
A5
B5
C5
A6
B6
C6
A7
B7
C7
A8
...
根据documentation的说法,如果任何IO
完成失败,则取消未完成的任务。改变这种机制的最好方法是什么,以便所有的IO
都能完成?
在我的例子中,一些IO
不返回任何东西(IO[Unit]
),我仍然希望确保一切都运行到完成或遇到错误为止。
发布于 2019-03-13 06:31:46
嗯,我在发布我的问题后不久就找到了一个可能的答案。我不确定这是否是处理这个问题的最好方法,但是像这样定义我的IO
对我来说很有效:
val ioA = IO(for(i <- 1 to 100) { println(s"A$i"); Thread.sleep(100) }).attempt
val ioB = IO(for(i <- 1 to 100) { println(s"B$i"); Thread.sleep(100) }).attempt
val ioC = IO(for(i <- 1 to 100) { println(s"C$i"); Thread.sleep(100) }).attempt
发布于 2021-07-10 04:22:00
据我所知,您的示例代码没有引起任何错误。所以你应该有一个像下面这样的代码才能看到这个特性:
val ioA = IO(for(i <- 1 to 100) { println(s"A$i"); Thread.sleep(100) })
val ioB = IO.raiseError[Unit](new Exception("boom"))
val ioC = IO(for(i <- 1 to 100) { println(s"C$i"); Thread.sleep(100) })
而且它看起来不是很好,因为attempt
函数会将内部结构更改为IO[Either[_,_]]
,这似乎不是您想要的,不是吗?
https://stackoverflow.com/questions/55131466
复制相似问题