我有一个当前接受命令/消息的演员系统。这些actors的状态是持久化的Akka.Persistance。我们现在要为这个actor系统构建查询系统。基本上我们的问题是我们希望有一种方法来获取这些特定参与者的所有状态的聚合/列表。虽然我并没有严格遵守 CQRS 模式，但我认为这可能是一种很好的解决方法。
我最初的想法是让一个用于查询的参与者作为其状态的一部分保存正在执行“数据写入”的其他参与者的状态聚合。为了做到这一点，这个演员将订阅它感兴趣的演员，这些演员只会在他们经历某种状态变化时向查询演员发送他们的状态。这是解决这个问题的方法吗？有没有更好的方法来做到这一点？
【问题讨论】：
我对实现这种模式的建议是在此处为您的演员使用 pub-sub 和 push-and-pull 消息传递的组合。
对于每个“聚合”，此 Actor 应该能够订阅来自您要查询的各个子 Actor 的事件。每当孩子的状态发生变化时，都会将一条消息推送到所有订阅的聚合中，并且每个聚合的状态都会自动更新。
当一个新的聚合上线并需要检索它错过的状态（从它存在之前）它应该能够从每个孩子那里提取当前状态并使用它来构建它的当前状态，使用来自孩子的增量更新前进以保持其对子状态的聚合视图一致。
这是我用于此类工作的模式，并且开箱即用地在本地运行良好。通过网络，您可能必须确保可交付性保证，这通常很容易做到。你可以在这里阅读更多关于如何做到这一点的信息：https://petabridge.com/blog/akkadotnet-at-least-once-message-delivery/
【讨论】：
一些 Akka.Persistence 后端（即使用 SQL 的后端）也实现了称为 Akka.Persistence.Query 的东西。它允许您订阅生成的事件流，并将其用作Akka.Streams 语义的源。
如果您使用 SQL 日志，则需要 Akka.Persistence.Query.Sql 和 Akka.Streams 包。从那里您可以为特定参与者创建一个实时（这意味着不断更新）事件源，并将其用于您喜欢的任何操作，即打印它们：
using (var system = ActorSystem.Create("system"))
using (var materializer = system.Materializer())
{
var queries = Sys.ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier)
queries.EventsByPersistenceId("<persistence-id>", 0, long.MaxValue)
.Select(envelope => envelope.Event)
.RunForEach(e => Console.WriteLine(e), materializer);
}
【讨论】：
EventsByPersistenceId 是实时查询。这意味着，直到达到最大 sequenceNr 才会完成。如果只想查询目前存储的数据，使用CurrentEventsByPersistenceId（或其他以Current...前缀开头的方法）。