java - 使用 akka 搜索网络中的节点。

Question

有没有一种方法可以搜索网络中的节点，以便如果任何节点出现故障，我们可以在它恢复正常时重新连接它。目前我正在使用 mdns，但我想用 akka 来实现这一点。我正在使用 akka 和 java。

Answer 1

您可以在此处使用两种方法:

1) 外部 - 运行 consul.io 的实例与集群一起。通过读取 consul 集群的状态，您将了解连接到该集群的所有机器 - 这将使您对节点及其 IP 地址的总集有所了解。现在，在任何地方使用 JMX 检索 Akka 集群的状态 - 您正在此处查找 akka:type=Cluster bean 及其属性 Members或集群状态。这些将返回集群状态 - 因此您将知道 akka 集群中可用的节点集。如果从 consul 集群中的节点中减去 Akka 集群中的节点集 - 您将得到那些被 Akka 集群仲裁删除/隔离的节点，并且大多数可能这些必须重新启动。

2) 内部 - 一旦我有 added Akka 远程处理的 ThisActorSystemQuarantinedEvent 。当节点从集群中删除时触发此事件，然后将其置于隔离状态 - 因此它无法自动重新连接到集群。所以你可以创建这样的东西:

  object ClusterWatcher {

    private class DefaultClusterWatcher extends Actor with ClusterWatcher {
      override def receive = handleQuarantinedRestart
    }

    def registerRestartJVMWatcherActor(sys: ActorSystem) = {
      val ref = sys.actorOf(Props[DefaultClusterWatcher])
      sys.eventStream.subscribe(ref, classOf[ThisActorSystemQuarantinedEvent])
    }

  }

  trait ClusterWatcher {

    _: Actor ⇒

    def handleQuarantinedRestart: Actor.Receive = {
      case err: ThisActorSystemQuarantinedEvent ⇒
        import SlackNotification._
        slackEmergency(s"Actor system ${err.localAddress} is quarantined by ${err.remoteAddress}, restarting")
        sys.exit(1)
    }

  }

并在创建 Actor 系统后使用它:

val sys = ActorSystem("MyCluster", config)
ClusterWatcher.registerRestartJVMWatcherActor(sys)

然后，只要 DefaultClusterWatcher 收到隔离事件 - 它将重新启动整个 JVM(或对 Actor 系统执行其他操作 - 由您决定)。

我发现这些方法的组合是最强大的 - 在我的设置中，它允许我在具有适度网络连接的不同数据中心运行由 30 个节点组成的自维持分布式集群。