akka 将任务拆分为更小的并折叠结果

Question

问题是关于 Akka 演员 库的。 A 想要将一项大任务拆分为较小的任务，然后将它们的结果折叠成一个“大”结果。这会给我更快的计算利润。如果较小的任务是独立的，则可以并行计算。

假设我们需要像这样计算一些东西。函数count2X比较耗时，所以在一个线程中多次使用不是最优的。

//NOT OPTIMAL
def count2X(x: Int) = {
  Thread.sleep(1000)
  x * 2
}

val sum = count2X(1) + count2X(2) + count2X(3)
println(sum)

问题来了。

如何调度任务并收集结果然后折叠它们，全部使用 akka actor？ Akka 已经提供了这样的功能还是我需要自己实现它？这种方法的最佳实践是什么。

这是对我的问题的“视觉”解释：

             /-> [SMALL_TASK_1] -\
[BIG_TASK] -+--> [SMALL_TASK_1] --> [RESULT_FOLD]
             \-> [SMALL_TASK_1] -/

以下是我的脚手架实现，其中缺少/错误的实现:)

case class Count2X(x: Int)

class Count2XActor extends Actor {
  def receive = {
    case Count2X(x) => count2X(x); // AND NOW WHAT ?
  }
}

case class CountSumOf2X(a: Int, b: Int, c: Int)

class SumOf2XActor extends Actor {
  val aCounter = context.actorOf(Props[Count2XActor])
  val bCounter = context.actorOf(Props[Count2XActor])
  val cCounter = context.actorOf(Props[Count2XActor])

  def receive = {
    case CountSumOf2X(a, b, c) => // AND NOW WHAT ? aCounter ! Count2X(a); bCounter ! Count2X(b); cCounter ! Count2X(c);
  }
}

val aSystem = ActorSystem("mySystem")
val actor = aSystem.actorOf(Props[SumOf2XActor])

actor ! CountSumOf2X(10, 20, 30)

感谢您的帮助。

Answer 1

在 Akka 我会做这样的事情：

val a = aCounter ? Count2X(10) mapTo[Int]
val b = bCounter ? Count2X(10) mapTo[Int]
val c = cCounter ? Count2X(10) mapTo[Int]
Await.result(Future.sequence(a, b, c) map (_.sum), 1 second).asInstanceOf[Int]

我确信有更好的方法 - 在所有 Future-s 并行完成之后，您开始汇总结果，对于简单的任务没关系，但通常您不应该等待这么久

Answer 2

你可以做两件事：

1) 使用 Akka 期货。这些允许您分派操作并以异步方式折叠它们。查看http://doc.akka.io/docs/akka/2.0.4/scala/futures.html 了解更多信息。

2) 您可以将工作分派给多个“工人”actor，然后让一个“主”actor 聚合它们，通过在消息本身中存储信息来跟踪哪些消息正在等待/处理。我在这里有一个使用 Akka 演员的简单股票报价示例：https://github.com/ryanlecompte/quotes