Scala 中的并发 Akka 代理

Question

我现在正在做一个 scala 项目，我决定在 actor 模型上使用 Akka 的代理库，因为它允许更实用的并发方法。但是，我在运行许多不同的方法时遇到了问题代理一次。似乎我只能同时运行三到四个代理。

import akka.actor._
import akka.agent._
import scala.concurrent.ExecutionContext.Implicits.global

object AgentTester extends App {
// Create the system for the actors that power the agents
implicit val system = ActorSystem("ActorSystem")

// Create an agent for each int between 1 and 10
val agents = Vector.tabulate[Agent[Int]](10)(x=>Agent[Int](1+x))

// Define a function for each agent to execute
def printRecur(a: Agent[Int])(x: Int): Int = {
    // Print out the stored number and sleep.
    println(x)
    Thread.sleep(250)

    // Recur the agent
    a sendOff printRecur(a) _

    // Keep the agent's value the same
    x
}

// Start each agent
for(a <- agents) {
    Thread.sleep(10)
    a sendOff printRecur(a) _
}
}

上面的代码创建了一个代理，它保存 1 到 10 之间的每个整数。底部的循环将 printRecur 函数发送给每个代理。程序的输出应该显示数字 1 到 10 每四分之一秒打印一次（尽管没有任何顺序）。但是，出于某种原因，我的输出只显示了输出的数字 1 到 4。

在 Akka 中是否有更规范的方式来使用代理？我来自 clojure 背景，并且之前在那里成功使用过这种模式，所以我天真地在 Scala 中使用了相同的模式。

Answer 1

我的猜测是您在 4 核机器上运行，这就是您只看到数字 1-4 的部分原因。这里最重要的是您使用的是默认执行上下文，我猜您的系统上使用了一个只有 4 个线程的线程池（每个内核一个线程）。使用您以这种递归方式编写代码的方式，我的猜测是前 4 个代理永远不会放弃线程，它们是唯一会打印任何东西的代理。

您可以通过删除此行轻松解决此问题：

import scala.concurrent.ExecutionContext.Implicits.global

并在创建 ActorSystem 后添加此行

import system.dispatcher

这将使用 Actor 系统的默认调度程序，它是一个 fork join 调度程序，它似乎与您在示例中导入的默认执行上下文没有相同的问题。

您还可以考虑使用send 而不是sendOff，因为这将使用您构建代理时可用的执行上下文。我认为有人会在他们明确想要使用另一个执行上下文的情况下使用sendOff。