我有 50,000 个任务，想用 10 个线程执行它们。 在 Java 中，我应该创建 Executers.threadPool(10) 并将 runnable 传递给然后等待处理所有内容。据我了解，Scala 对于该任务特别有用，但我在文档中找不到解决方案。
【问题讨论】：
标签： multithreading scala concurrency
最简单的方法是使用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 提供基于参与者的并发，并与 scala.concurrent 互操作。
这个最简单的方法是使用 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)
【讨论】：
您想查看 Scala 演员库或 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
您需要创建一个参与者实例池并将您的任务作为消息发送给它们。这是一篇关于 Akka actor pooling 的帖子，可能会有所帮助：http://vasilrem.com/blog/software-development/flexible-load-balancing-with-akka-in-scala/
在您的情况下，每个任务一个参与者可能是合适的（参与者与线程相比非常轻量级，因此您可以在单个 VM 中拥有很多参与者），或者您可能需要在它们之间进行一些更复杂的负载平衡。
编辑： 使用上面的示例actor，向它发送消息就像这样简单：
myActor ! "test"
然后演员将“收到的测试”输出到标准输出。
消息可以是任何类型，当与 Scala 的模式匹配结合使用时，您将拥有一个强大的模式来构建灵活的并发应用程序。
一般来说，Akka 演员会在线程共享方面“做正确的事”，并且对于 OP 的需求，我想默认值是可以的。但如果需要，您可以将 actor 应使用的调度程序设置为以下几种类型之一：
* Thread-based
* Event-based
* Work-stealing
* HawtDispatch-based event-driven
为演员设置调度程序很简单：
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 的“杀手级应用”或“杀手级框架”。
【讨论】：
如果您想“使用 10 个线程执行它们”，请使用线程。 Scala 的 Actor 模型，通常是人们说 Scala 擅长并发时所说的，它隐藏这样的细节，所以你不会看到它们。
使用actors，或者你所拥有的futures只是简单的计算，你只需创建50000个它们并让它们运行。您可能会遇到问题，但它们的性质不同。
【讨论】：
这是另一个类似于 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
【讨论】：