Akka流媒体错误处理.如何知道哪一行失败



我在akka streams错误处理

上阅读了本文

http://doc.akka.io/docs/akka/2.5.4/scala/stream/stream/stream-erream-error.html

并编写了此代码。

val decider: Supervision.Decider = {
  case _: Exception => Supervision.Restart
  case _ => Supervision.Stop
}
implicit val actorSystem = ActorSystem()
implicit val actorMaterializer = ActorMaterializer(ActorMaterializerSettings(actorSystem).withSupervisionStrategy(decider))
val source = Source(1 to 10)
 val flow = Flow[Int].map{x => if (x != 9) 2 * x else throw new Exception("9!")}
 val sink : Sink[Int, Future[Done]] = Sink.foreach[Int](x => println(x))
 val graph = RunnableGraph.fromGraph(GraphDSL.create(sink){implicit builder => s =>
  import GraphDSL.Implicits._
  source ~> flow ~> s.in
  ClosedShape
})
val future = graph.run()
future.onComplete{ _ =>
  actorSystem.terminate()
}
Await.result(actorSystem.whenTerminated, Duration.Inf)

这很好。...除了我需要扫描输出以查看哪一行没有被处理。我有没有办法打印/记录失败的行?[不用明确的尝试/捕获块中的每个流程?]

因此,例如,如果我正在使用演员(而不是流),我本可以写一个演员的生命周期事件,当演员重新启动时,我可以记录下来,以及在重新启动时正在处理的消息。

,但在这里,我没有明确使用演员(尽管它们是内部使用的)。流量/源/水槽是否有生命周期事件?

只是对您的代码的小修改:

  val decider: Supervision.Decider = {
  case e: Exception =>
    println("Exception handled, recovering stream:" + e.getMessage)
    Supervision.Restart
  case _ => Supervision.Stop
}

如果您将有意义的消息传递给流中的异常,例如,您可以在监督决定者中打印它们。

我使用println给出了一个快速,简短的答案,但强烈建议使用一些记录库,例如scala-logging

相关内容

  • 没有找到相关文章

最新更新