如何解决在重新平衡或集群重启后以不同的行为启动实体actor
val numberOfShards = 6
val extractEntityId: ShardRegion.ExtractEntityId = {
case msg: NewSubscribe => (msg.id.toString,msg)
case msg: Unsubscribe => (msg.id.toString,msg)
}
val extractShardId: ShardRegion.ExtractShardId = {
case msg: NewSubscribe => (msg.id % numberOfShards).toString
case msg: Unsubscribe => (msg.id % numberOfShards).toString
case ShardRegion.StartEntity(id) =>
logger.info(s"Starting old subscription with id $id")
(id.toInt % numberOfShards).toString
}
val settings = ClusterShardingSettings(system)
val cluster: ClusterSharding = ClusterSharding(system)
val strategy: ShardAllocationStrategy = cluster.defaultShardAllocationStrategy(settings)
val counterRegion: ActorRef = cluster.start(
typeName = SubscribeActor.name,entityProps = Props(new SubscribeActor()),settings = settings,extractEntityId = extractEntityId,extractShardId = extractShardId,allocationStrategy = strategy,handOffStopMessage = SubscribeActor.FinishSubscribe()
)
当我使用命令 NewSubscribe
或 Unsubscribe
打开新订阅或关闭已经运行的订阅时,一切正常。但是在重新平衡之后,我想用不同的命令启动旧的实体 actor,例如 RestoreSubscription
。到目前为止,我发现了这样的东西
case ShardRegion.StartEntity(id) =>
logger.info(s"Starting old subscription with id $id")
(id.toInt % numberOfShards).toString
但是没有办法发送任何命令。我知道我可以使用 PersistentActor
来保存订阅的状态,然后我可以从中确定它是旧订阅还是新订阅,但是有没有更简单的方法来做到这一点?
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。