scala – 如何并行执行多个任务？

2023年5月2日 330次阅读

我参加了
Parallel Programming课程,它显示了并行界面：

def parallel[A, B](taskA: => A, taskB: => B): (A, B) = {
  val ta = taskA
  val tb = task {taskB}
  (ta, tb.join())
}

以下是错误的：

def parallel[A, B](taskA: => A, taskB: => B): (A, B) = {
  val ta = taskB
  val tb = task {taskB}.join()
  (ta, tb)
}

在https://gist.github.com/ChenZhongPu/fe389d30626626294306264a148bd2aa更多地看到界面

它还向我们展示了执行四项任务的正确方法：

def parallel[A, B, C, D](taskA: => A, taskB: => B, taskC: => C, taskD: => D): (A, B, C, D) = {
    val ta = task { taskA }
    val tb = task { taskB }
    val tc = task { taskC }
    val td = taskD
    (ta.join(), tb.join(), tc.join(), td)
}

我的问题：如果我不知道任务的数量(任务列表),我怎样才能正确地为每个任务调用join？

tasks.map(_.join()) // wrong

编辑

类似的讨论也发生在Discuss this week’s module: Parallel Programming

最佳答案 从
Parallel Programming课程使用
framework

你可以实现这样的方法：

def parallel[A](tasks: (() => A)*): Seq[A] = {
  if (tasks.isEmpty) Nil
  else {
    val pendingTasks = tasks.tail.map(t => task { t() })
    tasks.head() +: pendingTasks.map(_.join())
  }
}

(注意你can’t have variable number of by-name arguments – 虽然这是can change)

然后像这样使用它：

object ParallelUsage {
  def main(args: Array[String]) {
    val start = System.currentTimeMillis()

    // Use a list of tasks:
    val tasks = List(longTask _, longTask _, longTask _, longTask _)
    val results = parallel(tasks: _*)
    println(results)

    // or pass any number of individual tasks directly:
    println(parallel(longTask, longTask, longTask))
    println(parallel(longTask, longTask))
    println(parallel(longTask))
    println(parallel())

    println(s"Done in ${ System.currentTimeMillis() - start } ms")
  }

  def longTask() = {
    println("starting longTask execution")
    Thread.sleep(1000)
    42 + Math.random
  }
}

使用Scala’s parallel collections

你不能比这简单：

val tasks = Vector(longTask _, longTask _, longTask _)
val results = tasks.par.map(_()).seq
点赞