Akka 是一个用 Scala 编写的库,用于简化编写容错的、高可伸缩性的 Java 和 scala 的 Actor 模型应用。它已经成功运用在电信行业。系统几乎不会宕机(高可用性 99.9999999 % 一年只有 31 ms 宕机。
用akka计算pi,计算公式:泰勒展开式
这样,我们把这个公式每连续的elements个分成一段,一共分成message段。
然后开worker个actor同时计算,把结果合并。算出最终结果和耗时。
package ljt.scalalearing.pi
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.routing.RoundRobinRouter
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)
/**
* 一:计算某一段的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的路由器来简化把任务平均分配给工作者的过程。
*/
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 + "\n 用时 : " + 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)
}
}