如何在 .mapParallelUnordered 中处理错误并使 Observable 保持活动状态

Question

我正在使用 Monix 3 并且有一种这样的代码：

  Observable.fromIterable(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9))
    .flatMap(i =>
      if (i % 2 == 0) {   // Bad i
        Observable.empty
      } else
        Observable.pure(i)
    )
    .foreachL(i => print(s"Good i: $i"))   /*Output: Good i: 1
                                                     Good i: 3
                                                     Good i: 5
                                                     Good i: 7
                                                     Good i: 9*/

这段代码运行良好，但我有很多长时间的 IO 操作，所以决定用 .mapParallelUnordered:

重构

  Observable.fromIterable(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9))
    .mapParallelOrdered(3)(i =>
      if (i % 2 == 0) {
        Task.raiseError(new Exception(s"Bad i: $i"))
      } else
        Task.pure(i)
    )
    .foreachL(i => print(s"Good i: $i"))    /*Output: Good i: 1*/

我试图获得与第一个示例相同的结果，但在并行处理中。问题是 Task.raiseError 杀死了整个 observable，所以它停止在 i = 2。

如何处理错误并使 Observable 保持活动状态？

Answer 1

你可以试试这个：

Observable
  .fromIterable(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9))
  .mapParallelOrdered(3)(
    i =>
      if (i % 2 == 0) {
        Task.pure(Left("Error"))
      } else
        Task.pure(Right(i))
  )
  .collect {
    case Right(i) => i
  }
  .foreachL(i => print(s"Good i: $i"))

如何在 .mapParallelUnordered 中处理错误并使 Observable 保持活动状态

How to handle errors and keep an Observable alive in .mapParallelUnordered

scala

observable

monix