在Scala中使用actor时如何限制并发？答案

【问题标题】：How to limit concurrency when using actors in Scala?在Scala中使用actor时如何限制并发？
【发布时间】：2011-01-19 17:03:22
【问题描述】：

我来自 Java，在那里我将 Runnables 提交给由线程池支持的 ExecutorService。在 Java 中如何设置线程池大小的限制非常清楚。

我对使用 Scala 演员很感兴趣，但我不清楚如何限制并发。

假设我正在创建一个接受“作业”的 Web 服务。提交了带有POST 请求的作业，我希望我的服务将作业排入队列，然后立即返回202 Accepted — 即异步处理作业。

如果我使用演员来处理队列中的作业，我如何限制同时处理的作业数量？

我可以想出几种不同的方法来解决这个问题；我想知道是否有社区最佳实践，或者至少有一些在 Scala 世界中有些标准的明确确立的方法。

我想到的一种方法是让一个协调器参与者来管理作业队列和作业处理参与者；我想它可以使用一个简单的 int 字段来跟踪当前正在处理的作业数。但是，我确信这种方法会有一些问题，例如确保跟踪错误发生的时间以减少数字。这就是为什么我想知道 Scala 是否已经为此提供了一种更简单或更封装的方法。

顺便说一句，我试图问这个问题a while ago，但我问得很糟糕。

谢谢！

【问题讨论】：

标签： scala concurrency actor

【解决方案1】：

您可以覆盖系统属性actors.maxPoolSize 和actors.corePoolSize，它们限制了actor 线程池的大小，然后将尽可能多的作业扔到池中，就像您的actor 可以处理的那样。为什么你认为你需要限制你的反应？

【讨论】：

非常有用，谢谢！我不确定我是否会使用术语throttle，但无论哪种方式，有时都需要限制同时“进程”的数量，因为它们所做的工作是资源密集型的。
这种方法可能不会产生预期的结果。它将允许作业排队，直到 JVM 内存不足。限制参与者可以使用的线程数量只会限制实际并发执行的作业数量。我通过比演员以前更快的速度生成工作产生了 OOM 错误，所以你必须小心。
我认为这种方法的一个缺点是它是全球性的。有时我需要运行不同类型的进程，这些进程具有不同级别的资源利用率——使用 Java 线程池，我可以轻松地使用具有不同设置的不同池。使用actors.maxPoolSize，我只能为所有参与者使用一个数字，因为它们都由同一个线程池提供支持，对吧？

【解决方案2】：

我真的鼓励您看看 Akka，它是 Scala 的替代 Actor 实现。

http://www.akkasource.org

Akka 已经集成了 JAX-RS[1]，您可以将它与 LoadBalancer[2] 结合使用来限制可以并行执行的操作数：

[1]http://doc.akkasource.org/rest [2]http://github.com/jboner/akka/blob/master/akka-patterns/src/main/scala/Patterns.scala

【讨论】：

部分链接失效。

【解决方案3】：

你真的有两个问题。

首先是控制参与者使用的线程池。这可以通过设置系统属性actors.maxPoolSize来完成。

第二个是已提交到池中的任务数量的失控增长。您可能关心也可能不关心这个问题，但是完全有可能触发故障条件，例如内存不足错误，并且在某些情况下，由于太快生成太多任务，可能会引发更微妙的问题。

每个工作线程都维护一个任务队列。出队被实现为一个数组，工作线程将动态扩大到某个最大大小。在 2.7.x 中，队列本身会变得非常大，我已经看到当与大量并发线程结合使用时会触发内存不足错误。最大出队大小小于 2.8。出队也可以填满。

解决此问题需要您控制生成的任务数量，这可能意味着您已经概述了某种协调器。当启动某种数据处理管道的参与者比管道中稍后的参与者快得多时，我遇到了这个问题。为了控制这个过程，我通常让链中较晚的参与者在每 X 条消息后 ping 回链中较早的参与者，并让链中较早的参与者在 X 条消息后停止并等待 ping 返回。您也可以使用更集中的协调器来做到这一点。

【讨论】：