问题陈述
假设我有一个文件，里面有逐行处理的句子。在我的情况下，我需要从这些行中提取命名实体(个人、组织…)。不幸的是，这个标签很慢。因此，我决定将计算并行化，这样就可以相互独立地处理线条，并将结果收集在中心位置。
当前方法
我目前的方法包括使用单一生产者多消费者的概念。然而，我对Akka相对陌生，但我认为我的问题描述很适合它的能力。让我给你看一些代码：
生产者
Producer逐行读取文件并将其发送给Consumer。如果达到总行限制，则将结果传播回WordCount。
class Producer(consumers: ActorRef) extends Actor with ActorLogging {
var master: Option[ActorRef] = None
var result = immutable.List[String]()
var totalLines = 0
var linesProcessed = 0
override def receive = {
case StartProcessing() => {
master = Some(sender)
Source.fromFile("sent.txt", "utf-8").getLines.foreach { line =>
consumers ! Sentence(line)
totalLines += 1
}
context.stop(self)
}
case SentenceProcessed(list) => {
linesProcessed += 1
result :::= list
//If we are done, we can propagate the result to the creator
if (linesProcessed == totalLines) {
master.map(_ ! result)
}
}
case _ => log.error("message not recognized")
}
}
消费者
class Consumer extends Actor with ActorLogging {
def tokenize(line: String): Seq[String] = {
line.split(" ").map(_.toLowerCase)
}
override def receive = {
case Sentence(sent) => {
//Assume: This is representative for the extensive computation method
val tokens = tokenize(sent)
sender() ! SentenceProcessed(tokens.toList)
}
case _ => log.error("message not recognized")
}
}
WordCount(母版)
class WordCount extends Actor {
val consumers = context.actorOf(Props[Consumer].
withRouter(FromConfig()).
withDispatcher("consumer-dispatcher"), "consumers")
val producer = context.actorOf(Props(new Producer(consumers)), "producer")
context.watch(consumers)
context.watch(producer)
def receive = {
case Terminated(`producer`) => consumers ! Broadcast(PoisonPill)
case Terminated(`consumers`) => context.system.shutdown
}
}
object WordCount {
def getActor() = new WordCount
def getConfig(routerType: String, dispatcherType: String)(numConsumers: Int) = s"""
akka.actor.deployment {
/WordCount/consumers {
router = $routerType
nr-of-instances = $numConsumers
dispatcher = consumer-dispatcher
}
}
consumer-dispatcher {
type = $dispatcherType
executor = "fork-join-executor"
}"""
}
WordCount参与者负责创建其他参与者。当Consumer完成时，Producer发送具有所有令牌的消息。但是，如何再次传播信息并接受和等待它呢？第三个WordCount参与者的体系结构可能是错误的。
主例程
case class Run(name: String, actor: () => Actor, config: (Int) => String)
object Main extends App {
val run = Run("push_implementation", WordCount.getActor _, WordCount.getConfig("balancing-pool", "Dispatcher") _)
def execute(run: Run, numConsumers: Int) = {
val config = ConfigFactory.parseString(run.config(numConsumers))
val system = ActorSystem("Counting", ConfigFactory.load(config))
val startTime = System.currentTimeMillis
system.actorOf(Props(run.actor()), "WordCount")
/*
How to get the result here?!
*/
system.awaitTermination
System.currentTimeMillis - startTime
}
execute(run, 4)
}
问题
正如您所看到的，实际的问题是将结果传播回Main例程。你能告诉我如何正确地做这件事吗？问题还在于，如何等待结果，直到消费者完成？我简要地看了一下AkkaFuture文档部分，但对于初学者来说，整个系统有点让人不知所措。像var future = message ? actor这样的东西似乎很合适。不知道该怎么做。此外，使用WordCountactor会导致额外的复杂性。也许有可能想出一个不需要这个演员的解决方案？
考虑使用Akka聚合器模式。这就照顾到了低级的原始人(看演员、毒丸等)。你可以专注于管理状态。
您对system.actorOf()的调用返回一个ActorRef，但您没有使用它。您应该向该参与者询问结果。类似这样的东西：
implicit val timeout = Timeout(5 seconds)
val wCount = system.actorOf(Props(run.actor()), "WordCount")
val answer = Await.result(wCount ? "sent.txt", timeout.duration)
这意味着您的WordCount类需要一个接受String消息的receive方法。这段代码应该聚合结果并告诉sender()，如下所示：
class WordCount extends Actor {
def receive: Receive = {
case filename: String =>
// do all of your code here, using filename
sender() ! results
}
}
此外，您可以应用一些处理Futures的技术，而不是用上面的Await阻塞结果。