scala - Akka 流(Scala): Filtering out exceptions

Question

我的 akka 流管道中的步骤之一是在接收到无效输入时抛出异常的转换。我想放弃这些有问题的输入。因此，我想出了以下解决方案:

...
.map( input => Try( transformation( input ) ).toOption )
.filter( _.nonEmpty )
.map( _.get )
...

实际上，这只是一个 flatMap，需要 3 个步骤。

有更直接的 akka 方法吗？

Answer 1

您可以使用监督策略。摘自文档:

val decider: Supervision.Decider = {
  case _: ArithmeticException => Supervision.Resume
  case _                      => Supervision.Stop
}

val flow = Flow[Int]
  .filter(100 / _ < 50)
  .map(elem => 100 / (5 - elem))
  .withAttributes(ActorAttributes.supervisionStrategy(decider))

您可以配置决策器来执行您需要的任何操作。如果您需要为所有异常跳过该元素，请使用

 case _: Throwable => Supervision.Resume

查看 https://doc.akka.io/docs/akka/current/stream/stream-error.html