Akka -负载均衡和充分利用处理器

Question

我写了一个矩阵乘法算法，它使用并行集合来加速乘法。

它是这样的：

(0 until M1_ROWS).grouped(PARTITION_ROWS).toList.par.map( i => 
  singleThreadedMultiplicationFAST(i.toArray.map(m1(_)), m2) 
).reduce(_++_) 

现在我想在Akka中做同样的事情，所以我所做的是：

val multiplyer = actorOf[Pool] 
multiplyer start 
val futures = (0 until M1_ROWS).grouped(PARTITION_ROWS).map( i => 
  multiplyer ? MultiplyMatrix(i.toArray.map(m1(_)), m2) 
) 
futures.map(_.get match { case res :Array[Array[Double]] => res }).reduce(_++_) 

class Multiplyer extends akka.actor.Actor{ 
  protected def receive = { 
    case MultiplyMatrix(m1, m2) => self reply singleThreadedMultiplicationFAST (m1,m2) 
  } 
} 
class Pool extends Actor with DefaultActorPool 
  with FixedCapacityStrategy with RoundRobinSelector { 

  def receive = _route 
  def partialFill = false 
  def selectionCount = 1 
  def instance = actorOf[Multiplyer] 
  def limit = 32 // I tried 256 with no effect either 
} 

事实证明，在我的i7 sandybridge上，基于角色的算法版本只使用了200%，而并行集合版本使用了600%的处理器，并且速度快了4-5倍。我想可能是调度员的原因，然后试了试：

self.dispatcher = Dispatchers.newThreadBasedDispatcher(self, mailboxCapacity = 100) 

还有这个(我在演员之间分享了这个)：

val messageDispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher("d1")
  .withNewBoundedThrea dPoolWithLinkedBlockingQueueWithUnboundedCapacity(100)
  .setCorePoolSize(16)
  .setMaxPoolSize(128)
  .setKeepAliveTimeInMillis(60000).build 

但我没有观察到任何变化。仍然只有200%的处理器使用率，算法比并行集合版本慢4-5倍。

我确信我在做一些愚蠢的事情，所以请帮帮我！：)

Answer 1

此表达式：

val futures = (0 until M1_ROWS).grouped(PARTITION_ROWS).map( i => 
  multiplyer ? MultiplyMatrix(i.toArray.map(m1(_)), m2) 
) 

创建一个惰性的集合，这样你的_.get就可以让你的整个程序成为串行的。因此，解决方案是通过添加toList或类似内容来严格执行该表达式。