Akka流Http singleRequest在等待响应时阻塞整个流答案

【问题标题】：Akka stream Http singleRequest blocks whole stream while waiting for responseAkka流Http singleRequest在等待响应时阻塞整个流
【发布时间】：2020-09-07 21:17:54
【问题描述】：

我正在尝试将 Akka Http 集成到我的 Akka 流中，但在极少数情况下，流会卡住。

  implicit val system: ActorSystem = ActorSystem("actor-system")

  Source(0 to 10)
    .mapAsync(10)(i ⇒ {
      val url =
        if (i == 1) "http://run.mocky.io/v3/40ff086f-1389-4ca5-ace8-1f0b3ac75582?mocky-delay=10s"
        else "http://google.com"

      Http().singleRequest(HttpRequest(uri = url))
    })
    .runForeach(r ⇒ println(s"${System.currentTimeMillis()}: ${r._1}"))

此代码将在第一次输出 1599480226827: 301 Moved Permanently 后卡住 10 秒，然后将同时刷新其余代码。

输出将是：

1599480226827: 301 Moved Permanently
1599480236826: 302 Found
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently
1599480236826: 301 Moved Permanently

我希望它能按顺序输出所有内容，除了延迟的。

为什么我的流被这样的请求阻止了？又该如何避免呢？

【问题讨论】：

标签： akka akka-stream akka-http

【解决方案1】：

来自mapAsync的scaladocs

并行运行的 Future 数量作为 mapAsync 的第一个参数给出。这些 Future 可以按任意顺序完成，但下游发出的元素与从上游接收的顺序相同。

您的请求是并行发送的，但runForeach 下的函数以特定顺序调用，导致输出结果出现延迟。它正在等待第二个响应可用。

您可以使用mapAsyncUnordered 来处理可用的回复。

【讨论】：