在Scala中同时处理
作为
in my own answer to my own question,我遇到的情况是我正在处理大量到达队列的事件.每个事件都以完全相同的方式处理,甚至可以独立于所有其他事件处理每个事件.
我的程序利用了Scala并发框架,并且涉及的许多过程都被建模为Actors.当Actors按顺序处理它们的消息时,它们并不适合这个特定的问题(即使我的其他actor正在执行顺序的动作).因为我希望Scala“控制”所有线程创建(我假设它首先有一个并发系统),我似乎有两个选择: >将事件发送到我控制的事件处理器池 我原以为#1否定了使用actors子系统的意义:我应该创建多少个处理器actor?是一个明显的问题.这些东西据说对我来说是隐藏的,并由子系统解决. 我的回答是做以下事情: val eventProcessor = actor { loop { react { case MyEvent(x) => //I want to be able to handle multiple events at the same time //create a new actor to handle it actor { //processing code here process(x) } } } } 有更好的方法吗?这不正确吗? 编辑:一个可能更好的方法是: val eventProcessor = actor { loop { react { case MyEvent(x) => //Pass processing to the underlying ForkJoin framework Scheduler.execute(process(e)) } } } 解决方法
这似乎是另一个问题的重复.所以我会重复我的回答
Actor一次处理一条消息.处理多个消息的经典模式是为消费者角色池设置一个协调者角色.如果使用react,则使用者池可能很大,但仍然只使用少量JVM线程.这是一个例子,我为他们创建了一个由10个消费者和一个协调员组成的池. import scala.actors.Actor import scala.actors.Actor._ case class Request(sender : Actor,payload : String) case class Ready(sender : Actor) case class Result(result : String) case object Stop def consumer(n : Int) = actor { loop { react { case Ready(sender) => sender ! Ready(self) case Request(sender,payload) => println("request to consumer " + n + " with " + payload) // some silly computation so the process takes awhile val result = ((payload + payload + payload) map {case '0' => 'X'; case '1' => "-"; case c => c}).mkString sender ! Result(result) println("consumer " + n + " is done processing " + result ) case Stop => exit } } } // a pool of 10 consumers val consumers = for (n <- 0 to 10) yield consumer(n) val coordinator = actor { loop { react { case msg @ Request(sender,payload) => consumers foreach {_ ! Ready(self)} react { // send the request to the first available consumer case Ready(consumer) => consumer ! msg } case Stop => consumers foreach {_ ! Stop} exit } } } // a little test loop - note that it's not doing anything with the results or telling the coordinator to stop for (i <- 0 to 1000) coordinator ! Request(self,i.toString) 此代码测试以查看哪些消费者可用并向该使用者发送请求.替代方案是随机分配给消费者或使用循环调度程序. 根据您的工作情况,您可能会更好地使用Scala的期货.例如,如果你真的不需要演员,那么所有上述机器都可以写成 import scala.actors.Futures._ def transform(payload : String) = { val result = ((payload + payload + payload) map {case '0' => 'X'; case '1' => "-"; case c => c}).mkString println("transformed " + payload + " to " + result ) result } val results = for (i <- 0 to 1000) yield future(transform(i.toString)) (编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |