上篇我们讨论了Akka-Remoting。我们说Akka-Remoting是一种点对点的通讯方式,能使两个不同JVM上Akka-ActorSystem上的两个Actor之间可以相互沟通。Akka-Remoting还没有实现完全的Actor位置透明(location transparency),因为一个Actor还必须在获得对方Actor确切地址信息后才能启动与之沟通过程。Akka-Remoting支持“远程查找”和“远程构建”两种沟通方式。由于篇幅所限,我们只介绍了“远程查找”。在这一篇里我们将会讨论“远程构建”方式。
同样,我们先通过项目结构来分析:
lazy val local = (project in file(".")).settings(commonSettings).settings(name := "remoteCreateDemo").aggregate(calculator,remote).dependsOn(calculator)lazy val calculator = (project in file("calculator")).settings(commonSettings).settings(name := "calculator")lazy val remote = (project in file("remote")).settings(commonSettings).settings(name := "remoteSystem").aggregate(calculator).dependsOn(calculator)
远程构建的过程大致是这样的:由local通知remote启动构建Actor;remote从本地库中查找Actor的类定义(class)并把它载入内存。由于驱动、使用远程Actor是在local进行的,所以local,remote项目还必须共享Calculator,包括Calculator的功能消息。这项要求我们在.sbt中用aggregate(calculator)来协同编译。
我们把Calculator的监管supervisor也包括在这个源码文件里。现在这个calculator是个包括监管、功能、消息的完整项目了。Calculator源代码如下:
package remoteCreation.calculatorimport akka.actor._
import scala.concurrent.duration._object Calcultor {sealed trait MathOpscase class Num(dnum: Double) extends MathOpscase class Add(dnum: Double) extends MathOpscase class Sub(dnum: Double) extends MathOpscase class Mul(dnum: Double) extends MathOpscase class Div(dnum: Double) extends MathOpssealed trait CalcOpscase object Clear extends CalcOpscase object GetResult extends CalcOpsdef props = Props(new Calcultor)def supervisorProps = Props(new SupervisorActor)
}class Calcultor extends Actor with ActorLogging {import Calcultor._var result: Double = 0.0 //internal stateoverride def receive: Receive = {case Num(d) => result = dcase Add(d) => result += dcase Sub(d) => result -= dcase Mul(d) => result *= dcase Div(d) =>val _ = result.toInt / d.toInt //yield ArithmeticExceptionresult /= dcase Clear => result = 0.0case GetResult =>sender() ! s"Result of calculation is: $result"}override def preRestart(reason: Throwable, message: Option[Any]): Unit = {log.info(s"Restarting calculator: ${reason.getMessage}")super.preRestart(reason, message)}
}class SupervisorActor extends Actor {def decider: PartialFunction[Throwable,SupervisorStrategy.Directive] = {case _: ArithmeticException => SupervisorStrategy.Resume}override def supervisorStrategy: SupervisorStrategy =OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds){decider.orElse(SupervisorStrategy.defaultDecider)}val calcActor = context.actorOf(Calcultor.props,"calculator")override def receive: Receive = {case msg@ _ => calcActor.forward(msg)}}
与上一个例子的”远程查找式“相同,remote需要为Remoting公开一个端口。我们可以照搬.conf配置文件内容:remote/src/main/resources/application.conf
akka {actor {provider = remote}remote {enabled-transports = ["akka.remote.netty.tcp"]netty.tcp {hostname = "127.0.0.1"port = 2552}log-sent-messages = onlog-received-messages = on} }
由于远程构建和使用是在local上进行的,在remote上我们只需要启动ActorSystem就行了:
import com.typesafe.config.ConfigFactory
import akka.actor._object CalculatorRunner extends App {val remoteSystem = ActorSystem("remoteSystem",ConfigFactory.load("application"))println("Remote system started.")scala.io.StdIn.readLine()remoteSystem.terminate()}
Calculator的构建是在localSystem上启动的,我们需要在配置文件中描述远程构建标的(还是未能实现位置透明):local/src/main/resources/application.conf
akka {actor {provider = remote,deployment {"/calculator" {remote = "akka.tcp://remoteSystem@127.0.0.1:2552"}}}remote {netty.tcp {hostname = "127.0.0.1",port=2554}}
}
注意:上面这个/calculator设置实际上指的是SupervisorActor。
现在我们可以在local上开始构建calculator,然后使用它来运算了:
import akka.actor._
import remoteCreation.calculator.Calcultor._
import scala.concurrent.duration._
import akka.pattern._object RemotingCreate extends App {val localSystem = ActorSystem("localSystem")val calcActor = localSystem.actorOf(props,name = "calculator") //created SupervisorActorimport localSystem.dispatchercalcActor ! ClearcalcActor ! Num(13.0)calcActor ! Mul(1.5)implicit val timeout = akka.util.Timeout(1 second)((calcActor ? GetResult).mapTo[String]) foreach printlnscala.io.StdIn.readLine()calcActor ! Div(0.0)calcActor ! Div(1.5)calcActor ! Add(100.0)((calcActor ? GetResult).mapTo[String]) foreach printlnscala.io.StdIn.readLine()localSystem.terminate()}
从代码上看构建calculator(SupervisorActor)过程与普通的Actor构建没分别,所有细节都放在配置文件里了。但是,要注意actorOf的name必须与配置文档中的设置匹配。
试运行结果与上一个例子相同。值得注意的是实际远程构建的是一个SupervisorActor。Calculator的构建是SupervisorActor构建的其中一部分。从运算结果看:这个SupervisorActor也实现了它的功能。
下面是这次示范的源代码:
local/build.sbt
azy val commonSettings = seq (name := "RemoteCreateDemo",version := "1.0",scalaVersion := "2.11.8",libraryDependencies := Seq("com.typesafe.akka" %% "akka-actor" % "2.5.2","com.typesafe.akka" %% "akka-remote" % "2.5.2")
)lazy val local = (project in file(".")).settings(commonSettings).settings(name := "remoteCreateDemo").aggregate(calculator).dependsOn(calculator)lazy val calculator = (project in file("calculator")).settings(commonSettings).settings(name := "calculator")lazy val remote = (project in file("remote")).settings(commonSettings).settings(name := "remoteSystem").aggregate(calculator).dependsOn(calculator)
calculator/calculator.scala
package remoteCreation.calculatorimport akka.actor._
import scala.concurrent.duration._object Calcultor {sealed trait MathOpscase class Num(dnum: Double) extends MathOpscase class Add(dnum: Double) extends MathOpscase class Sub(dnum: Double) extends MathOpscase class Mul(dnum: Double) extends MathOpscase class Div(dnum: Double) extends MathOpssealed trait CalcOpscase object Clear extends CalcOpscase object GetResult extends CalcOpsdef props = Props(new Calcultor)def supervisorProps = Props(new SupervisorActor)
}class Calcultor extends Actor with ActorLogging {import Calcultor._var result: Double = 0.0 //internal stateoverride def receive: Receive = {case Num(d) => result = dcase Add(d) => result += dcase Sub(d) => result -= dcase Mul(d) => result *= dcase Div(d) =>val _ = result.toInt / d.toInt //yield ArithmeticExceptionresult /= dcase Clear => result = 0.0case GetResult =>sender() ! s"Result of calculation is: $result"}override def preRestart(reason: Throwable, message: Option[Any]): Unit = {log.info(s"Restarting calculator: ${reason.getMessage}")super.preRestart(reason, message)}
}class SupervisorActor extends Actor {def decider: PartialFunction[Throwable,SupervisorStrategy.Directive] = {case _: ArithmeticException => SupervisorStrategy.Resume}override def supervisorStrategy: SupervisorStrategy =OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds){decider.orElse(SupervisorStrategy.defaultDecider)}val calcActor = context.actorOf(Calcultor.props,"calculator")override def receive: Receive = {case msg@ _ => calcActor.forward(msg)}}
remote/src/main/resources/application.conf
akka {actor {provider = remote}remote {enabled-transports = ["akka.remote.netty.tcp"]netty.tcp {hostname = "127.0.0.1"port = 2552}log-sent-messages = onlog-received-messages = on}
}
remote/CalculatorRunner.scala
package remoteCreation.remote
import com.typesafe.config.ConfigFactory
import akka.actor._object CalculatorRunner extends App {val remoteSystem = ActorSystem("remoteSystem",ConfigFactory.load("application"))println("Remote system started.")scala.io.StdIn.readLine()remoteSystem.terminate()}
local/src/main/resources/application.conf
akka {actor {provider = remote,deployment {"/calculator" {remote = "akka.tcp://remoteSystem@127.0.0.1:2552"}}}remote {netty.tcp {hostname = "127.0.0.1",port=2554}}
}
local/RemotingCreation.scala
import akka.actor._
import remoteCreation.calculator.Calcultor._
import scala.concurrent.duration._
import akka.pattern._object RemotingCreate extends App {val localSystem = ActorSystem("localSystem")val calcActor = localSystem.actorOf(props,name = "calculator") //created SupervisorActor
import localSystem.dispatchercalcActor ! ClearcalcActor ! Num(13.0)calcActor ! Mul(1.5)implicit val timeout = akka.util.Timeout(1 second)((calcActor ? GetResult).mapTo[String]) foreach printlnscala.io.StdIn.readLine()calcActor ! Div(0.0)calcActor ! Div(1.5)calcActor ! Add(100.0)((calcActor ? GetResult).mapTo[String]) foreach printlnscala.io.StdIn.readLine()localSystem.terminate()}