scala - 被 scala 中的 future 困住了

Question

基本上，我在 cassandra 上运行两个 future 查询，然后我需要进行一些计算并返回值(值的平均值)。

这是我的代码:

object TestWrapFuture {
  def main(args: Array[String]) {
    val category = 5392
    ExtensiveComputation.average(category).onComplete {
      case Success(s) => println(s)
      case Failure(f) => throw new Exception(f)
    }
  }
}

class ExtensiveComputation {

  val volume = new ListBuffer[Int]()

  def average(categoryId: Int): Future[Double] = {

    val productsByCategory = Product.findProductsByCategory(categoryId)

    productsByCategory.map { prods =>
      for (prod <- prods if prod._2) {
        Sku.findSkusByProductId(prod._1).map { skus =>
          skus.foreach(sku => volume += (sku.height.get * sku.width.get * sku.length.get))
        }
      }

      val average = volume.sum / volume.length
      average
    }
  }
}

object ExtensiveComputation extends ExtensiveComputation

那么问题出在哪里呢？

skus.foreach 将结果值附加到 ListBuffer 中。由于一切都是异步的，当我尝试在主程序中获取结果时，我收到一条错误消息，提示我无法除以零。

事实上，由于我的 Sku.findSkusByProduct 返回一个 Future，当我尝试计算平均值时，卷是空的。

我应该在计算之前阻止任何事情，还是应该做其他事情？

编辑

好吧，我尝试像这样阻止:

  val volume = new ListBuffer[Int]()

  def average(categoryId: Int): Future[Double] = {

    val productsByCategory = Product.findProductsByCategory(categoryId)

    val blocked = productsByCategory.map { prods =>
      for (prod <- prods if prod._2) {
        Sku.findSkusByProductId(prod._1).map { skus =>
          skus.foreach(sku => volume += (sku.height.get * sku.width.get * sku.length.get))
        }
      }
    }

    Await.result(blocked, Duration.Inf)
    val average = volume.sum / volume.length
    Future.successful(average)
  }

然后我从这段代码中得到了两个不同的结果:

    Sku.findSkusByProductId(prod._1).map { skus =>
          skus.foreach(sku => volume += (sku.height.get * sku.width.get * sku.length.get))
        }

1 - 当 cassandra 上只有 50 个这样的数据时，它会运行并给出结果

2 - 当有很多像 1000 时，它给了我

java.lang.ArithmeticException: / by zero

编辑2

我尝试了@Olivier Michallat 提议的代码

   def average(categoryId: Int): Future[Double] = {

    val productsByCategory = Product.findProductsByCategory(categoryId)

    productsByCategory.map { prods =>
      for (prod <- prods if prod._2) findBlocking(prod._1)
      volume.sum / volume.length
    }
  }

  def findBlocking(productId: Long) = {
    val future = Sku.findSkusByProductId(productId).map { skus =>
      skus.foreach(sku => volume += (sku.height.get * sku.width.get * sku.length.get))
    }

    Await.result(future, Duration.Inf)
  }

以及 @kolmar 提议的以下内容:

   def average(categoryId: Int): Future[Int] = {
    for {
      prods <- Product.findProductsByCategory(categoryId)
      filtered = prods.filter(_._2)
      skus <- Future.traverse(filtered)(p => Sku.findSkusByProductId(p._1))
    } yield {
      val volumes = skus.flatten.map(sku => sku.height.get * sku.width.get * sku.length.get)
      volumes.sum / volumes.size
    }
  }

两者都可以在少数 sku 上找到 50 左右，但两者都因许多 sku 找不到 1000 左右而失败，并抛出 ArithmeticException:/by Zero

似乎它无法在返回 future 之前计算出所有内容......

Answer 1

您需要等到 findSkusByProductId 生成的所有 future 都完成后才能计算平均值。因此，将所有这些 future 累积在 Seq 中，对其调用 Future.sequence 以获得 Future[Seq]，然后将该 future 映射到计算平均值的函数。然后将 productsByCategory.map 替换为 flatMap。