scala - Akka 和 future 的同时请求太多

Question

我的情况是这样的。

我有一个包含大量报告的向量，需要使用 rest api 调用发送这些报告。

我正在使用 Futures.traverse(1 中提到的向量)

由于向量太大，它会因超出最大打开请求而失败。

我能想到的一个初始解决方案是增加 max-open-requests 设置。但这里的问题是我不知道需要事先发送的报告数量。

有人可以提出替代解决方案，例如限制通过 Futures.traverse 发生的并行性

Answer 1

因为你用 akka 标记了这个问题，我假设您使用 akka-http 进行调用。您可以使用 akka-streams 批量发出请求，以避免连接溢出，例如:

Source(reportsVector)
  .grouped(safeValue)
  .mapAsync(1)(reps => Future.traverse(reps)(x => ...)) //do your stuff
  .mapConcat(identity)
  .runWith(Sink.seq)

该示例将执行 safeValue一次并发调用并将所有结果收集到一个集合中，该集合将在整个流完成时返回。您还可以与其他运营商一起玩，例如 sliding和 splitWhen为了使其更适合您的用例，您可以调整 safeValue和 mapAsync并发值也是如此。请注意，此流的来源是一个已知向量 ( reportsVector )，但它也可能是未知的有限报告流。