Почему мой Диспетчерский актер уменьшен в Акке? - PullRequest
Новая
       33

Почему мой Диспетчерский актер уменьшен в Акке?

8 голосов
/ 28 апреля 2011

У меня есть пул актеров из 100 действующих актеров, которые совместно используют диспетчер рабочих краж с установленным значением CorePoolSize равным 100. Но теперь при отправке 19 сообщений одному из акторов 19 сообщений не распараллеливаются на 19 актеров, есть5 сообщений, работающих параллельно.Когда эти 5 сообщений завершены, следующие 5 сообщений снова обрабатываются этими же 5 субъектами и так далее.Почему мои 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

edit: я использую Akka 1.0

Ответы [ 2 ]

12 голосов
/ 29 апреля 2011

спасибо за ваш запрос, я локализовал узкое место и исправил его в этом коммите для мастера Akka:

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

Он будет выпущен в Akka 1.1-RC1

Ура, √

2 голосов
/ 28 апреля 2011

Я думаю, что диспетчер позволяет вам настроить свойство пропускной способности. Это определяет количество сообщений для конкретного субъекта, который диспетчер должен обработать за один раз. Вы можете добавить следующую конфигурацию в ваш файл akka.conf 

actor {
  throughput = 20
}

По умолчанию это 5

...