如何在处理速率限制时异步发送HTTP请求

Question

免责声明：我是sttp和Monix的新手，这是我尝试了解更多关于这些库的信息。我的目标是通过HTTP GET请求从给定的API获取数据(客户端(->解析JSON响应->将这些信息写入数据库。我的问题只涉及第一部分。我的目标是以异步(希望是快速(的方式运行get请求，同时能够避免或处理速率限制。

下面是我已经尝试过的一个片段，似乎适用于一个请求：

package com.github.client
import io.circe.{Decoder, HCursor}
import sttp.client._
import sttp.client.circe._
import sttp.client.asynchttpclient.monix._
import monix.eval.Task
object SO extends App {
case class Bla(paging: Int)
implicit val dataDecoder: Decoder[Bla] = (hCursor: HCursor) => {
for {
next_page <- hCursor.downField("foo").downArray.downField("bar").as[Int]
} yield Bla(next_page)
}
val postTask = AsyncHttpClientMonixBackend().flatMap { implicit backend =>
val r = basicRequest
.get(uri"https://foo.bar.io/v1/baz")
.header("accept", "application/json")
.header("Authorization", "hushh!")
.response(asJson[Bla])
r.send() // How can I instead of operating on a single request, operate on multiple
.flatMap { response =>
Task(response.body)
}
.guarantee(backend.close())
} 
import monix.execution.Scheduler.Implicits.global
postTask.runSyncUnsafe() match {
case Left(error) => println(s"Error when executing request: $error")
case Right(data) => println(data)
}
}

我的问题：

1. 如何通过使用Monix操作多个GET请求(而不是单个请求(，同时保持代码异步和可组合
2. 如何避免或处理api服务器施加的速率限制

顺便说一句，如果能支持限速目标，我在使用另一个后端方面也很灵活。

Answer 1

你可以像这个一样使用monix.reactive.Observable

Observable.repeatEval(postTask) // we generate infinite observable of the task
.throttle(1.second, 3) // set throttling
.mapParallelOrderedF(2)(_.runToFuture) // set execution parallelism and execute tasks
.subscribe() // start the pipline


while (true) {}