当抛出异常时,是否有一种方法来获取正在映射的项?

不与 Supervision.Decide 但你可以用不同的方式来实现它。

查看此程序:

object Streams extends App{

  implicit val system = ActorSystem("test")

  implicit val mat = ActorMaterializer()

  val source = Source(List("1", "2", "3")).map { item =>
    Try {
      if (item == "2") {
        throw new RuntimeException("Error")
      } else {
        item
      }
    }
  }
  source
    .alsoTo(
      Flow[Try[String]]
        .filter(_.isFailure)
        .to(Sink.foreach(t => println("failure: " + t))))
    .to(
      Flow[Try[String]]
        .filter(_.isSuccess)
        .to(Sink.foreach(t => println("success " + t)))).run()

}

输出:

success Success(1)
failure: Failure(java.lang.RuntimeException: Error)
success Success(3)

这有点复杂,但您可以通过将映射函数包装在流中并使用 flatMapConcat 像这样:

Source(List(item1, item2, item3)).flatMapConcat { item =>
  Source(List(item))
    .map(mapF)
    .withAttributes(ActorAttributes.supervisionStrategy { e: Throwable =>
      logger.error("Exception in stream with itemId:" + item.id, e)
      Supervision.Resume
    })
}

def mapF(item: Item) =
  if (item.property < 0) {
    throw new RuntimeException("Error")
  } else {
    i
  }

这是可能的,因为每个流阶段都可以有自己的监督策略。

你可以使用 Supervision.Decider 记录这些属性。

object Test extends App {

  implicit val system = ActorSystem("test")

  implicit val mat = ActorMaterializer()

  val testSupervisionDecider: Supervision.Decider = {
    case ex: RuntimeException =>
      println(s"some run time exception ${ex.getMessage}")
      Supervision.Resume
    case ex: Exception =>
     //if you want to stop the stream
   Supervision.Stop
  }

  val source = Source(List("1", "2", "3")).map { item =>
    if (item == "2") {
      throw new RuntimeException(s"$item")
    } else {
      item
    }
  }

  source
    .to(Sink.foreach(println(_)))
    .withAttributes(ActorAttributes.supervisionStrategy(testSupervisionDecider))
    .run

}

输出是:

1
some run time exception 2
3