Akka入门Scala版(Idea,SBT)

Akka

异步处理框架，自己Google

 

前期准备：

idea12（http://www.jetbrains.com/idea/）

在插件库中下载SBT,SCALA插件（Intellij Idea->Preferences...->Plugins)

下载SBT0(0.12)必须这么版本因为SBT-IDEA这个插件需要这个版本；

1.设置SBT:

 设置IDE-SETING中的SBT-LAUNCHER的路径即可

 

实例介绍：

计算圆周率Pi,算法如下：

算法实现：假设N为100000，我们可以把它分成多段比如10000一个处理线程，多个线程并发处理，最后把这十个线程的结果加起来就是Pi的值。在分布式的环境中，我们可以把这些线程放在不同机器上，同时并发处理，这样会获得更高的性能。在Akka里面要实现上述逻辑是相当简单的。下面我们来详细看看怎么在Idea环境中编写上诉代码。

 

项目搭建

1.新建Scala项目，设置项目路径和ScalaHome的路径

3.在项目根目录下创建buid.sbt文件，内容如下：

name := "hello"

version := "1.0"

scalaVersion := "2.9.2"

resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/"

libraryDependencies += "com.typesafe.akka" % "akka-actor" % "2.0.2"

 注意：每一行内容需要空一行，否则编译不错

 

2.增加插件，sbt-idea.

项目目录下新建文件夹：project,在project目录下新建文件plugins.sbt,内容如下：

addSbtPlugin("com.github.mpeltonen" % "sbt-idea" % "1.2.0")

3.点开左下角的SBT控制台，点击绿色的运行按钮，请耐心等待，SBT将会下载一堆有的没得，不知道什么原因在我的环境在居然搞了快2个小时

4.运行gen-idea命令，sbt将会生成IDEA的项目文件，重新载入项目，至此SCALA+SBT的环境就搭建好了，可以开始编码了。我在运行时还碰到一个小问题，就是hello-build模块的编译输出路径和测试输出路径一样，导致scala无法编译通过，只需要修改Scala的编译测试输出路径即可。

 

回到刚才的例子：

我将会用两个Actor来完成这个工作。Worker(工作线程，完成分段计算）,Master(主线程，完成分段线程的划分，调度，结果合并，打印），下面来看看代码：

package main.scala

import akka.actor.{ActorSystem, Props, Actor}
import akka.routing.RoundRobinRouter

/**
 * Created with IntelliJ IDEA.
 * User: criss
 * Date: 13-2-21
 * Time: 上午1:37
 * To change this template use File | Settings | File Templates.
 */
object Pi extends App {
    //Actor传递消息特质，所有的消息都实现这个特质
    sealed trait PieMessage
    //工作线程的运行指令，start:分段开始的值，nrOfElements:本段的数目
    case class Work(start: Int, nrOfElements: Int) extends PieMessage
    //工作线程结果返回指令
    case class Result(value:Double) extends PieMessage
    //主线程运行指令
    case object Calculate extends PieMessage

    caculate(10000,10,10000)

    class Worker extends Actor {

        def caculate(start:Int,nf:Int) = {
            var acc:Double = 0.0
            //计算start到start+nf的数值，详见上面公式
            for (i <- start until start+nf)
            {
                acc +=  4.0 * (1 - (i % 2) * 2) / (2 * i + 1)
            }

            acc
        }

        def receive = {
            //处理运行指令
            case Work(start, nfOfElement) =>
            //返回给主线程结果
            sender !  Result(caculate(start,nfOfElement))
        }
    }
    //nrOfMesseage 工作线程数
    //nfWorks工作线程调度池的大小
    //nrOfElements每个工作线程计算的数值大小
    class Master(nrOfMesseage:Int,nfWorks:Int,nrOfElements:Int) extends Actor
    {
        //最后的结果
        var pi:Double = _
        //计数，计算已经运行完多少个Worker
        var nrOfResults:Int = 0
        //你可以理解成一个线程池，用于调度Worker
        val workerRouter = context.actorOf(
            Props[Worker].withRouter(RoundRobinRouter(nfWorks)),name="workers"
        )
        //存储开始时间
        var start:Long  = _
        def receive = {
            case Calculate =>
                start = System.currentTimeMillis;
                for (i <- 0 until nrOfMesseage)
                    workerRouter ! Work(i*nrOfElements,nrOfElements)
            case Result(value) =>
                //当有一个结果返回时，合并结果
                pi += value
                //计算当前返回的结果总数
                nrOfResults += 1
                //当全部结果返回时，打印结果退出
                if(nrOfResults == nrOfMesseage)
                {
                    println(pi)
                    println("耗时:"+(System.currentTimeMillis()-start))
                    context.system.shutdown()
                }

        }
    }

    def caculate(nrOfMesseage:Int,nfWorks:Int,nrOfElements:Int)
    {
       //初始化ActorSystem
        val system = ActorSystem("PiSystem")
       //初始化主线程
        val master = system.actorOf(Props(new Master(nrOfMesseage
        ,nfWorks,nrOfElements)),name="master")
        //发送计算指令
        master ! Calculate
    }



}

 

 

 

 

评论

1 楼 itstarting 2014-07-10  

谢谢分享。

需要指出的是，代码中有两个小错误：

1. 在build.sbt中，需要导入akka-routing
libraryDependencies += "com.typesafe.akka" %% "akka-routing" % "2.0.2"

2. 需要把RoundRobinRouter改为RoundRobinPool，因为此类已被重新命名