如何使用 Akka 以一秒的延迟限制 Futures答案

【问题标题】：How to throttle Futures with one-second delay with Akka如何使用 Akka 以一秒的延迟限制 Futures
【发布时间】：2018-12-18 12:45:59
【问题描述】：

我有一个 URI 列表，我想在请求之间延迟一秒。我该怎么做？

val uris: List[String] = List()
// How to make these URIs resolve 1 second apart?
val responses: List[Future[Response]] = uris.map(httpRequest(_))

【问题讨论】：

这看起来像 Java 代码，而不是 Scala...您在 Scala 中使用 val 或 var 声明变量。

标签： scala akka akka-stream akka-http

【解决方案1】：

您可以从 URI 列表中创建一个 Akka Streams Source，然后将每个 URI 转换为 throttle Future[Response]：

def httpRequest(uri: String): Future[Response] = ???

val uris: List[String] = ???

val responses: Future[Seq[Response]] =
  Source(uris)
    .throttle(1, 1 second)
    .mapAsync(parallelism = 1)(httpRequest)
    .runWith(Sink.seq[Response])

【讨论】：

【解决方案2】：

可能是这样的：

  @tailrec
  def withDelay(
    uris: Seq[String], 
    delay: Duration = 1 second, 
    result: List[Future[Response]] = Nil,
  ): Seq[Future[Response]] = uris match {
     case Seq() => result.reversed
     case (head, tail@_*) => 
        val v = result.headOption.getOrElse(Future.successful(null))
          .flatMap { _ => 
            akka.pattern.after(delay, context.system.scheduler)(httpRequest(head))
          }
        withDelay(tail, delay, v :: result)
   }

这在第一次执行之前也有延迟，但我希望，如果有必要如何摆脱它已经足够清楚了...... 另一个警告是，这假设所有期货都成功。一旦失败，所有后续处理都将中止。如果您需要不同的行为，您可能需要将 .flatMap 替换为 .transform 或添加 .recover 等。

如果愿意，您也可以使用.foldLeft 编写相同的内容：

  uris.foldLeft(List.empty[Future[Response]]) { case (results, next) => 
    results.headOption.getOrElse(Future.successful(null))
      .flatMap { _ => 
        akka.pattern.after(delay, context.system.scheduler)(httpRequest(next))
      } :: results
  }.reversed

【讨论】：

这不是将每个未来延迟 1 秒，而不是一个接一个地延迟吗？比如第一个延迟1秒，第二个延迟2秒，第三个延迟3秒
在执行每个未来之前会增加 1 秒的延迟。因此，实际上，是的，第 N 次执行总共延迟了 N 秒。这不是你想要的吗？

【解决方案3】：

akka 流有它out of the box with the throttle function（考虑到您正在使用 akka-http 并为 akka 流添加了标签）

【讨论】：