【Akka】Akka入门编程实例

引言

这篇文章主要是第一次学习Akka编程，先试试水，探探坑，对Akka和SBT的使用有一个直观的了解，以几个简单的akka编程实例来说明akka的使用。希望在日后的学习和编程中，能有更多自己的体会和经验总结来分享。

Actor模型

Actor实例可以想象成是服务器上的Web服务，你无法控制，只能通过发送消息去请求执行任务或查询信息，而不能直接在Web服务中修改状态或者处理资源。通过发送不可改变的消息，虽然看上去有些限制，但是可以很简单安全的编写并发程序。

Actor系统的形象理解

处理故障

Actor模型的一个重要内容是处理故障。在工作工程中，如果出现错误或者抛出异常，actor和其子actor都将暂停，然后发送一条信息给监督者（supervisor）actor，报告出现故障的信号。根据工作任务和故障的性质，监督者actor将会作出几种选择：

Hello,Actor实例

现在我用一个最简单的actor编程实例来介绍akka编程，先给出代码：

importakka.actor.Actor importakka.actor.ActorSystem importakka.actor.Props  classHelloActorextendsActor{ defreceive = { case"hello"=> println("hello back to you.") case_ => println("huh?")  } }  objectTest1_HelloActorextendsApp{ // actor need an ActorSystem valsystem = ActorSystem("HelloSystem") // create and start the actor valhelloActor = system.actorOf(Props[HelloActor], name="helloActor") // send two messages  helloActor ! "hello"  helloActor ! "what" // shutdown the actor system  system.shutdown } 

代码注解：

ActorSystem

一个actor system是actors的层级集团，分享公共配置信息（比如分发器dispatchers，部署deployments，远程功能remote capabilities，地址addresses）。它同时也是创建和查询actors的入口。ActorSystem是为你的应用程序分配线程资源的结构。

ActorRef

这个引用用来处理actor，你可以将其看做是处理实际actor的代理人（broker）或包装外观（facade）。ActorRef防止你破坏Actor模型，比如直接处理Actor实例，或直接修改Actor实例中的变量。所以只能通过给actor发送消息方式来执行任务，这种“袖手旁观（不干涉，hands-off）”的方法帮助巩固适宜的编程实践。

ActorRef有以下特点：

Actor之间的通信实例

下面给出的是两个actor实例相互发送消息进行通信的PingPong示例：

importakka.actor._  caseobjectPingMessage caseobjectPongMessage caseobjectStartMessage caseobjectStopMessage  classPing(pong: ActorRef)extendsActor{ varcount =0 defincrementAndPrint {count +=1; println(s"$count:ping")} defreceive = { caseStartMessage =>  incrementAndPrint  pong ! PongMessage casePingMessage =>  incrementAndPrint if(count >99) {  sender ! StopMessage  println("ping stopped")  context.stop(self)  } else  sender ! PongMessage case_ => println("Ping got unexpected information")  } }  classPongextendsActor{ varcount =0 defreceive = { caseStopMessage =>  println("pong stopped")  context.stop(self) casePongMessage =>  count += 1  println(s"$count:pong")  sender ! PingMessage case_ => println("Pong got unexpected information")  } }  objectPingPangTestextendsApp{ valsystem = ActorSystem("PingPongTest") valpongActor = system.actorOf(Props[Pong], name="pong") valpingActor = system.actorOf(Props(newPing(pongActor)),  name = "ping")  pingActor ! StartMessage } 

代码注释：

启动Actor

importakka.actor._  caseclassCreateChild(name: String) caseclassName(name: String)  classChildextendsActor{ varname ="No name" overridedefpostStop: Unit = {  println(s"D'oh! They killed me ($name): ${self.path}")  } defreceive = { caseName(name) =>this.name = name case_ => println(s"Child $name got message.")  } }  classParentextendsActor{ defreceive = { caseCreateChild(name) => // Parent creates a new Child here  println(s"Parent about to create Child ($name) ...") valchild = context.actorOf(Props[Child], name=s"$name")  child ! Name(name) case_ => println(s"Parent got some other message.")  } }  objectParentChildDemoextendsApp{ valactorSystem = ActorSystem("ParentChildTest") valparent = actorSystem.actorOf(Props[Parent], name="Parent")  // send messages to Parent to create to child actors  parent ! CreateChild("XiaoMing")  parent ! CreateChild("XiaoLiang")  Thread.sleep(500)  // lookup XiaoMing, the kill it  println("Sending XiaoMing a PoisonPill ... ") valxiaoming = actorSystem.actorSelection("/user/Parent/XiaoMing")  xiaoming ! PoisonPill  println("XiaoMing was killed")   Thread.sleep(5000)  actorSystem.shutdown } 

打印结果：

Parent about to create Child (XiaoMing) ... Parent about to create Child (XiaoLiang) ... Sending XiaoMing a PoisonPill ... XiaoMing was killed D'oh! They killed me (XiaoMing): akka://ParentChildTest/user/Parent/XiaoMing D'oh! They killed me (XiaoLiang): akka://ParentChildTest/user/Parent/XiaoLiang 

终止Actor

actor终止的相关处理

actor的终止分两步： 第一步actor将停止对邮箱的处理，向所有子actor发送终止命令，然后处理来自子actor的终止消息直到所有的子actor都完成终止， 最后终止自己(调用postStop,销毁邮箱,向DeathWatch发布Terminated,通知其监管者)。这个过程保证actor系统中的子树以一种有序的方式终止,将终止命令传播到叶子结点并收集它们回送的确认消息给被终止的监管者。如果其中某个actor没有响应(i.e.由于处理消息用了太长时间以至于没有收到终止命令), 整个过程将会被阻塞。

在 ActorSystem.shutdown被调用时, 系统根监管actor会被终止，以上的过程将保证整个系统的正确终止。

postStop() hook是在actor被完全终止以后调用的。这是为了清理资源:

overridedefpostStop() = { // 关闭文件或数据库连接 } 

PoisonPill和gracefulStop

如果你想等待终止过程的结束，或者组合若干actor的终止次序，可以使用gracefulStop。下面给出gracefulStop的代码示例：

importakka.actor._ importakka.pattern.gracefulStop importscala.concurrent.{Await, ExecutionContext, Future} importscala.concurrent.duration._ importscala.language.postfixOps  caseobjectTestActorStop  classTestActorextendsActor{ defreceive = { caseTestActorStop =>  context.stop(self) case_ => println("TestActor got message")  } overridedefpostStop {println("TestActor: postStop")} }  objectGracefulStopTestextendsApp{ valsystem = ActorSystem("GracefulStopTest") valtestActor = system.actorOf(Props[TestActor], name="TestActor") // try to stop the actor graceful try{ valstopped: Future[Boolean] = gracefulStop(testActor,2seconds, TestActorStop)  Await.result(stopped, 3seconds)  println("testActor was stopped")  } catch{ casee: akka.pattern.AskTimeoutException => e.printStackTrace  } finally{  system.shutdown  } } 

Kill消息

importakka.actor._  classNumber5extendsActor{ defreceive = { case_ => println("Number 5 got a message")  } overridedefpreStart { println("Number 5 is alive")} overridedefpostStop { println("Number 5::postStop called")} overridedefpreRestart(reason: Throwable, message: Option[Any]): Unit = {  println("Number 5::preRestart called")  } overridedefpostRestart(reason: Throwable): Unit = {  println("Number 5::postRestart called")  } }  objectKillTestextendsApp{ valsystem = ActorSystem("KillTestSystem") valnumber5 = system.actorOf(Props[Number5], name="Number5")  number5 ! "hello"  number5 ! Kill  system.shutdown } 

打印的信息：

Number5isalive Number5gotamessage [ERROR][01/17/2016 19:20:09.342][KillTestSystem-akka.actor.default-dispatcher-3][akka://KillTestSystem/user/Number5]Kill(akka.actor.ActorKilledException) Number5::postStopcalled