我在Scala 2.11.8上使用akka 2.4.7 .
事件来自extractShardId和extractEntityId，但它们没有被传播到actor的receive方法 . 有什么想法可能是什么？
https://bitbucket.org/kuzelac/apt-billing/overview分支价格碎片
定义了一个集群分片：
val dailyPriceAggregateActor: ActorRef = ClusterSharding(context.system).start(
typeName = "DailyPriceAggregateActor",
entityProps = DailyPriceAggregateActor(),
settings = ClusterShardingSettings(context.system),
extractEntityId = DailyPriceAggregateActor.extractEntityId,
extractShardId = DailyPriceAggregateActor.extractShardId)
actor对象是
object DailyPriceAggregateActor {
def apply() = Props(classOf[DailyPriceAggregateActor])
val extractEntityId: ShardRegion.ExtractEntityId = {
case e@DailyPriceSaved(userId, unitId, _, _, _) => (s"$userId$unitId", e)
case e@LookupPriceForDay(userId, unitId, _) => (s"$userId$unitId", e)
}
val extractShardId: ShardRegion.ExtractShardId = {
case _ => "one"
}
}
我把conf设置为
akka.actor.provider = "akka.cluster.ClusterActorRefProvider"
Actor由Persistence Query播种
def startSync(actor: ActorRef) = {
val queries = PersistenceQuery(context.system).readJournalFor[ScalaDslMongoReadJournal](MongoReadJournal.Identifier)
val src: Source[EventEnvelope, NotUsed] =
queries.eventsByPersistenceId(PriceAggregateActor.persistenceId, 0L, Long.MaxValue)
src.runForeach(actor ! _.event)
}
1 回答
忘了添加种子节点