Scala使用多个Web服务和合并输出的惯用和最快方式

Question

我有一个Scala应用程序，它使用多个数据源并合并组合输出，该应用程序并行调用多个HTTP调用并等待它们的结果，当所有问题都解决后，它将它们的输出合并到一个结果对象中，并将响应作为Scala Future发送回来。代码:

val providers = getProviders()
val futures = for (p <- providers) yield searchSource(p, req)
val result = waitAll(futures)
val emptyList = Option(ProductList(Vector.empty, ListSummary(0, 0, 0)))

result.map { x =>
  x.foldLeft(emptyList)((r, c) => {
    if (c.isSuccess) ProductList.merge(r, c.get) else r
  })
}

它首先获取提供程序字符串的列表，然后在 searchSource 中并行触发对每个提供程序的调用。

searchSource 函数本身返回 Future[Option[ProductList]]

该方法等待所有未来完成，为此调用waitAll函数：

  protected def waitAll[T](futures: Seq[Future[T]])(implicit ec: CustomExecutionContext): Future[Seq[Try[T]]] =
    Future.sequence(lift(futures))

  protected def lift[T](futures: Seq[Future[T]])(implicit ec: CustomExecutionContext): Seq[Future[Try[T]]] =
    futures.map(_.map { Success(_) }.recover { case t => Failure(t) })

在等待所有未来完成后，它将结果聚合到单个复合对象中，并将其作为未来返回。为此，我使用了foldLeft函数。

是否有任何“官方”的一元/惯用方法来解决此类问题，或者是否有此类用例的性能优化示例？

Answer 1

当然这是一种常见的模式，使用猫效应的IO，而不是未来确实有帮助（还有其他选择）。例如，您可以实现如下内容（为了清晰起见，我对类型进行了明确说明）：

def searchSource(provider: String, request: Request[_]): IO[Option[ProductList]] = ???

val result: IO[List[Option[ProductList]]] = 
  providers.traverse[IO, Option[ProductList]](p => searchSource(p, req))

请注意，遍历本质上与映射后跟序列相同。然后，您可以在结果IO程序中执行任何操作（例如，映射它等），以将其转换为所需的最终结果。在程序结束时，您可以执行以下操作：

result.attempt.unsafeRunSync() //Either[Throwable, List[Option[ProductList]]]

有关错误处理，请查看 IO 的句柄错误使用方法。您可以在 searchSource 级别和/或结果级别处理错误，因为它们都是 IO。