Akka Singleton - 不接受消息

Question

是我愚蠢 - 我没有将索引器 Prop 传递到系统创建中。如果有人从中受益，我会在这里留下答案 *

我正在创建一个单例并发送这样的消息:

 val indexerProps = ClusterSingletonManager.props(had => Props(
            classOf[SingleCoreIndexer], dataProvider, publisher, name), name, End, None)

        val coreIndexer = system.actorOf(indexerProps, name)
        //val coreIndexer = system.actorOf(Props(classOf[SingleCoreIndexer], dataProvider, publisher, name))

        coreIndexer ! "start_indexing"

注释掉的行显示了工作正常的非单例 Prop

当我运行应用程序时，我收到以下错误:

[WARN] [06/21/2013 11:55:32.443] [deadcoreindexerstest-akka.actor.default-dispatcher-5] [akka://deadcoreindexerstest/user/node1] unhandled event start_indexing in state Start

所有其他功能都停止工作，这与暗示“coreIndexer”actor 未收到“start_indexing”消息的消息相关

更多代码:

class Indexer(systemCreator: SystemCreator, publisherProps: Props, dataProviderProps: Props, name: String) {

    def start {
        val system = systemCreator.create
        val dataProvider = system.actorOf(dataProviderProps)
        val publisher = system.actorOf(publisherProps)

        val indexerProps = ClusterSingletonManager.props(
            singletonProps = had => Props(classOf[SingleCoreIndexer], dataProvider, publisher, name),
            singletonName = "aaa",
            terminationMessage = End,
            role = None
            )

        val coreIndexer = system.actorOf(Props(classOf[SingleCoreIndexer], dataProvider, publisher, name))
        coreIndexer ! "start_indexing"
    }
}



class SingleCoreIndexer(dataProvider: ActorRef, publisher: ActorRef, name: String) extends Actor {

    def receive = {

        case "start_indexing" => {
            println("Single core indexer starting indexing")
            dataProvider ! new NextBatchOfDataPlease
        }

        case BatchOfData(data) => {
            publisher ! (name, data)
            self ! "next_batch"
        }

        case "next_batch" => {
            dataProvider ! new NextBatchOfDataPlease
        }
    }
}

看起来我正在向经理而不是单例发送消息。但是，当我向单例发送消息时，没有任何 react :

class Indexer(systemCreator: SystemCreator, publisherProps: Props, dataProviderProps: Props, name: String) {

    def start {
        val system = systemCreator.create
        val dataProvider = system.actorOf(dataProviderProps)
        val publisher = system.actorOf(publisherProps)

        val indexerProps = ClusterSingletonManager.props(
            singletonProps = had => Props(classOf[SingleCoreIndexer], dataProvider, publisher, name),
            singletonName = "singlecoreindexer",
            terminationMessage = End,
            role = None
            )

        system.actorOf(Props(classOf[SingleCoreIndexer], dataProvider, publisher, name))
        val coreIndexer = system.actorSelection(s"/user/$name/singlecoreindexer")
        coreIndexer ! "start_indexing"
    }
}

Answer 1

我知道这已经解决了，但即使有这里提供的信息，我仍然花了一段时间才弄清楚如何向单例发送消息，并认为我会留下我在这里找到的东西

Cluster Singleton未很好解释的两个关键概念文档是:

使用 ClusterSingletonManager.props 创建的 Actor 是实际实例的父级，

你应该只使用那个 Actor 的地址来创建一个 ClusterSingletonProxy

集群中的每个节点都将创建单例管理器，最终获胜的最老的节点是您实际想要与之交谈的单例的父节点。 ClusterSingletonProxy确保您正在与实际代理交谈，并且即使单例暂时不可用或迁移到另一个节点，您也始终在与正确的实例交谈。

鉴于这些信息，代码应该是:

class Indexer(systemCreator: SystemCreator, publisherProps: Props, dataProviderProps: Props, name: String) {

  def start {
    val system = systemCreator.create
    val dataProvider = system.actorOf(dataProviderProps)
    val publisher = system.actorOf(publisherProps)

    val indexerProps = ClusterSingletonManager.props(
      singletonProps = had => Props(classOf[SingleCoreIndexer], dataProvider, publisher, name),
      singletonName = "singlecoreindexer",
      terminationMessage = End,
      role = None
    )

    val singletonManager = system.actorOf(
      Props(classOf[SingleCoreIndexer],dataProvider, publisher, name)
    )

    val indexerPath = (singletonManager.path / name)
    val coreIndexer = system.actorOf(
      ClusterSingletonProxy.props(indexerPath, None),
      s"$name-proxy"
    )

    coreIndexer ! "start_indexing"
  }
}