scala - 如何防止 Future 永不完成

Question

假设我有多个要并行运行的任务。

每个任务(方法)都有一个内部递归函数，基本上从数据库中获取数据并将其保存在某些数据存储中。

[简化的内递归函数]

 def simplifiedSomeTask(): Unit = {
    @scala.annotation.tailrec
    def get(
        stream: Stream[SomeEntity],
        result: Seq[SomeEntity],
    ): Stream[SomeEntity] = result match {
      case Nil =>
        stream
      case _ =>
        val query = //query to fetch data from database
        get(
          stream append result.toStream,
          query.run.value, // get fetched data from database
        )
    }

    val buffer = collection.mutable.Map.empty[String, String]

    get(
      Stream.empty,
      query.run.value
    ).foreach { r =>
      buffer.put(r.loginId, r.userId)
    }
  }

当尝试运行 A 时，Future 由于某种原因永远不会完成。

[一]

val f1 =Future { someTask1() }
val f2 =Future { someTask2() }
val f3 =Future { someTask3() }

val f = for {
  _ <- f1 
  _ <- f2 
  _ <- f3 
} yield ()

Await.result(f, Duration.Inf)

但是，B 可以工作(尽管它不并行运行)

[B]

val f = for {
  _ <- Future { someTask1() }
  _ <- Future { someTask2() }
  _ <- Future { someTask3() }
} yield ()

Await.result(f, Duration.Inf)

我应该如何修改 A 使​​其按预期运行？

Answer 1

我无法重现您的问题，但奇怪行为的原因可能是第一个示例中的语法不完全正确。您应该编写您的第一个理解部分，例如:

val f = for {
  _ <- f1
  _ <- f2
  _ <- f3
} yield ()

但是 for 理解是按顺序进行的，并且您的 future 在第一个示例中并行运行的唯一原因是 Future 急切地启动(“Future 现在开始”)。

如果您想确保 Futures 将并行执行，请使用 Future.sequence:

val f = Future.sequence(
  List(
    Future { someTask1() },
    Future { someTask2() },
    Future { someTask3() }
  )
)