如何在Akka Stream中使用divertTo将Future失败转接到独立Sink？并获取错误发生阶段信息

要实现将flow1和flow2中产生的异常分流到errorSink，同时保留失败阶段的信息，我们需要先捕获每个阶段的失败并包装成带标识的错误对象，再通过divertTo实现分流。下面是具体步骤和代码：

1. 定义带阶段信息的自定义错误类型 #

首先创建一个自定义错误类，用来记录失败的阶段名称、出错的元素和原始异常，这样我们就能清晰定位是哪个环节出了问题：

// 自定义错误类型，携带阶段标识、出错元素和原异常
case class StageError(stageName: String, failedElement: Int, cause: Exception)

2. 修改Flow捕获失败并包装成Either类型 #

默认情况下，mapAsyncUnordered返回失败的Future会导致整个流终止，所以我们需要在每个Flow内部用recover捕获异常，将结果包装成Either[StageError, Int]——成功的元素用Right(x)，失败的用Left(StageError(...))：

import scala.concurrent.Future

val source = Source(1 to 10)

// 修正并修改flow1：捕获失败，包装成Either
val flow1 = Flow[Int].mapAsyncUnordered(2) { x =>
  if (x != 7) Future.successful(Right(x))
  else Future.failed(new Exception(s"x=$x has failed"))
    .recover { case ex => Left(StageError("flow1", x, ex)) }
}

// 同理修改flow2
val flow2 = Flow[Int].mapAsyncUnordered(2) { x =>
  if (x != 4) Future.successful(Right(x))
  else Future.failed(new Exception(s"x=$x has failed"))
    .recover { case ex => Left(StageError("flow2", x, ex)) }
}

3. 构建通用的错误分流Flow #

我们可以封装一个通用的分流Flow，负责将Left类型的错误元素发送到errorSink，同时将Right类型的成功元素继续传递到下游：

// 调整errorSink以接收自定义的StageError（保留完整错误信息）
val errorSink = Sink.fold(List[StageError])((errs, err: StageError) => err :: errs)

// 通用分流逻辑：将错误分流到errorSink，保留成功元素
def divertErrors(errorSink: Sink[StageError, _]): Flow[Either[StageError, Int], Int, _] =
  Flow[Either[StageError, Int]]
    .divertTo(
      // 提取Left中的错误并发送到errorSink
      Flow[Either[StageError, Int]].collect { case Left(err) => err }.to(errorSink),
      // 分流条件：仅处理Left类型的元素
      _.isLeft
    )
    // 留下成功的元素继续处理
    .collect { case Right(x) => x }

// 原有的结果收集sink
val sink = Sink.fold(List[Int])((xs, x: Int) => x :: xs)

4. 组装完整的流并获取结果 #

现在将所有组件组装起来，每个Flow之后都接入错误分流逻辑，确保每个阶段的异常都被捕获并分流。同时我们可以通过GraphDSL同时获取成功结果和错误列表：

import akka.actor.ActorSystem
import akka.stream.GraphDSL
import akka.stream.scaladsl.{Flow, Sink, Source, RunnableGraph}
import akka.stream.scaladsl.GraphDSL.Implicits._

implicit val system: ActorSystem = ActorSystem("ErrorDivertExample")
import system.dispatcher

// 用GraphDSL构建完整流，同时获取两个sink的结果
val graph = GraphDSL.create(sink, errorSink)((_, _)) { implicit builder => (mainSink, errSink) =>
  val splitAfterFlow1 = builder.add(Flow[Either[StageError, Int]]
    .divertTo(Flow[Either[StageError, Int]].collect { case Left(e) => e }.to(errSink), _.isLeft)
    .collect { case Right(x) => x })
  
  val splitAfterFlow2 = builder.add(Flow[Either[StageError, Int]]
    .divertTo(Flow[Either[StageError, Int]].collect { case Left(e) => e }.to(errSink), _.isLeft)
    .collect { case Right(x) => x })

  source ~> flow1 ~> splitAfterFlow1 ~> flow2 ~> splitAfterFlow2 ~> mainSink
  ClosedShape
}

val (mainResult, errorResult) = RunnableGraph.fromGraph(graph).run()

// 打印成功结果
mainResult.onComplete { res =>
  println(s"成功处理的元素（倒序）：$res")
}

// 打印错误详情，包含阶段信息
errorResult.onComplete { res =>
  println("\n捕获的错误详情：")
  res.foreach(_.foreach { err =>
    println(s"阶段[${err.stageName}]处理元素[${err.failedElement}]失败：${err.cause.getMessage}")
  })
}

关键说明 #

• 避免流终止：通过recover将失败的Future转换成Left类型的元素，确保流不会因为单个元素失败而终止，能继续处理其他元素。
• 阶段信息追踪：自定义的StageError类记录了失败的阶段名称、出错元素和原异常，让你能精准定位问题来源。
• divertTo的使用：divertTo接收一个目标Sink和一个分流条件，这里我们用_.isLeft判断是否为错误元素，将其分流到errorSink。

内容的提问来源于stack exchange，提问作者Tung Dang