后端微服务【免费下载链接】akka.netCanonical actor model implementation for .NET with local distributed actors in C# and F#.项目地址https://gitcode.com/gh_mirrors/ak/akka.net点击查看免费下载导读Akka.NET Persistence Query 是 Akka 持久化模块的“读侧”扩展它以 Akka Streams 的异步流SourceT, TMat为统一查询接口让各个 Journal 插件Cassandra、ADO.NET 关系库、内存 Journal 等以一致的方式暴露自己的查询能力。本文将以 docs/articles/persistence/persistence-query.md 为主线完整讲解 Read Journal 的获取方式、内置的预定义查询类型、事件标记Tagging与 Event Adapter 机制、查询结果封装EventEnvelope、Materialized Values、以及把事件投影到读侧存储的三种实战模式并穿插Akka.Persistence.Query源码实现src/core/Akka.Persistence.Query与官方内存实现Akka.Persistence.Query.InMemory作为底层原理佐证。读完本文你将能够在自己的 Actor 系统中获取 Read Journal、发起实时/一次性事件查询、为事件打标签并消费按标签聚合的事件流以及构建可断点续传的读侧投影。什么是 Persistence Query设计概览Persistence Query 是对 Akka Persistence写侧的补充。它的核心价值在于提供一个通用的、基于异步流Reactive Streams的查询接口让不同的 Journal 插件可以按自己的技术能力实现并暴露查询功能。最典型的使用场景是 CQRSCommand Query Responsibility Segregation命令与查询职责分离架构中的query side读侧写侧应用使用 Akka Persistence 持久化领域事件读侧Persistence Query 负责把写侧的数据“迁移”到读侧数据库例如关系型库、Elasticsearch、专用报表库。需要强调的是Persistence Query 本身并不是应用的读侧它是连接写侧与读侧之间的事件流通道。在非常简单的场景下它可能足以满足应用的查询需求但在 CQRS 精神下强烈建议在需求增长时把读写两侧拆分到各自独立的存储中。注意每个 Read Journal 插件必须显式文档化自己支持哪些查询类型及其语义例如流的完成策略、是否支持实时推送、顺序保证等。使用前请查阅你所选插件的文档。从设计上讲Akka Persistence Query 被刻意设计成一个非常松散、宽泛的 API。原因在于不同的 Journal 技术能力差异巨大——SQL Journal 可以利用复杂 SQL 查询而支持实时事件流的 Journal 则能暴露“事件即来即推”的流。松散接口让每个 Journal 都能发挥自己的最佳特性而不是被统一接口束缚。这在 Interfaces.cs 的接口注释中体现得很清楚“The purpose of the API is not to enforce compatibility between different journal implementations, because the technical capabilities may be very different.”因此虽然 Persistence Query 定义了若干最常见的预定义查询类型Journals 被建议实现但并不强制各插件仍然可以自由定义自己的专属查询方法。获取 Read JournalPersistenceQuery.Get与ReadJournalFor要发起查询第一步是获得一个IReadJournal实例。Read Journal 以社区插件形式提供每个插件对应一个具体的数据存储例如 Cassandra 或 ADO.NET 数据库。给定一个提供akka.persistence.query.my-read-journal插件的库获取 Journal 并查询事件非常简单var actorSystem ActorSystem.Create(query); // 通过插件 id 获取读 journal var readJournal PersistenceQuery.Get(actorSystem) .ReadJournalForSqlReadJournal(akka.persistence.query.my-read-journal); // 向 journal 发起查询 SourceEventEnvelope, NotUsed source readJournal .EventsByPersistenceId(user-1337, 0, long.MaxValue); // 物化流消费事件 var mat ActorMaterializer.Create(actorSystem); source.RunForeach(envelope { Console.WriteLine($event {envelope}); }, mat);从源码看PersistenceQuery.Get(ActorSystem)是一个 Akka 扩展IExtension通过PersistenceQueryProvider注册到 ActorSystem并且内部使用ConcurrentDictionarystring, IReadJournal按插件 id缓存已创建的 Read Journal 实例参见 PersistenceQuery.cs。ReadJournalForTJournal(string readJournalPluginId)会依次尝试从缓存中直接返回已创建的实例从 HOCON 配置中按插件 id 读取class字段通过反射实例化对应的IReadJournalProvider调用provider.GetReadJournal()得到真正的 Read Journal并缓存。ReadJournalFor要求泛型参数TJournal : IReadJournal且运行时类型必须能赋给该接口否则抛出ArgumentException。此外还提供了一个 ActorSystem 扩展方法便于链式调用var readJournal actorSystem.ReadJournalForSqlReadJournal(akka.persistence.query.my-read-journal);插件 id 的约定IdentifierJournal 实现者被鼓励把插件配置路径放入一个用户可见的静态常量中这样用户可以直接通过ReadJournalForSqlReadJournal(SqlReadJournal.Identifier)访问无需硬编码字符串虽然这不是强制要求。官方内存实现即遵循此约定——InMemoryReadJournal.Identifier akka.persistence.query.journal.inmem见 InMemoryReadJournal.cs测试代码中的DummyReadJournal.Identifier akka.persistence.query.journal.dummy也是如此见 DummyReadJournal.cs。在禁用动态类型加载如 Native AOT 裁剪场景时仓库还支持通过Akka.Persistence.Hosting在代码中注册 Read JournalReadJournalDetailsPersistenceQuery会优先使用注册信息而不是 HOCONclass反射参见 ReadJournalDetails.cs 与 PersistenceQuery.cs 的CreatePlugin逻辑。预定义查询类型Akka Persistence Query 内置了一组查询接口并建议 Journal 实现者按照下述语义实现。注意这些查询类型虽然非常常见但 Journal 并不被要求全部实现——例如某个 Journal 中某类查询可能效率极低。具体支持哪些务必以所用插件文档为准。所有接口均定义在 Interfaces.cs它们都继承自空标记接口IReadJournal。PersistenceIds与CurrentPersistenceIdsIPersistenceIdsQuery用于获取所有持久化 Actor 的persistenceId列表var queries PersistenceQuery.Get(actorSystem) .ReadJournalForSqlReadJournal(akka.persistence.query.my-read-journal); var mat ActorMaterializer.Create(actorSystem); Sourcestring, NotUsed src queries.AllPersistenceIds();注AllPersistenceIds即接口中的PersistenceIds()方法源码中IPersistenceIdsQuery暴露的方法名就是PersistenceIds()见 Interfaces.cs返回Sourcestring, NotUsed。语义要点返回流无序多次执行查询可能得到不同顺序。实时语义PersistenceIds流在到达当前已存在的persistenceId末尾时不会完成而是持续推送新建持久化 Actor 的persistenceId。对应的“到达末尾即完成”版本是CurrentPersistenceIds。轮询粒度查询侧通过分批拉取事件来周期性地发现新的persistenceId延迟最多可达配置的refresh-interval或传入的RefreshInterval提示。失败语义如果后端 Journal 执行查询失败流将以失败failure完成。EventsByPersistenceId与CurrentEventsByPersistenceId用于获取某个特定PersistentActor由persistenceId标识的事件var queries PersistenceQuery.Get(actorSystem) .ReadJournalForSqlReadJournal(akka.persistence.query.my-read-journal); var mat ActorMaterializer.Create(actorSystem); var src queries.EventsByPersistenceId(some-persistence-id, 0L, long.MaxValue); Sourceobject, NotUsed events src.Select(c c.Event);语义要点序列号区间通过fromSequenceNr与toSequenceNr可以只取部分事件传0L与long.MaxValue表示取全部。每个事件的序列号都包含在EventEnvelope.SequenceNr中因此可以随时从某个序列号断点续传。顺序保证返回流按序列号排序即与PersistentActor持久化事件的顺序一致。多次执行时返回相同顺序的前缀元素除非事件已被删除。实时语义EventsByPersistenceId流到达当前存储末尾后不完成持续推送新持久化的事件对应的一次性版本是CurrentEventsByPersistenceId。时效性写 Journal 会在事件持久化后尽快通知查询侧但出于效率考虑查询侧以批处理方式取数延迟最多可达refresh-interval。失败语义后端 Journal 查询失败时流以 failure 完成。EventsByTag与CurrentEventsByTag跨 Actor 的事件聚合EventsByTag允许跨persistenceId查询事件这是实现 DDD 中“按聚合根类型查询所有领域事件”的关键能力。该查询在部分 Journal 中实现困难或需要对存储做额外准备才能高效执行因此支持与否、如何支持必须查阅所用 Read Journal 插件文档。部分 Journal 支持通过Event Adapter把事件包装进Akka.Persistence.Journal.Tagged来附带标签其他 Journal 也可能支持别的打标签方式例如把事件以 JSON 存储时按字段检索。下面是一个写侧打标签的IWriteEventAdapter示例public class MyTaggingEventAdapter : IWriteEventAdapter { private ImmutableHashSetstring colors ImmutableHashSet.Create(green, black, blue); public string Manifest(object evt) { return string.Empty; } public object ToJournal(object evt) { var str evt as string; if (str ! null) { var tags colors.Aggregate( ImmutableHashSetstring.Empty, (acc, c) str.Equals(c) ? acc.Add(c): acc); if (tags.IsEmpty) return evt; else return new Tagged(evt, tags); } else { return evt; } } }Tagged定义在 Tagged.cs它是一个携带Payload真正的领域事件与TagsIImmutableSetstring不可变标签集合的包装结构Journal 会解包并把Payload存储。IWriteEventAdapter接口EventAdapters.cs要求实现Manifest(object evt)返回清单/类型提示不需要时返回空字符串与ToJournal(object evt)1 对 1 转换不允许丢弃事件完整的适配器还包括IReadEventAdapter.FromJournal用于读侧还原。更多细节可参考 docs/articles/persistence/event-adapters.md。查询“所有带蓝色标签的事件”并使用 Akka Streams 组合算子处理// 假设 journal 支持数值偏移可以这样 SourceEventEnvelope, NotUsed blueThings readJournal.EventsByTag(blue, 0L); // 找出前 10 个蓝色事件 TaskImmutableHashSetobject top10BlueThings blueThings .Select(c c.Event) .Take(10) // 拉取 10 个元素后取消查询流 .RunAggregate( ImmutableHashSetobject.Empty, (acc, c) acc.Add(c), mat); // 从已知偏移继续发起新查询 var furtherBlueThings readJournal.EventsByTag(blue, offset: 10);API 兼容性提示上述示例沿用了官方文档中“long偏移”的写法。需要说明的是当前仓库 Offset.cs 中IEventsByTagQuery.EventsByTag与IAllEventsQuery.AllEvents的签名已经是SourceEventEnvelope, NotUsed EventsByTag(string tag, Offset offset)——偏移参数为抽象类型Offset可通过Offset.NoOffset()全部事件、Offset.Sequence(value)数值偏移、Offset.TimeBasedUuid(guid)时间排序 UUID或Offset.FromEnd(count)从末尾倒数第 N 个构造详见下文“Offset 类型”一节。旧文档中的long偏移是历史版本形态具体以你所用的 Akka.NET 版本与插件文档为准。语义要点顺序弱保证跨多个persistenceId的查询如EventsByTag很少能保证事件在流中出现顺序稳定也不保证多次物化之间顺序一致。Journal 可以自行选择严格排序但必须明确文档化其顺序保证——例如关系型数据库上“按时间戳升序、与persistenceId无关”很容易实现而在纯 KV 存储上高效实现则很困难。可续传偏移内置EventsByTag提供可选的Offset参数Journal 可据此实现可续传流例如用 WHERE 子句从特定行开始读或在可按插入时间排序的存储中把 Long 当作时间戳只取旧事件。实时与一次性不需要实时流时使用CurrentEventsByTag它在到达当前结果集末尾时立即完成。AllEvents与CurrentAllEventsAllEvents允许回放并监控 Journal 中存储的全部事件无论它们属于哪个persistenceId// 假设 journal 支持数值偏移可以这样 SourceEventEnvelope, NotUsed allEvents readJournal.AllEvents(offset: 0L); // 回放最先存储的 10 个事件 TaskImmutableHashSetobject first10Things allEvents .Select(c c.Event) .Take(10) // 拉取 10 个元素后取消查询流 .RunAggregate( ImmutableHashSetobject.Empty, (acc, c) acc.Add(c), mat); // 从已知偏移继续发起新查询 var next10Things readJournal.AllEvents(offset: 10);语义要点实时语义AllEvents流到达最后一条已记录事件后不完成持续推送新持久化的事件CurrentAllEvents则在查询发起时已持久化的最后一条事件处完成。低延迟写 Journal 会在新事件产生后立即通知查询侧该查询不涉及周期性轮询或分批与基于refresh-interval的查询不同。顺序弱保证与EventsByTag相同跨persistenceId的事件顺序很少被保证或多次物化之间不稳定Journal 若提供严格顺序须显式文档化。同样支持可选的Offset参数用于实现可续传流用法与EventsByTag一致。Offset 类型流的位置标记Offset是EventsByTag、AllEvents等查询的起点参数抽象也作为EventEnvelope.Offset出现在每个事件上用于断点续传。当前仓库Offset.cs提供以下实现类型语义说明NoOffset从头开始用于获取全部事件Offset.NoOffset()单例Sequence(long)数值序列偏移有序且排他exclusive序列号等于该值的事件不会被包含因此可直接把EventEnvelope.Offset回传给下一次查询实现续传TimeBasedUuid(Guid)时间排序 UUID 偏移同样是排他的有序偏移适合分布式存储中按插入时间定位FromEnd(int count)从末尾倒数相对偏移仅作查询输入、永不回传在EventEnvelope中物化时解析为具体起点对实时查询是最佳努力best-effort不支持的 Journal 会抛出异常其中Sequence的排他性值得特别注意你在EventEnvelope.Offset中拿到的是“下一个起点”把它原样传回查询即可避免事件重复或遗漏。Sequence.CompareTo只允许与同类型比较混用不同Offset类型会抛出InvalidOperationException。EventEnvelope事件流中的元数据载体所有预定义查询返回的元素都是EventEnvelopeEventEnvelope.cs它在事件本体之外附加了关键元数据属性类型含义OffsetOffset该事件在流中的位置标记可用于断点续传PersistenceIdstring持久化该事件的 Actor 标识SequenceNrlong事件在persistenceId内的序列号Eventobject领域事件本体Journal 已解包Tagged后的载荷Timestamplong事件被存储的时间ticks与DateTime.Now.Ticks同基准Tagsstring[]事件携带的标签自 1.5.11 起提供ToString()输出形如EventEnvelope(persistenceId:..., seqNr:..., offset:..., event:..., timestamp:...)。Timestamp与Tags不参与Equals/GetHashCode保持向后兼容。Materialized Values让 Journal 提供流级元信息Akka Streams 的物化值Materialized Value机制允许在流物化时暴露额外信息。更高级的查询 Journal 可以利用它向用户暴露所物化流的特征——例如流是有限还是无限、是否严格有序等。物化值类型由返回Source的第二个类型参数决定Journal 可以据此返回专属的查询对象。示例定义富事件与查询元数据并让查询返回SourceRichEvent, QueryMetadatapublic class RichEvent { public RichEvent(ISetstring tags, object payload) { Tags tags; Payload payload; } public ISetstring Tags { get; } public object Payload { get; } } public class QueryMetadata { public QueryMetadata(bool deterministicOrder, bool infinite) { DeterministicOrder deterministicOrder; Infinite infinite; } public bool DeterministicOrder { get; } public bool Infinite { get; } }public SourceRichEvent, QueryMetadata ByTagsWithMeta(ISetstring tags) { }var query readJournal.ByTagsWithMeta(ImmutableHashSet.Create(red, blue)); query .MapMaterializedValue(meta { Console.WriteLine( $The query is: ordered deterministically: {meta.DeterministicOrder}, infinite: {meta.Infinite}); return meta; }) .Select(evt { Console.WriteLine($Event payload: {evt.Payload}); return evt; }) .RunWith(Sink.IgnoreRichEvent(), mat);MapMaterializedValue会在物化时拿到QueryMetadata并打印其特征之后照常消费事件。这是 Read Journal 向使用者“自我描述”能力边界的标准手法。性能与反规范化把事件投影到读侧存储构建 Event Sourcing CQRS 系统时必须认识到写侧与读侧的需求完全不同写侧优先保证写入吞吐。例如竞拍系统中“接受出价”需要尽快响应竞拍者因此写侧往往选择可水平扩展但查询表达能力较弱的数据存储。读侧可能需要复杂查询能力SQL、报表分析、统计视图。分析师可能要用这些数据制定竞拍策略与趋势分析这通常需要把写侧数据**投影project**到读优化的数据存储中。这里的Materialized View指“查询结果的某种持久化存储”视图被创建一次之后被多次查询——相比直接查源事件这种形态可能更高效或更有意义。下面给出三种投影模式。模式一投影到兼容 Reactive Streams 的数据存储如果读侧数据库驱动暴露了 Reactive Streams 接口投影就是一个“把 Read Journal 的流接到数据库驱动的订阅者”的简单问题var system ActorSystem.Create(MySystem); var mat ActorMaterializer.Create(system); var readJournal PersistenceQuery.Get(system).ReadJournalForMyReadJournal(JournalId); ISubscriberIImmutableListobject dbBatchWriter new ReactiveStreamsCompatibleDBDriver.BatchWriter(); // 使用示例的Reactive Streams数据库驱动 readJournal .EventsByPersistenceId(user-1337) .Select(envelope envelope.Event) .Select(ConvertToReadSideTypes) // 转换为读侧数据类型 .Grouped(20) // 每 20 条一批插入 .RunWith(Sink.FromSubscriber(dbBatchWriter), mat); // 批量写入读侧数据库模式二使用SelectAsync编写无状态投影如果目标数据库没有提供可写数据的 Reactive StreamsSubscriber就需要用普通函数或 Actor 实现写入逻辑。当写入逻辑无状态、只需把事件从一种类型转换为另一种再写入时public class ExampleStore { public Taskobject Save(object evt) { return Task.FromResult(evt); } }var store new ExampleStore(); readJournal .EventsByTag(bid, 0L) .SelectAsync(1, e store.Save(e)) .RunWith(Sink.Ignoreobject(), mat);SelectAsync(1, ...)表示并发度为 1的异步映射——即保证写入顺序与事件到达顺序一致这对保证读侧一致性很重要如果存储能容忍乱序可适当提高并发度以换取吞吐。模式三可续传投影Resumable Projections很多投影不需要每次从“时间起点”重跑而是从上次处理到的位置继续。这个模式并未内置于框架但实现起来很简单把已处理事件的序列号或 offset持久化下次启动时从该位置恢复。下面的例子同时演示了用 Actor 承担读侧写入逻辑当需要复杂聚合时并保存进度var timeout new Timeout(TimeSpan.FromSeconds(3)); var bidProjection new MyResumableProjection(bid); var writerProps Props.Create(typeof(TheOneWhoWritesToQueryJournal), bid); var writer system.ActorOf(writerProps, bid-projection-writer); readJournal .EventsByTag(bid, bidProjection.LatestOffset ?? 0L) .SelectAsync(8, envelope writer.Ask(envelope.Event, timeout).ContinueWith(t envelope.Offset, TaskContinuationOptions.OnlyOnRanToCompletion)) .SelectAsync(1, offset bidProjection.saveProgress(offset)) .RunWith(Sink.Ignoreobject(), mat); public class TheOneWhoWritesToQueryJournal(id: String) : ActorBase { public TheOneWhoWritesToQueryJournal(string id) {} private DummyStore _store new DummyStore(); private ComplexState _state ComplexState(); protected override bool Receive(object message) { _state UpdateState(_state, message); if (_state.IsReadyToSave()) _store.Save(new Record(_state)); return true; } private ComplexState UpdateState(ComplexState state, object msg) { // 这里做复杂的聚合逻辑 ... return state; } }关键点查询从bidProjection.LatestOffset ?? 0L开始——首次运行从头开始之后从上次进度续跑第一段SelectAsync(8, ...)并发 8 向写 Actor 发送事件成功后回传envelope.Offset即已完成事件的位置第二段SelectAsync(1, offset bidProjection.saveProgress(offset))串行保存进度保证进度单调推进、不落后于已写入的事件写 Actor 内部可维护复杂状态UpdateState聚合足够多的事件后一次性落库批量投影的典型形态。这套组合是 Akka.NET 社区实现 Event Sourcing 读侧投影的成熟骨架并发写 串行存进度兼顾吞吐与一致性。配置Read Journal 插件的配置项定义在其配置 section 中绝对路径与插件标识identifier一一对应。以官方内存实现为例reference.confakka.persistence.query.journal.inmem { # InMemory ReadJournalProvider 的实现类 class Akka.Persistence.Query.InMemory.InMemoryReadJournalProvider, Akka.Persistence.Query.InMemory # 该查询 journal 所连接的写 journal 插件配置的绝对路径。 # 若未定义或为 则连接 akka.persistence.journal.plugin 指定的默认 journal。 write-plugin refresh-interval 3s # 一次查询回放取多少条事件并缓存到向下游投递为止。 max-buffer-size 100 }参数说明classIReadJournalProvider的实现类型全名PersistenceQuery通过它反射创建插件实例参见上文ReadJournalFor的加载流程。write-plugin读侧连接到的写 Journal 配置路径留空则使用默认写 Journal。refresh-interval实时查询EventsByPersistenceId、EventsByTag、PersistenceIds等分批拉取新事件的间隔。注意InMemoryReadJournal的默认值是3s构造代码中兜底为 1 秒见 InMemoryReadJournal.cs意味着新事件最多延迟该时长才会出现在查询流中——这就是文档中“延迟最多可达 refresh-interval”的实际来源。max-buffer-size单次回放拉取并缓冲的事件条数控制查询侧与写 Journal 之间的批量粒度。InMemoryReadJournal通过DefaultConfiguration()静态方法从程序集资源加载这份reference.conf并在构造函数中读取这三个配置项config.GetString/config.GetInt/config.GetTimeSpan。源码级验证InMemoryReadJournal 如何实现全部预定义查询为了印证上述语义可以看官方内存实现InMemoryReadJournal.cs——它一次性实现了全部八个预定义查询接口public class InMemoryReadJournal : IPersistenceIdsQuery, ICurrentPersistenceIdsQuery, ICurrentEventsByPersistenceIdQuery, IEventsByPersistenceIdQuery, ICurrentEventsByTagQuery, IEventsByTagQuery, ICurrentAllEventsQuery, IAllEventsQuery其实现全部基于Source.ActorPublisherT(...)——即每个查询流背后都是一个 Publisher ActorPersistenceIds()→LivePersistenceIdsPublisher带refreshInterval并.MapMaterializedValue(_ NotUsed.Instance)把物化值归一为NotUsedCurrentPersistenceIds()→CurrentPersistenceIdsPublisher不传刷新间隔天然一次性完成EventsByPersistenceId/CurrentEventsByPersistenceId→EventsByPersistenceIdPublisher实时版传入_refreshInterval一次性版传nullEventsByTag/CurrentEventsByTag→EventsByTagPublisher内部通过ResolveTagOffset把Offset解析为回放起点NoOffset或Sequence(0)→ 从头At(0)Sequence(s)→At(s.Value 1)印证Sequence偏移的排他性起点是“已消费位置 1”FromEnd(n)→LastN(n)延迟到物化时解析其他类型 → 抛出ArgumentException印证“不支持的 Offset 类型会被拒绝”。AllEvents/CurrentAllEvents→AllEventsPublisher实时版传入_refreshInterval一次性版传null。对应地Akka.Persistence.Query.Tests 中的PersistenceQuerySpec验证了PersistenceQuery.Get(Sys).ReadJournalForDummyReadJournal(DummyReadJournal.Identifier)的获取与缓存行为以及配置缺失时抛出异常Identifier -fail场景。OffsetSpec/OffsetCompareSpecs则验证了各Offset类型的相等性、比较与排他语义。这套“Publisher Actor refresh-interval 分批推送 Offset 解析”的实现正是前文所有语义实时推送、一次性完成、排他偏移、失败完成、顺序保证在真实代码中的落地方式可作为你自己编写 Read Journal 插件时的参考蓝本。小结与最佳实践用前先查文档每个 Read Journal 插件都必须文档化它支持的查询类型、顺序保证与流完成策略本文介绍的所有预定义查询都是“建议实现”而非“必须实现”。分清实时与一次性需要持续消费新事件用EventsByPersistenceId/EventsByTag/AllEvents/PersistenceIds只需要当前快照用对应的Current*版本。断点续传三件套EventEnvelope.Offset含SequenceNr是续传的钥匙——Sequence偏移是排他的把信封上的 Offset 原样传回即可无重复、无遗漏地继续。读侧投影推荐骨架Reactive Streams 驱动直接对接无状态转换用SelectAsync(1, ...)保序需要复杂聚合或断点续传时用“并发SelectAsync写 Actor 串行保存 offset”的组合。配置命名空间插件配置的绝对路径 插件 identifierrefresh-interval决定实时查询的推送延迟max-buffer-size决定批量拉取粒度。延伸阅读事件适配器含Tagged打标签的完整配置与读写转换详见 docs/articles/persistence/event-adapters.md事件溯源模型与写侧 API 见 docs/articles/persistence/event-sourcing.md各具体数据存储的 Read Journal 插件Cassandra、关系型库等请查阅其各自的社区文档。赞分享后端微服务【免费下载链接】akka.netCanonical actor model implementation for .NET with local distributed actors in C# and F#.项目地址https://gitcode.com/gh_mirrors/ak/akka.net点击查看免费下载相关推荐Akka Persistence Query 实战指南用统一异步流接口构建 CQRS 读侧查询Akka Persistence Query 实战指南用统一异步流接口构建 CQRS 读侧查询 Akka Persistence Query 是 Akka 持后端并发编程异步编程TabPFN-2.6安全与伦理表格AI模型在敏感数据中的应用规范指南 TabPFN 2.6安全与伦理表格AI模型在敏感数据中的应用规范指南 表格AI模型TabPFN 2.6在金融、医疗、法律等敏感数据领域展现出强大预测能力Akka Persistence Query 的 LevelDB 实现读取日志查询完整实战指南Akka Persistence Query 的 LevelDB 实现读取日志查询完整实战指南 本指南以 Akka 官方文档 persistence quer后端并发编程异步编程上一篇Lynx事件处理机制终极指南从触摸到手势的全链路实现下一篇Wails框架性能优化如何将Electron应用从300MB压缩到42MB创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考