2

我正在尝试创建一个 Akka 分片集群。我想在其中一个节点上使用仅代理模式，以便将消息路由到分片区域。我收到以下警告：

[WARN] [02/11/2019 17:04:17.819] [ClusterSystem-akka.actor.default-dispatcher-21] [akka.tcp://ClusterSystem@127.0.0.1:2555/system/sharding/ShardnameProxy] Trying to register to coordinator at [Some(ActorSelection[Anchor(akka.tcp://ClusterSystem@127.0.0.1:2551/), Path(/system/sharding/ShardnameCoordinator/singleton/coordinator)])], but no acknowledgement. Total [1] buffered messages.

**Main.java: ** 使用 application.conf 中的配置启动集群（代码稍后添加）

object Main {
  val shardName = "Shardname"
  val role = "Master"
  var shardingProbeLocalRegin: Option[ActorRef] = None
  def main(args: Array[String]): Unit = {
    val conf = ConfigFactory.load()
    val system = ActorSystem("ClusterSystem",conf.getConfig("main"))
    ClusterSharding(system).start(shardName,Test.props,ClusterShardingSettings(system),ShardDetails.extractEntityId,ShardDetails.extractShardId)
  }
}

Test.java：分片集群的实体

object Test {
  def props: Props = Props(classOf[Test])

  class Test extends Actor {
    val log = Logger.getLogger(getClass.getName)


    override def receive = {
      case msg: String =>
        log.info("Message from " + sender().path.toString + " Message is " + msg)
        sender() ! "Done"

    }
  }

}

MessageProducer.java(Proxy Only Mode)消息生产者每秒向 Shard 发送一条消息。

object MessageProducer {

  var shardingProbeLocalRegin: Option[ActorRef] = None
  object DoSharding
  def prop:Props = Props(classOf[MessageProducer])
  var numeric : Long = 0
  def main(args: Array[String]): Unit = {
    val conf = ConfigFactory.load
    val system = ActorSystem("ClusterSystem",conf.getConfig("messgaeProducer"))
    ClusterSharding(system).startProxy(Main.shardName,None,extractEntityId,extractShardId)
    shardingProbeLocalRegin  = Some(ClusterSharding(system).shardRegion(Main.shardName))
    val actor = system.actorOf(Props[MessageProducer],"message")
  }
}

class RemoteAddressExtensionImpl(system: ExtendedActorSystem) extends Extension {
  def address = system.provider.getDefaultAddress
}

object RemoteAddressExtension extends ExtensionKey[RemoteAddressExtensionImpl]

class MessageProducer extends Actor{
  val log = Logger.getLogger(getClass.getName)


  override def preStart(): Unit = {
    println("Starting "+self.path.address)
    context.system.scheduler.schedule(10 seconds,1 second ,self,DoSharding)
  }


  override def receive = {
    case DoSharding =>
      log.info("sending message" + MessageProducer.numeric)
      MessageProducer.shardingProbeLocalRegin.foreach(_ ! "" + (MessageProducer.numeric))
      MessageProducer.numeric += 1

  }

}

**application.conf: ** 配置文件

    main {
      akka {
        actor {
          provider = "akka.cluster.ClusterActorRefProvider"
        }

        remote {
          log-remote-lifecycle-events = on

          netty.tcp {
            hostname = "127.0.0.1"
            port = 2551
          }

        }

        cluster {
          seed-nodes = [
            "akka.tcp://ClusterSystem@127.0.0.1:2551"
          ]

          sharding.state-store-mode = ddata
          auto-down-unreachable-after = 1s
        }

        akka.extensions = ["akka.cluster.metrics.ClusterMetricsExtension", "akka.cluster.ddata.DistributedData"]

      }
    }
messgaeProducer {
  akka {
    actor {
      provider = "akka.cluster.ClusterActorRefProvider"
    }

    remote {
      log-remote-lifecycle-events = on

      netty.tcp {
        hostname = "192.168.2.96"
        port = 2554
      }

    }

    cluster {
      seed-nodes = [
        "akka.tcp://ClusterSystem@127.0.0.1:2551"
        //, "akka.tcp://ClusterSystem@127.0.0.1:2552"
      ]

      sharding.state-store-mode = ddata
      auto-down-unreachable-after = 1s
    }

    akka.extensions = ["akka.cluster.metrics.ClusterMetricsExtension", "akka.cluster.ddata.DistributedData"]


  }
}

我做错什么了吗？有没有其他方法可以申请这种方法。我的主要目标是避免集群出现单点故障。如果任何节点发生故障，那么它不应该影响任何其他状态。谁能帮我这个？

4

1 回答 1

1

解决了吗？如果没有，请检查您的 akka.cluster 配置。你必须像这样设置配置。它对我有用

代理

akka.cluster {
  roles = ["Proxy"]
  sharding {
    role = "Master"
  } 
}

为主人

akka.cluster {
  roles = ["Master"]
  sharding {
    role = "Master"
  } 
}
于 2020-03-27T06:43:19.667 回答