标签:
Akka 是一个用 Scala 编写的库,用于简化编写容错的、高可伸缩性的 Java 和 Scala 的 Actor 模型应用。它已经成功运用在电信行业。系统几乎不会宕机(高可用性 99.9999999 % 一年只有 31 ms 宕机。
用akka计算pi,计算公式:
这样,我们把这个公式每连续的elements个分成一段,一共分成message段。
然后开worker个actor同时计算,把结果合并。算出最终结果和耗时。
一:计算某一段的actor
我们需要一个工作actor,用来计算某一段的和,并把结果传出去。
class Worker extends Actor { //计算从start开始,连续elements个的和 def calculatePiFor(start: Int, elements: Int): Double = { var acc = 0.0 for (i <- start until (start + elements)) acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1) acc } def receive = { // sender 用来访问当前消息的发送者的引用 case Work(start, elements) => sender ! Result(calculatePiFor(start, elements)) } }二:创建主actor
主actor的工作是把任务分配下去,并且把Worker执行的结果进行收集处理,并且把处理完的最终结果返回给监听actor
分配任务我们就需要创建一个round-robin的路由器来简化把任务平均分配给工作者的过程。
//创建了一个路由器,启动了workers个Worker val workerRouter = context.actorOf( Props[Worker].withRouter(RoundRobinRouter(workers)), name = "workerRouter")对于主actor,会接受两种消息,一种是计算,一种是收到结果处理。
class Master(workers: Int, messages: Int, elements: Int, listener: ActorRef) extends Actor { var pi: Double = 0.0 var finish: Int = 0 val startTime: Long = System.currentTimeMillis() //创建了一个路由器,启动了workers个Worker val workerRouter = context.actorOf( Props[Worker].withRouter(RoundRobinRouter(workers)), name = "workerRouter") def receive = { //收到计算的请求,把计算的任务分配下去 case Calculate => for (i <- 0 until messages) workerRouter ! Work(i * elements, elements) //收到计算的结果,把计算的结果加到pi上,并且判断下发的任务有没有全部执行完毕 //如果全部执行完毕,那么给监听者发一个消息 case Result(value) => pi += value finish += 1 if (finish == messages) { listener ! PiApproximation(pi, duration = (System.currentTimeMillis - startTime)) context.stop(self) } } }三:监听actor
监听actor比较简单,收到数据,直接输出就好
class Listener extends Actor { def receive = { case PiApproximation(pi, duration) => println("计算结束,结果为: " + pi + "用时 : " + duration) } }
四:最终结果
package akka.study.base import akka.actor._ import akka.routing.RoundRobinRouter import java.util.concurrent.TimeUnit import java.lang.System._ import java.time.Duration sealed trait PiMessage case object Calculate extends PiMessage case class Work(start: Int, elements: Int) extends PiMessage case class Result(value: Double) extends PiMessage case class PiApproximation(pi: Double, duration: Long) class Worker extends Actor { //计算从start开始,连续elements个的和 def calculatePiFor(start: Int, elements: Int): Double = { var acc = 0.0 for (i <- start until (start + elements)) acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1) acc } def receive = { // sender 用来访问当前消息的发送者的引用 case Work(start, elements) => sender ! Result(calculatePiFor(start, elements)) } } class Master(workers: Int, messages: Int, elements: Int, listener: ActorRef) extends Actor { var pi: Double = 0.0 var finish: Int = 0 val startTime: Long = System.currentTimeMillis() //创建了一个路由器,启动了workers个Worker val workerRouter = context.actorOf( Props[Worker].withRouter(RoundRobinRouter(workers)), name = "workerRouter") def receive = { //收到计算的请求,把计算的任务分配下去 case Calculate => for (i <- 0 until messages) workerRouter ! Work(i * elements, elements) //收到计算的结果,把计算的结果加到pi上,并且判断下发的任务有没有全部执行完毕 //如果全部执行完毕,那么给监听者发一个消息 case Result(value) => pi += value finish += 1 if (finish == messages) { listener ! PiApproximation(pi, duration = (System.currentTimeMillis - startTime)) context.stop(self) } } } class Listener extends Actor { def receive = { case PiApproximation(pi, duration) => println("计算结束,结果为: " + pi + "用时 : " + duration) } } object Pi { def main(args: Array[String]) { def calculate(workers: Int, elements: Int, messages: Int) { val system = ActorSystem("PiSystem") val listener = system.actorOf(Props[Listener], name = "listener") val master = system.actorOf(Props(new Master( workers, messages, elements, listener)), name = "master") master ! Calculate } calculate(6, 10000, 10000) } }
标签:
原文地址:http://blog.csdn.net/rowanhaoa/article/details/44227819