我有一个使用Akka 2.2.4的系统,它创建了一堆本地actor并将它们设置为广播路由器的路由.根据我们传递的一些哈希范围,每个工作者处理总工作的某些部分.它很棒.
现在,我必须将此应用程序集群以进行故障转移.基于在集群上只存在/触发每个散列范围一个工作者的要求,在我看来,将每个作为ClusterSingletonManager设置将是有意义的……然而,我无法使其工作. actor系统启动,它创建了ClusterSingletonManager,它将下面引用的代码中的路径添加到广播路由器,但它从来没有实例化我的实际工作者演员来处理我的消息.我得到的只是一条日志消息：“处于状态Start的未处理事件${my message}”.我究竟做错了什么？启动这个单实例集群还需要做些什么吗？我向错误的演员发送了一条消息吗？
这是我的akka配置(我使用默认配置作为后备)：
akka{
cluster{
roles=["workerSystem"]
min-nr-of-members = 1
role {
workerSystem.min-nr-of-members = 1
}
}
daemonic = true
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = ${akkaPort}
}
}
actor{
provider = akka.cluster.ClusteractorRefProvider
single-message-bound-mailBox {
# FQCN of the MailBoxType. The Class of the FQCN must have a public
# constructor with
# (akka.actor.ActorSystem.Settings, com.typesafe.config.Config) parameters.
mailBox-type = "akka.dispatch.BoundedMailBox"
# If the mailBox is bounded then it uses this setting to determine its
# capacity. The provided value must be positive.
# NOTICE:
# Up to version 2.1 the mailBox type was determined based on this setting;
# this is no longer the case, the type must explicitly be a bounded mailBox.
mailBox-capacity = 1
# If the mailBox is bounded then this is the timeout for enqueueing
# in case the mailBox is full. Negative values signify infinite
# timeout, which should be avoided as it bears the risk of dead-lock.
mailBox-push-timeout-time = 1
}
worker-dispatcher{
type = Pinneddispatcher
executor = "thread-pool-executor"
# Throughput defines the number of messages that are processed in a batch
# before the thread is returned to the pool. Set to 1 for as fair as possible.
throughput = 500
thread-pool-executor {
# Keep alive time for threads
keep-alive-time = 60s
# Min number of threads to cap factor-based core number to
core-pool-size-min = ${workerCount}
# The core pool size factor is used to determine thread pool core size
# using the following formula: ceil(available processors * factor).
# Resulting size is then bounded by the core-pool-size-min and
# core-pool-size-max values.
core-pool-size-factor = 3.0
# Max number of threads to cap factor-based number to
core-pool-size-max = 64
# Minimum number of threads to cap factor-based max number to
# (if using a bounded task queue)
max-pool-size-min = ${workerCount}
# Max no of threads (if using a bounded task queue) is determined by
# calculating: ceil(available processors * factor)
max-pool-size-factor = 3.0
# Max number of threads to cap factor-based max number to
# (if using a bounded task queue)
max-pool-size-max = 64
# Specifies the bounded capacity of the task queue (< 1 == unbounded)
task-queue-size = -1
# Specifies which type of task queue will be used, can be "array" or
# "linked" (default)
task-queue-type = "linked"
# Allow core threads to time out
allow-core-timeout = on
}
fork-join-executor {
# Min number of threads to cap factor-based parallelism number to
parallelism-min = 1
# The parallelism factor is used to determine thread pool size using the
# following formula: ceil(available processors * factor). Resulting size
# is then bounded by the parallelism-min and parallelism-max values.
parallelism-factor = 3.0
# Max number of threads to cap factor-based parallelism number to
parallelism-max = 1
}
}
}
}
这是我创建Actors的地方(它是用Groovy编写的)：
Props clusteredProps = ClusterSingletonManager.defaultProps("worker".toString(), PoisonPill.getInstance(), "workerSystem",
new ClusterSingletonPropsFactory(){
@Override
Props create(Object handOverData) {
log.info("called in ClusterSingetonManager")
Props.create(WorkerActorCreator.create(applicationContext, it.start, it.end)).withdispatcher("akka.actor.worker-dispatcher").withMailBox("akka.actor.single-message-bound-mailBox")
}
} )
ActorRef manager = system.actorOf(clusteredProps, "worker-${it.start}-${it.end}".toString())
String path = manager.path().child("worker").toString()
path
当我尝试向实际的工作者发送消息时,上面的路径应该解决吗？目前它没有.
我究竟做错了什么？此外,这些actor位于Spring应用程序中,并且worker actor设置了一些@Autowired依赖项.虽然这种Spring集成在非集群环境中运行良好,但是在集群环境中是否有任何问题需要我注意？
谢谢
仅供参考：我也在akka-user google群组中发布了此内容.这是link.
解决方法:
代码中的路径是您在具有角色“workerSystem”的每个节点上启动的ClusterSingletonManager actor.它将在集群中最旧的节点上创建名为“worker – ${it.start} – ${it.end}”的子actor(WorkerActor),即集群中的单例.
您还应该定义ClusterSingletonManager的名称,例如system.actorOf(clusteredProps,“workerSingletonManager”).
您无法将消息发送到ClusterSingletonManager.您必须将它们发送到活动工作程序的路径,即包括最旧节点的地址.这可以通过documentation中的ConsumerProxy来说明.
我不确定你应该使用单身人士.所有工作人员都将在最早的同一节点上运行.我更愿意在akka-user谷歌小组讨论您的问题的替代解决方案.
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
