为什么阿卡流吞噬我的例外？

Question

为什么会有例外

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source

object TestExceptionHandling {
  def main(args: Array[String]): Unit = {
    implicit val actorSystem = ActorSystem()
    implicit val materializer = ActorMaterializer()(defaultActorSystem)

    Source(List(1, 2, 3)).map { i =>
      if (i == 2) {
        throw new RuntimeException("Please, don't swallow me!")
      } else {
        i
      }
    }.runForeach { i =>
      println(s"Received $i")
    }
  }
}

静默忽略？我可以看到流在打印收到1后停止，但没有记录任何内容。请注意，问题不是一般的日志配置，因为如果我在我的application.conf文件中设置akka.log-config-on-start=on，我会看到很多输出。

Answer 1

我对Akka Streams吞下我的异常有不同的问题。我将在这里发布它，因为这是Google的顶级结果。

在这样的情况下，其中source是Source[ByteString， Any]：

source.runWith(StreamConverters.fromOutputStream(() => outputStream))

这将返回一个未来[IOResult]。如果对输出流的写入失败（例如，源失败），那么未来仍将返回成功。在这种情况下，您实际上必须检查IOResult是否存在错误：

source.runWith(StreamConverters.fromOutputStream(() => output)).
      map(ior => {
        if (!ior.wasSuccessful) 
          throw new RuntimeException(ior.getError)
      })

这样做的结果将是一个失败的未来，只有正确的例外。

Answer 2

当我开始使用akk streams时，我也有类似的问题<代码>监督。Decider有帮助，但并不总是如此。

不幸的是，它没有捕获ActionPublisher中抛出的异常。我看到它被处理了，ActorPublisher。调用了onError，但它没有到达监督。决策者。它使用文档中提供的简单流。

如果我使用Sink.actorRef，错误也不会到达参与者。

为了实验，我尝试了以下样本

val stream = Source(0 to 5).map(100 / _)
stream.runWith(Sink.actorSubscriber(props))

在这种情况下，异常被决策者捕获，但从未到达参与者订阅者。

总的来说，我认为这是不一致的行为。我不能使用一种机制来处理流中的错误。

我最初的SO问题：自定义监督。决定器不捕获ActorPublisher产生的异常

这里是akka问题，它被跟踪：https://github.com/akka/akka/issues/18359

Answer 3

我现在正在使用一个自定义的监控。Decider确保正确记录异常，可以这样设置：

val decider: Supervision.Decider = { e =>
  logger.error("Unhandled exception in stream", e)
  Supervision.Stop
}

implicit val actorSystem = ActorSystem()
val materializerSettings = ActorMaterializerSettings(actorSystem).withSupervisionStrategy(decider)
implicit val materializer = ActorMaterializer(materializerSettings)(actorSystem)

此外，正如Vikor Klang所指出的，在上面给出的示例中，也可以通过

Source(List(1, 2, 3)).map { i =>
  if (i == 2) {
    throw new RuntimeException("Please, don't swallow me!")
  } else {
    i
  }
}.runForeach { i =>
  println(s"Received $i")
}.onComplete {
  case Success(_) =>
    println("Done")
  case Failure(e) =>
    println(s"Failed with $e")
}

然而，请注意，这种方法不会帮助您解决问题

Source(List(1, 2, 3)).map { i =>
  if (i == 2) {
    throw new RuntimeException("Please, don't swallow me!")
  } else {
    i
  }
}.to(Sink.foreach { i =>
  println(s"Received $i")
}).run()

由于run（）返回Unit。