Как регулировать фьючерсы с задержкой в ​​одну секунду с помощью Akka

Question

У меня есть список URI, каждый из которых я хочу запросить с задержкой в ​​одну секунду между ними.Как я могу это сделать?

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

Answer 1

akka streams имеет его из коробки с функцией throttle (учитывая, что вы используете akka-http и добавленный тег для потоков akka)

Answer 2

Вы можете создать Akka Streams Source из списка URI, затем throttle преобразование каждого URI в 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])

Answer 3

Что-то вроде этого perahps:

  @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