卡在斯卡拉的未来

Question

基本上，我在cassandra上运行两个期货查询，然后我需要做一些计算并返回值（值的平均值）。

这是我的代码：

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-当卡桑德拉只有几个像50个的人需要寻找时，它就会运行并给我结果

2 - 当有很多像1000这样的人时，它给了我

java.lang.ArithmeticException：/by零

编辑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个，抛出算术异常：/by零

似乎在返回未来之前，它无法计算出一切......

Answer 1

由于您必须调用一个在参数序列上返回Future的函数，因此最好使用Future.traverse。

例如：

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

Answer 2

您需要等到findSkusByProductId生成的所有期货完成后，才能计算平均值。因此，将所有这些期货累积在序列中，调用期货。在其上排序，以获得一个未来[Seq]值，然后将该未来值映射到一个计算平均值的函数。然后将替换为productsByCategory。地图，带有平面图。