使用Scala actor执行CPU绑定任务？

for(i <- 1 to 20) {
  actor {
    print(i); 
    Thread.sleep(1000);
  }
}

这里的缺点取决于任务的数量,为每个任务创建线程的成本可能非常昂贵,因为线程在Java中并不便宜.

创建有界工作者角色池然后通过消息传递将任务分发给他们的简单方法是：

import scala.actors.Actor._

val numWorkers = 4
val pool = (1 to numWorkers).map { i =>
  actor { 
    loop { 
      react { 
        case x: String => println(x) 
      } 
    }
  }
}

for(i <- 1 to 20) {
  val r = (new util.Random).nextInt(numWorkers)
  pool(r) ! "task "+i
}

我们想要创建多个actor的原因是因为一个actor一次只处理一个消息(即任务),所以为你需要创建多个任务获得并行性.

旁注：当涉及到I / O绑定任务时,默认调度程序变得尤为重要,因为在这种情况下,您肯定希望更改线程池的大小.关于此的详细信息有两个很好的博客文章：Explore the Scheduling of Scala Actors和Scala actors thread pool pitfall.

话虽如此,Akka是一个Actor框架,它为使用Actors的更高级工作流提供工具,我将在任何实际应用程序中使用它.这是一个负载平衡(而不是随机)任务执??行器：

import akka.actor.Actor
import Actor._
import akka.routing.{LoadBalancer,CyclicIterator}

class TaskHandler extends Actor {
  def receive = {
    case t: Task =>
      // some computationally expensive thing
      t.execute
    case _ => println("default case is required in Akka...")
  }
}

class TaskRouter(numWorkers: Int) extends Actor with LoadBalancer {
   val workerPool = Vector.fill(numWorkers)(actorOf[TaskHandler].start())
   val seq = new CyclicIterator(workerPool)
}

val router = actorOf(new TaskRouter(4)).start()

for(i <- 1 to 20) {
  router ! Task(..)
}

您可以使用不同类型的负载平衡(CyclicIterator是循环分配),因此您可以查看文档here以获取更多信息.