Akka ClusterSingletonProxy 到远程部署的单例

Question

我正在尝试通过另一个参与者向部署在远程节点上的单例参与者发送消息。

这是等待 memberUp 事件的管理器，然后在该节点上部署 Worker 演员，然后向单例发送消息：

object Manager extends App {
  val sys = ActorSystem("mySys", ConfigFactory.load("application").getConfig("manager"))
  sys.actorOf(Props[Manager], "manager")
}

class Manager extends Actor with ActorLogging {
  override def receive: Receive = {
    case MemberUp(member) if member.address != Cluster(context.system).selfAddress =>
      context.system.actorOf(ClusterSingletonManager.props(
        singletonProps = Props(classOf[Worker]),
        singletonName = "worker",
        terminationMessage = End,
        role = Some("worker")).withDeploy(Deploy(scope = RemoteScope(member.address))))

      context.actorOf(ClusterSingletonProxy.props(
        singletonPath = s"/user/singleton/worker",
        role = Some(s"worker")), "worker") ! "hello"
  }
  override def preStart(): Unit = {
    Cluster(context.system).subscribe(self,classOf[MemberUp])
  }
}

这是工人：

object Worker extends App{
  ActorSystem("mySys", ConfigFactory.load("application").getConfig("worker"))
}

class Worker extends Actor with ActorLogging {

  override def receive: Receive = {
    case msg =>
      println(s"GOT MSG : $msg from : ${sender().path.name}")
  }
}

还有application.conf：

manager {
  akka {
    actor {
      provider = "akka.cluster.ClusterActorRefProvider"
    }
    cluster {
      auto-down-unreachable-after = 20s
      seed-nodes = [
        "akka.tcp://mySys@127.0.0.1:2552"
      ]
      roles.1 = "manager"

    }
    remote.netty.tcp.port = 2552

  }
}

worker {
  akka {
    cluster {
      auto-down-unreachable-after = 20s
      seed-nodes = [
        "akka.tcp://mySys@127.0.0.1:2552"
      ]
      roles.1 = "worker"
    }
    remote.netty.tcp.port = 2554
    actor {
      provider = "akka.cluster.ClusterActorRefProvider"
    }
  }
}

worker 已初始化（我可以在日志中看到 state change [Start -> Oldest] 消息），但从 manager 发送的消息永远不会到达 worker。当我在远程节点上部署单例时，它曾经工作正常，但现在我希望管理器部署它。

我也尝试将其部署为管理器的子节点（使用上下文而不是 context.system）并将单例路径更改为 user/manager/singleton/worker，但没有成功。

我正在使用 Akka 2.3.11

编辑： sbt 文件：

name := "MyProject"
version := "1.0"
scalaVersion := "2.10.5"
libraryDependencies +=
    "com.typesafe.akka" %% "akka-actor" % "2.3.11",
    "com.typesafe.akka" %% "akka-cluster" % "2.3.11",
    "joda-time" % "joda-time" % "2.0",
    "com.typesafe.akka" %% "akka-contrib" % "2.3.11"

Answer 1

所以我尝试了创建ClusterSingletonManagers 的不同选项，我认为远程部署它们会破坏单例模式中的某些东西。我为此收集了一些指标：

• 由于是远程部署，工作节点上ClusterSingletonManager 的路径为/remote/akka.tcp/mySys@127.0.0.1:2552/user/worker。我认为图书馆不能/不会处理这个问题，因为它需要/user/worker

• 当尝试使用 ClusterSingletonProxy 从主节点发送消息时，登录 DEBUG 模式状态 No singleton available, stashing message hello worker 和 Trying to identify singleton at akka.tcp://mySys@127.0.0.1:2552/user/worker/singleton（失败并重试）-> 它正在寻找单例错误的节点，因为没有可用的管理器，并且显然不知道单例在工作节点上。

直接在工作节点上创建ClusterSingletonManager 时，一切都按预期工作。

您对经理的命名也有问题。你的singletonName 是worker 而你的经理本身（演员）没有任何名字。创建代理时，您使用路径/user/singleton/worker，但路径应如下所示：/user/{actorName}/{singletonName}。所以在我的代码中，我使用worker 作为actorName，singleton 作为singletonName。

这是我的工作代码：

object Manager extends App {
  val sys = ActorSystem("mySys", ConfigFactory.load("application").getConfig("manager"))
  sys.actorOf(Props[Manager], "manager")
}

class Manager extends Actor with ActorLogging {
  override def receive: Receive = {
    case MemberUp(member) if member.address != Cluster(context.system).selfAddress =>
      context.actorOf(ClusterSingletonProxy.props(
        singletonPath = s"/user/worker/singleton",
        role = Some("worker")), name = "workerProxy") ! "hello worker"
  }
  override def preStart(): Unit = {
    Cluster(context.system).subscribe(self,classOf[MemberUp])
  }
}

---

object Worker extends App{
  val sys = ActorSystem("mySys", ConfigFactory.load("application").getConfig("worker"))

  sys.actorOf(ClusterSingletonManager.props(
    singletonProps = Props(classOf[Worker]),
    singletonName = "singleton",
    terminationMessage = PoisonPill,
    role = Some("worker")), name = "worker")
}

class Worker extends Actor with ActorLogging {

  override def receive: Receive = {
    case msg =>
      println(s"GOT MSG : $msg from : ${sender().path.name}")
  }
}

application.conf 和 build.sbt 保持不变。

编辑

通过使用工作节点上的实际路径引用ClusterSingletonProxy（计算它是网络路径）来使用它。我不确定我是否会推荐这个，因为我仍然不确定该库是否被设计为能够做到这一点，但它至少在这个最小的例子中有效：

object Manager extends App {
  val sys = ActorSystem("mySys", ConfigFactory.load("application").getConfig("manager"))
  sys.actorOf(Props[Manager], "manager")
}

class Manager extends Actor with ActorLogging {
  override def receive: Receive = {
    case MemberUp(member) if member.address != Cluster(context.system).selfAddress =>
      val ref = context.system.actorOf(ClusterSingletonManager.props(
        singletonProps = Props(classOf[Worker]),
        singletonName = "singleton",
        terminationMessage = PoisonPill,
        role = Some("worker")).withDeploy(Deploy(scope = RemoteScope(member.address))), name = "worker")

      context.actorOf(ClusterSingletonProxy.props(
        singletonPath = s"${ref.path.toStringWithoutAddress}/singleton", // /remote/akka.tcp/mySys@127.0.0.1:2552/user/worker/singleton
        role = Some("worker")), name = "workerProxy") ! "hello worker"
  }
  override def preStart(): Unit = {
    Cluster(context.system).subscribe(self,classOf[MemberUp])
  }
}

---

object Worker extends App{
  val sys = ActorSystem("mySys", ConfigFactory.load("application").getConfig("worker"))
}

class Worker extends Actor with ActorLogging {

  override def receive: Receive = {
    case msg =>
      println(s"GOT MSG : $msg from : ${sender().path.name}")
  }
}