scala - FS2 Stream 异常处理不起作用

Question

我在 FS2 和异常处理方面遇到问题。我想要的是，给定一个 Stream[IO,A]，当我使用可以抛出异常的 f: A => B 映射到它时，我得到一个Stream[IO,Either[Throwable,B]]。

我尝试了以下方法，它按预期工作:

import cats.effect.IO
val x1 = fs2.Stream.emits(Vector(1,2,3,4)).covary[IO]
  .map(x => x * x)
  .map{ i => if(i == 9) throw new RuntimeException("I don't like 9s") else i}
  .attempt
x1.compile.toVector.unsafeRunSync().foreach(println)

它打印:

Right(1)
Right(4)
Left(java.lang.RuntimeException: I don't like 9s)

但是，当我尝试对该 Stream 执行任何操作时，我的问题就开始了。

val x1 = fs2.Stream.emits(Vector(1,2,3,4)).covary[IO]
  .map(x => x * x)
  .map{ i => if(i == 9) throw new RuntimeException("I don't like 9s") else i}
  .attempt.map(identity)

x1.compile.toVector.unsafeRunSync().foreach(println)

抛出异常并杀死应用程序:

java.lang.RuntimeException: I don't like 9s
    at swaps.fm.A$A32$A$A32.$anonfun$x1$2(tmp2.sc:7)
    at scala.runtime.java8.JFunction1$mcII$sp.apply(tmp2.sc:8)
    ...

更奇怪的是，使用 take 让 Stream 只返回我认为没问题的元素，仍然以同样的方式爆炸:

val x1 = fs2.Stream.emits(Vector(1,2,3,4)).covary[IO]
  .map(x => x * x)
  .map{ i => if(i == 9) throw new RuntimeException("I don't like 9s") else i}
  .attempt.take(2)

x1.compile.toVector.unsafeRunSync().foreach(println)

有人能解释一下为什么会这样吗？这是错误还是(非)预期的行为？

注意 此行为存在于 FS2 0.10.0-M7 和 0.10.0

Answer 1

这里的问题是，要使用fs2，您必须编写纯代码。抛出异常不是纯粹的，因此如果您希望管道中的某个步骤可能失败，则需要将其明确化。这里有两种方式:

import cats.effect.IO
val x1 = fs2.Stream.emits(Vector(1,2,3,4)).covary[IO]
  .map(x => x * x)
  .map{ i => if(i == 9) Left[Throwable, Int](new RuntimeException("I don't like 9s")) else Right(i)}
x1.compile.toVector.unsafeRunSync().foreach(println)
// Explicit Left annotation is so you can .rethrow if desired; it can be omitted or added later with .widen

或

import cats.effect.IO
val x1 = fs2.Stream.emits(Vector(1,2,3,4)).covary[IO]
  .map(x => x * x)
  .flatMap { i => if(i == 9) Stream.raiseError(new RuntimeException("I don't like 9s")) else Stream.emit(i) }
  .attempt
x1.compile.toVector.unsafeRunSync().foreach(println)

其中第一个是可取的，因为 flatMap 和 emit 将导致 size-1 block ，通常效率较低。如果您想在出现第一个错误时停止处理，请将 .rethrow 添加到流的末尾。