我认为很难（或不可能）采用高级策略来实现它，主要是因为您想重试“ n”次（根据评论中的讨论），而且我认为您无法跟踪次数在使用监督时尝试了该元素。

我认为有两种方法可以解决此问题。可以将风险操作作为单独的流处理，也可以创建图表来进行错误处理。我将提出两个解决方案。

还要注意，Akka Streams区分错误和失败，因此，如果您不处理失败，它们将最终使流程崩溃（如果未引入任何策略），因此在下面的示例中，我将它们转换为Either，表示成功或错误。

分开的流

您可以做的是将每个字母视为一个单独的流，并使用重试策略分别处理每个字母的失败和一些延迟。

// this comes after your helloFormat

// note that the method is somehow simpler because it's
// using implicit dispatcher and scheduler from outside scope,
// you may also want to pass it as implicit arguments
def retry[T](f: => Future[T], delay: FiniteDuration, c: Int): Future[T] =
  f.recoverWith {
    // you may want to only handle certain exceptions here...
    case ex: Exception if c > 0 =>
      println(s"failed - will retry ${c - 1} more times")
      akka.pattern.after(delay, system.scheduler)(retry(f, delay, c - 1))
  }

val singleElementFlow = httpFlow.mapAsync[Hello](1) {
  case (Success(response), _) =>
    val f = Unmarshal(response).to[Hello]
    f.recoverWith {
      case ex: Exception =>
        // see https://github.com/akka/akka/issues/20192
        response.entity.dataBytes.runWith(Sink.ignore).flatMap(_ => f)
    }
  case (Failure(e), _) => Future.failed(e)
}

// so the searches can either go ok or not, for each letter, we will retry up to 3 times
val searches =
  Source('a' to 'z').map(search).mapAsync[Either[Throwable, Hello]](1) { elem =>
    println(s"trying $elem")
    retry(
      Source.single(elem).via(singleElementFlow).runWith(Sink.head[Hello]),
      1.seconds, 3
    ).map(ok => Right(ok)).recover { case ex => Left(ex) }
  }
// end

图形

此方法会将故障整合到图形中，并允许重试。此示例使所有请求并行运行，并希望重试失败的请求，但是，如果您不希望这种行为并逐个运行它们，那么我也可以做到。

// this comes after your helloFormat

// you may need to have your own class if you
// want to propagate failures for example, but we will use
// right value to keep track of how many times we have
// tried the request
type ParseResult = Either[(HttpRequest, Int), Hello]

def search(query: Char): (HttpRequest, (HttpRequest, Int)) = {
  val request = HttpRequest(uri = Uri("https://example.org").withQuery(Query("q" -> query.toString)))
  (request, (request, 0)) // let's use this opaque value to count how many times we tried to search
}

val g = GraphDSL.create() { implicit b =>
  import GraphDSL.Implicits._

  val searches = b.add(Flow[Char])

  val tryParse =
    Flow[(Try[HttpResponse], (HttpRequest, Int))].mapAsync[ParseResult](1) {
      case (Success(response), (req, tries)) =>
        println(s"trying parse response to $req for $tries")
        Unmarshal(response).to[Hello].
          map(h => Right(h)).
          recoverWith {
            case ex: Exception =>
              // see https://github.com/akka/akka/issues/20192
              response.entity.dataBytes.runWith(Sink.ignore).map { _ =>
                Left((req, tries + 1))
              }
          }
      case (Failure(e), _) => Future.failed(e)
    }

  val broadcast = b.add(Broadcast[ParseResult](2))

  val nonErrors = b.add(Flow[ParseResult].collect {
    case Right(x) => x
    // you may also handle here Lefts which do exceeded retries count
  })

  val errors = Flow[ParseResult].collect {
    case Left(x) if x._2 < 3 => (x._1, x)
  }
  val merge = b.add(MergePreferred[(HttpRequest, (HttpRequest, Int))](1, eagerComplete = true))

  // @formatter:off
  searches.map(search) ~> merge ~> httpFlow ~> tryParse ~> broadcast ~> nonErrors
                          merge.preferred <~ errors <~ broadcast
  // @formatter:on

  FlowShape(searches.in, nonErrors.out)
}

def main(args: Array[String]): Unit = {
  val source = Source('a' to 'z')
  val sink = Sink.seq[Hello]

  source.via(g).toMat(sink)(Keep.right).run().onComplete {
    case Success(seq) =>
      println(seq)
    case Failure(ex) =>
      println(ex)
  }

}

基本上，这里发生的事情是我们运行搜索httpFlow，然后尝试解析响应，然后广播结果并拆分错误和非错误，将非错误沉入接收器，并将错误发送回循环。如果重试次数超过计数，我们将忽略该元素，但您也可以使用该元素做其他事情。

无论如何，我希望这能给您一些想法。