Scala中的并发Akka代理

Question

我正在研究一个scala项目,我决定使用Akka的代理库而不是actor模型,因为它允许更多功能性的并发方法.但是,我遇到了运行许多不同代理的问题.时间.看起来我只能同时运行三到四个代理.

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当他们有一个他们明确想要使用另一个执行上下文的情况时会使用.