【发布时间】：2017-01-21 20:13:33
【问题描述】：
我正在尝试使用 Scala 和 Akka 进行一些并行编程，我是新手。我有一个非常简单的 Monte Carlo Pi 应用程序（在一个圆圈中近似 pi），我用多种语言构建了它。然而，我在 Akka 中构建的版本的性能让我感到困惑。
我有一个用纯 Scala 编写的顺序版本，通常需要大约 400 毫秒才能完成。
与 1 个 worker actor 相比，Akka 版本大约需要 300-350 毫秒，但是随着我增加 actor 的数量，时间会急剧增加。对于 4 个演员，时间可以在 500 毫秒到 1200 毫秒或更高之间的任何地方。
迭代次数在工作角色之间分配，因此理想情况下，他们越多，性能应该越好，目前它变得越来越差。
我的代码是
object MCpi{
//Declare initial values
val numWorkers = 2
val numIterations = 10000000
//Declare messages that will be sent to actors
sealed trait PiMessage
case object Calculate extends PiMessage
case class Work(iterations: Int) extends PiMessage
case class Result(value: Int) extends PiMessage
case class PiApprox(pi: Double, duration: Double)
//Main method
def main(args: Array[String]): Unit = {
val system = ActorSystem("MCpi_System") //Create Akka system
val master = system.actorOf(Props(new MCpi_Master(numWorkers, numIterations))) //Create Master Actor
println("Starting Master")
master ! Calculate //Run calculation
}
}
//Master
class MCpi_Master(numWorkers: Int, numIterations: Int) extends Actor{
var pi: Double = _ // Store pi
var quadSum: Int = _ //the total number of points inside the quadrant
var numResults: Int = _ //number of results returned
val startTime: Double = System.currentTimeMillis() //calculation start time
//Create a group of worker actors
val workerRouter = context.actorOf(
Props[MCpi_Worker].withRouter(RoundRobinPool(numWorkers)), name = "workerRouter")
val listener = context.actorOf(Props[MCpi_Listener], name = "listener")
def receive = {
//Tell workers to start the calculation
//For each worker a message is sent with the number of iterations it is to perform,
//iterations are split up between the number of workers.
case Calculate => for(i <- 0 until numWorkers) workerRouter ! Work(numIterations / numWorkers);
//Receive the results from the workers
case Result(value) =>
//Add up the total number of points in the circle from each worker
quadSum += value
//Total up the number of results which have been received, this should be 1 for each worker
numResults += 1
if(numResults == numWorkers) { //Once all results have been collected
//Calculate pi
pi = (4.0 * quadSum) / numIterations
//Send the results to the listener to output
listener ! PiApprox(pi, duration = System.currentTimeMillis - startTime)
context.stop(self)
}
}
}
//Worker
class MCpi_Worker extends Actor {
//Performs the calculation
def calculatePi(iterations: Int): Int = {
val r = scala.util.Random // Create random number generator
var inQuadrant: Int = 0 //Store number of points within circle
for(i <- 0 to iterations){
//Generate random point
val X = r.nextFloat()
val Y = r.nextFloat()
//Determine whether or not the point is within the circle
if(((X * X) + (Y * Y)) < 1.0)
inQuadrant += 1
}
inQuadrant //return the number of points within the circle
}
def receive = {
//Starts the calculation then returns the result
case Work(iterations) => sender ! Result(calculatePi(iterations))
}
}
//Listener
class MCpi_Listener extends Actor{ //Recieves and prints the final result
def receive = {
case PiApprox(pi, duration) =>
//Print the results
println("\n\tPi approximation: \t\t%s\n\tCalculation time: \t%s".format(pi, duration))
//Print to a CSV file
val pw: FileWriter = new FileWriter("../../../..//Results/Scala_Results.csv", true)
pw.append(duration.toString())
pw.append("\n")
pw.close()
context.system.terminate()
}
}
普通的 Scala 顺序版本是
object MCpi {
def main(args: Array[String]): Unit = {
//Define the number of iterations to perform
val iterations = args(0).toInt;
val resultsPath = args(1);
//Get the current time
val start = System.currentTimeMillis()
// Create random number generator
val r = scala.util.Random
//Store number of points within circle
var inQuadrant: Int = 0
for(i <- 0 to iterations){
//Generate random point
val X = r.nextFloat()
val Y = r.nextFloat()
//Determine whether or not the point is within the circle
if(((X * X) + (Y * Y)) < 1.0)
inQuadrant += 1
}
//Calculate pi
val pi = (4.0 * inQuadrant) / iterations
//Get the total time
val time = System.currentTimeMillis() - start
//Output values
println("Number of Iterations: " + iterations)
println("Pi has been calculated as: " + pi)
println("Total time taken: " + time + " (Milliseconds)")
//Print to a CSV file
val pw: FileWriter = new FileWriter(resultsPath + "/Scala_Results.csv", true)
pw.append(time.toString())
pw.append("\n")
pw.close()
}
}
非常欢迎任何关于为什么会发生这种情况或如何提高性能的建议。
编辑：我要感谢大家的回答，这是我在这个网站上的第一个问题，所有的答案都非常有帮助，我现在有很多东西要看:)
【问题讨论】：
-
在这种情况下，有关您运行它的处理器类型的一些信息可能会有所帮助。
-
1) 请在 SO 上发布您的代码。在发布之前对其进行格式化。 2) 当您多次执行
calculatePi方法时，您甚至对actor 实现有什么期望，这是我所看到的，相当于您的顺序实现？从我所见，您只是多次计算 PI（计算次数等于工人演员的数量，这可能是减速的原因）？如我错了请纠正我。 3) 您是否考虑过在这种情况下使用 Actor 模型可能不会获得任何收益？ -
@Jasper-M 处理器是 Intel i7-4510U 四核 @ 3.1GHz @Branislav 1) 好的，我稍后有空时会尝试用代码更新帖子。 2)
calculatePi由每个工人运行，它生成许多随机点并测量这些点是否在特定大小的“圆圈”内（在本例中为 1.0），然后返回圆圈中有多少点（quadSum） ，一旦从每个工人那里得到结果，计算就会完成一次，以计算出 Pi 是什么（在主演员中）。 3) 我假设我会得到某种性能提升，将工作分配给多个演员。 -
这是一个 CPU 密集型任务，参与者共享一个线程池，因此添加更多参与者而不配置池以托管更多线程会降低性能。请记住，actor 模式是一种用于并发通信的工具，而不是用于并行计算
-
@MustafaSimav 在四核上仍然有 2 个演员而不是 1 个演员可能会显示出加速。 Actor 可用于并行计算以及并发通信。
标签： performance scala parallel-processing akka