【问题标题】:Why is my Dispatching on Actors scaled down in Akka?为什么我在 Akka 中对 Actors 的调度会按比例缩小?
【发布时间】:2011-04-28 12:38:03
【问题描述】:

我有一个包含 100 个正在运行的 Actor 的 Actor 池,它们共享一个工作窃取调度程序,其 CorePoolSize 设置为 100。但是现在当向其中一个 Actor 发送 19 条消息时,这 19 条消息没有并行化到 19 个 Actor,只有5 条消息并行运行。当这 5 条消息完成后,接下来的 5 条消息将由这 5 个相同的 Actor 再次处理,依此类推。为什么我的 19 条消息没有并行运行,我在这里缺少什么?

我的代码基本上是这样的:

object TestActor {
  val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher("pool")
                   .setCorePoolSize(100)
                   .setMaxPoolSize(100)
                   .build
}

class TestActor(val name: Integer) extends Actor {
    self.lifeCycle = Permanent
    self.dispatcher = TestActor.dispatcher
    def receive = {
       case num: Integer => {  println("Actor: " + name + " Received: " + num)
                               Thread.sleep(10000)
                            }
    }
}

trait CyclicLoadBalancing extends LoadBalancer { this: Actor =>
    val testActors: List[ActorRef]
    val seq = new CyclicIterator[ActorRef](testActors)
}

trait TestActorManager extends Actor {
     self.lifeCycle = Permanent
     self.faultHandler = OneForOneStrategy(List(classOf[Exception]), 5, 5000)
     val testActors: List[ActorRef]
     override def preStart = testActors foreach { self.startLink(_) }
     override def postStop = self.shutdownLinkedActors()
}

val supervisor = actorOf(new TestActorManager with CyclicLoadBalancing {
val testActors = (1 until 100 toList) map (i => actorOf(new TestActor(i)))   
}).start

println("Number of Actors: " +  Actor.registry.actorsFor(classOf[TestActor]).length)

val testActor = Actor.registry.actorsFor(classOf[TestActor]).head

(1 until 20 toList) foreach { testActor ! _ }

输出:

Actor: 4 Received: 16
Actor: 3 Received: 17
Actor: 1 Received: 19
Actor: 59 Received: 1
Actor: 2 Received: 18

// 10 secs. are passing..

Actor: 4 Received: 15
Actor: 3 Received: 14
Actor: 1 Received: 13
Actor: 59 Received: 2
Actor: 2 Received: 12

// 10 secs. are passing..

Actor: 4 Received: 11
Actor: 3 Received: 10
Actor: 59 Received: 3
Actor: 2 Received: 8
Actor: 1 Received: 9

// 10 secs. are passing..

Actor: 4 Received: 7
Actor: 3 Received: 6
Actor: 59 Received: 4
Actor: 2 Received: 5

编辑:我正在使用 Akka 1.0

【问题讨论】:

标签: scala akka


【解决方案1】:

感谢您的询问,我已在对 Akka master 的提交中定位了瓶颈并修复了它:

https://github.com/akka/akka/commit/e4e99ef56399e892206ce4a46b9a9107da6c7770

它将在 Akka 1.1-RC1 中发布

干杯, √

【讨论】:

  • 嗨,谢谢,会检查一下 :) 还要感谢 Akka 的出色工作,到目前为止我对框架非常满意 :)
【解决方案2】:

我认为调度程序允许您自定义吞吐量属性。这定义了调度程序应在一次扫描中处理的特定 Actor 的消息数量。您可以在 akka.conf 文件中添加以下配置

actor {
  throughput = 20
}

默认为 5

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 2018-06-12
    • 1970-01-01
    • 2015-09-16
    • 1970-01-01
    • 2012-06-02
    • 2013-11-25
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多