逸言

Akka中的Remote Actor

| Comments

Akka的设计目标就是为分布式准备的，因此所有Actor之间的交互都是通过消息，且所有动作都是异步的。这种做法就是为了确保Akka的所有功能无论是在单独的JVM，还是包含了成百上千机器的Cluster，都是可用的。

然而，本地与分布式总是存在区别，主要牵涉到两点：

  • 消息需要支持序列号；

  • 消息传递的可靠性问题；

为了保证本地处理与分布式处理的透明化，Akka几乎没有特别为Remoting Layer提供专门的API，区别仅在于配置。开发者只需遵循Actor设计的原则，然后在配置中指定Actor子树的远程配置即可。当然，在代码层面，Akka也提供了唯一与众不同的API，就是在包含部署信息的Props中，可以允许设置Deploy实例。不过，这件事情是可以配置的。倘若两者都设置了，配置文件优先。

若要支持Scale up，Akka提供了多个Actor子树以支持并行处理。然后以多种方式进行路由。开发者唯一要做的事情是声明一个确定的Actor作为“withRouter”，事实上就是创建一个路由Actor，它能够生成所需类型Children的数量值，该数量值是可以配置的。

Remote Actor

要调用Akka的Remote Actor，则需要对Remote Actor进行部署。首先，我们可以通过Akka的官方网站下载Akka的库。我这里下载的是2.2.3版本。下载后解压。与部署有关的目录包括bin和deploy。在bin目录下是运行Akka的脚本。而在deploy目录下，除了Readme文件外，此时为空。

现在，我们可以编写一个Remote Actor。Akka要求我们定义一个Actor，以及它对应的Application。Remote Actor与普通的Actor定义没有什么区别，例如：

package com.agiledon.akka

import akka.actor.Actor

class RemoteActor extends Actor {
  def receive = {
    case message: String =>
      sender ! message + "got something"
  }
}

但对于Application而言，则要求派生自akka kernel下的Bootable。akka kernel是单独的一个包，并没有包含在akka-actor中。因此需要定义依赖，例如使用sbt：

scalaVersion := "2.10.2"

libraryDependencies += "com.typesafe.akka" % "akka-actor_2.10" % "2.2.3"

libraryDependencies += "com.typesafe.akka" % "akka-kernel_2.10" % "2.2.3"

定义的Application如下所示：

package com.agiledon.akka

import akka.kernel.Bootable
import akka.actor.{Props, ActorSystem}
import com.typesafe.config.ConfigFactory


class RemoteNodeApplication extends Bootable {
  val system = ActorSystem("RemoteNodeApp", ConfigFactory.load().getConfig("RemoteSys"))

  def startup = {
    system.actorOf(Props[RemoteActor], name = "remoteActor")
  }

  def shutdown = {
    system.shutdown()
  }
}

我们需要实现startup与shutdown方法。而在Application中，会加载配置文件application.conf的配置信息创建ActorSystem。配置如下：

RemoteSys {
    akka {
        actor {
            provider = "akka.remote.RemoteActorRefProvider"
        }
        remote {
            enabled-transports = ["akka.remote.netty.tcp"]
            netty.tcp {
                hostname = "192.168.3.34"
                port = 2552
            }
        }
    }
}

端口号2552也是Akka的默认端口号。

部署Remote Actor

application.conf文件应该放到resources目录下。编译打包。然后将编译得到的jar包拷贝到akka的deploy目录下。然后转移到bin目录下，执行akka脚本（windows下是执行akka.bat)： ./akka com.agiledon.akka.RemoteNodeApplication

akka命令后面是application类的full name。如果一切正常，就可以显示如下界面：

Client(Local) Actor

要调用部署了的Remote Actor，客户端的Actor可以通过将Remote Actor的address传递给actorSelection()方法（之前的版本为actorFor，目前已经被Deprecated），以此来获得Remote Actor的Ref。如下所示：

package com.agiledon.akka

import akka.actor.{ActorLogging, Actor}
import akka.util.Timeout
import akka.pattern.ask
import scala.concurrent.Await
import scala.concurrent.duration._

class LocalActor extends Actor with ActorLogging {
  val remoteActor = context.actorSelection("akka.tcp://RemoteNodeApp@192.168.3.34:2552/user/remoteActor")
  implicit val timeout = Timeout(5 seconds)

  def receive = {
    case message: String =>
      val future = (remoteActor ? message).mapTo[String]
      val result = Await.result(future, timeout.duration)
      log.info("Message received from server -> {}", result)
  }
}

同样需要定义配置文件，但无需指定hostname与port了：

LocalSys {
  akka {
      actor {
          provider = "akka.remote.RemoteActorRefProvider"
      }
  }
}

编写Application，使其可以被运行以调用Remote Actor：

package com.agiledon.akka

import akka.actor.{Props, ActorSystem}
import com.typesafe.config.ConfigFactory

object LocalNodeApplication extends App {
  val system = ActorSystem("LocalNodeApp", ConfigFactory.load().getConfig("LocalSys"))

  val localActor = system.actorOf(Props[LocalActor], name = "localActor")
  localActor ! "hello demo actor"

  Thread.sleep(4000)
  system.shutdown()

}

在客户端模块的根目录下创建build.sbt文件：


name := "actor-local"

version := "1.0"

scalaVersion := "2.10.2"

libraryDependencies += "com.typesafe.akka" % "akka-actor_2.10" % "2.2.3"

libraryDependencies += "com.typesafe.akka" % "akka-remote_2.10" % "2.2.3"

转移到客户端模块的根目录下，执行sbt命令，进入SBT。执行compile命令编译客户端模块actor-local。然后执行run命令，若运行成功，即可看到如下信息：

[info] Running com.agiledon.akka.LocalNodeApplication
[INFO] [02/18/2014 19:11:47.461] [run-main] [Remoting] Starting remoting
[INFO] [02/18/2014 19:11:47.725] [run-main] [Remoting] Remoting started; listening on addresses :[akka.tcp://LocalNodeApp@192.168.3.38:2552]
[INFO] [02/18/2014 19:11:48.066] [LocalNodeApp-akka.actor.default-dispatcher-3] [akka://LocalNodeApp/user/localActor] Message received from server -> hello demo actor got something

Comments