scala - 如何在 .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"))