Scala未来运行顺序作业

Question

我试图顺序启动三个作业，但是当我尝试此代码时：

val jobs = Seq("stream.Job1","stream.Job2","stream.Job3")
    Future.sequence {
          jobs.map { jobClass =>
            Future {
              println(s"Starting the spark job from class $jobClass...")
              % gcloud("sparkC", "jobs", "submit", "spark", s"--cluster=$clusterName", s"--class=$jobClass", "--region=global", s"--jars=$JarFile")
              println(s"Starting the spark job from class $jobClass...DONE")
            }
          }
        }  

我并联三个作业，然后是顺序。我认为解决方案是使用flatMap，但我无法实施。
请任何帮助。

Answer 1

尝试此

val jobs = Seq("stream.Job1","stream.Job2","stream.Job3")
jobs.foldLeft(Future.successful[Unit]()) {
  case (result, jobClass) =>
    result.flatMap[Unit] {_ =>
      Future {
        println(s"Starting the spark job from class $jobClass...")
        % gcloud("sparkC", "jobs", "submit", "spark", s"--cluster=$clusterName", s"--class=$jobClass", "--region=global", s"--jars=$JarFile")
        println(s"Starting the spark job from class $jobClass...DONE")
      }
    }.
      recoverWith {
      case NonFatal(e) => result
    }
}

这将迭代您的工作，并在上一个完成后立即运行。我添加了recoverWith块以独立处理所有Futures，如果其中任何一个失败

Answer 2

如果作业不依赖彼此，并且如果您想拥有结果列表最后，您可以使用以下方式：

import scala.concurrent._
def runIndependentSequentially[X]
  (futs: List[() => Future[X]])
  (implicit ec: ExecutionContext): Future[List[X]] = futs match {
  case Nil => Future { Nil }
  case h :: t => for {
    x <- h()
    xs <- runIndependentSequentially(t)
  } yield x :: xs
}

现在，您可以在工作期货清单上使用它，如下所示：

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.language.postfixOps
val jobs = List("stream.Job1","stream.Job2","stream.Job3")
val futFactories = jobs.map { jobClass =>
  () => Future {
    println(s"Starting the spark job from class $jobClass...")
    Thread.sleep(5000)
    "result[" + jobClass + "," + (System.currentTimeMillis / 1000) % 3600 + "]"
  }
}
println(Await.result(runIndependentSequentially(futFactories), 30 seconds))

这会产生以下输出：

Starting the spark job from class stream.Job1...
Starting the spark job from class stream.Job2...
Starting the spark job from class stream.Job3...
List(result[stream.Job1,3011], result[stream.Job2,3016], result[stream.Job3,3021])

update ：由List[() => Future[X]]替换了期货列表，以便即使在论证传递给该论点之前，对期货的评估也不会开始 runIndependentSequentially方法。非常感谢@Evgeny指出它！