如何在 Scala 中执行多个任务？

Question

我有 50,000 个任务，想用 10 个线程执行它们。 在Java中，我应该创建Executers.threadPool(10)并将runnable传递给然后等待处理所有线程。据我了解，Scala 对于该任务特别有用，但我在文档中找不到解决方案。

Answer 1

Scala 2.9.3 及更高版本

最简单的方法是使用 scala.concurrent.Future 类和关联的基础设施。 scala.concurrent.future 方法异步计算传递给它的块，并立即返回表示异步计算的 Future[A]。 Futures 可以通过多种非阻塞方式进行操作，包括映射、flatMapping、过滤、恢复错误等。

例如，这里有一个创建 10 个任务的示例，其中每个任务休眠任意时间，然后返回平方传递给它的值。

import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

val tasks: Seq[Future[Int]] = for (i <- 1 to 10) yield future {
  println("Executing task " + i)
  Thread.sleep(i * 1000L)
  i * i
}

val aggregated: Future[Seq[Int]] = Future.sequence(tasks)

val squares: Seq[Int] = Await.result(aggregated, 15.seconds)
println("Squares: " + squares)

在此示例中，我们首先创建一系列单独的异步任务，这些任务完成后提供一个 int。然后，我们使用 Future.sequence 将这些异步任务合并为一个异步任务 - 交换 Future 和 Seq 的位置类型。最后，我们在等待结果时阻塞当前线程最多 15 秒。在示例中，我们使用全局执行上下文，它由 fork/join 线程池支持。对于重要的示例，您可能希望使用特定于应用程序的 ExecutionContext。

一般来说，应尽可能避免阻塞。 Future 类上还有其他组合器可以帮助以异步方式进行编程，包括 onSuccess、onFailure 和 onComplete< /代码>。

另外，请考虑研究 Akka 库，它为 Scala 和 Java 提供基于 actor 的并发性，并与 scala 进行互操作。并发。

Scala 2.9.2 及更早版本

最简单的方法是使用 Scala 的 Future 类，它是 Actors 框架的子组件。 scala.actors.Futures.future 方法为传递给它的块创建一个 Future。然后，您可以使用 scala.actors.Futures.awaitAll 等待所有任务完成。

例如，下面的示例创建了 10 个任务，其中每个任务休眠任意时间，然后返回传递给它的值的平方。

import scala.actors.Futures._

val tasks = for (i <- 1 to 10) yield future {
  println("Executing task " + i)
  Thread.sleep(i * 1000L)
  i * i
}

val squares = awaitAll(20000L, tasks: _*)
println("Squares: " + squares)

Answer 2

您想查看 Scala actor 库或 Akka。 Akka 具有更简洁的语法，但两者都可以解决问题。

因此，听起来您需要创建一个知道如何处理任务的参与者池。 Actor 基本上可以是具有接收方法的任何类 - 来自 Akka 教程 (http:// doc.akkasource.org/tutorial-chat-server-scala)：

class MyActor extends Actor {
  def receive = {
    case "test" => println("received test")
    case _ =>      println("received unknown message")
 }}

val myActor = Actor.actorOf[MyActor]
myActor.start

您需要创建一个 Actor 实例池并将任务作为消息发送给它们。这是一篇关于 Akka actor 池的文章，可能会有所帮助：http://vasilrem.com/blog/software-development/flexible-load-balancing-with-akka-in-scala/

在您的情况下，每个任务一个参与者可能是合适的（参与者非常轻量级）与线程相比，您可以在单个虚拟机中拥有很多线程），或者您可能需要在它们之间进行一些更复杂的负载平衡。

编辑：
使用上面的示例 actor，向其发送消息就像这样简单：

myActor ! "test"

然后 actor 将“收到的测试”输出到标准输出。

消息可以是任何类型，当与 Scala 的模式匹配相结合时，您就拥有了构建灵活的并发应用程序的强大模式。

一般来说，Akka Actor 会在线程共享方面“做正确的事情”，并且对于 OP 的需求，我认为默认值就可以了。但如果需要，您可以将 actor 应该使用的调度程序设置为以下几种类型之一：

* Thread-based
* Event-based
* Work-stealing
* HawtDispatch-based event-driven

为 actor 设置调度程序很简单：

class MyActor extends Actor {
  self.dispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher("thread-pool-dispatch")
    .withNewThreadPoolWithBoundedBlockingQueue(100)
    .setCorePoolSize(10)
    .setMaxPoolSize(10)
    .setKeepAliveTimeInMillis(10000)
    .build
}

请参阅 http://doc.akkasource.org/dispatchers-scala

通过这种方式，您可以限制线程池大小，但同样，原始用例可能会满足 50K Akka使用默认调度程序的 Actor 实例可以很好地并行化。

这实际上只是 Akka 功能的冰山一角。它为 Scala 语言带来了 Erlang 提供的许多功能。 Actor 可以监视其他 Actor 并重新启动它们，从而创建自我修复的应用程序。 Akka 还提供软件事务内存和许多其他功能。它可以说是 Scala 的“杀手级应用程序”或“杀手级框架”。

Answer 3

如果你想“用10个线程执行它们”，那么就使用线程。 Scala 的 Actor 模型（通常是人们在说 Scala 适合并发时所谈论的）隐藏这些细节，这样您就看不到它们。

使用 actor 或 futures 以及你所拥有的一切都是简单的计算，你只需创建 50000 个并让它们运行。您可能会遇到问题，但它们的性质不同。

Answer 4

这是另一个类似于 mpilquist 响应的答案，但没有弃用的 API，并通过自定义 ExecutionContext 包含线程设置：

import java.util.concurrent.Executors
import scala.concurrent.{ExecutionContext, Await, Future}
import scala.concurrent.duration._

val numJobs = 50000
var numThreads = 10

// customize the execution context to use the specified number of threads
implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(numThreads))


// define the tasks
val tasks = for (i <- 1 to numJobs) yield Future {
  // do something more fancy here
  i
}

// aggregate and wait for final result
val aggregated = Future.sequence(tasks)
val oneToNSum = Await.result(aggregated, 15.seconds).sum