后端消息队列微服务【免费下载链接】CAP基于最终一致性的微服务分布式事务解决方案也是一种采用 Outbox 模式的事件总线。项目地址https://gitcode.com/dotnetcore/CAP点击查看免费下载导读本文讲解 CAP 分布式事务框架中基于内存的消息存储方案DotNetCore.CAP.InMemoryStorage。它适合开发与测试环境快速验证 CAP 的发布/订阅能力无需部署 MySQL、SQL Server 或 MongoDB 等持久化中间件读完本文你将掌握内存存储的安装配置、5 分钟自动清理机制、内存数据结构的底层实现以及不支持事务发布这一关键限制背后的原因与替代方案。一、内存存储是什么为什么用它CAP 是一个基于 Outbox 模式的事件总线与最终一致性分布式事务解决方案其核心可靠性来源于先持久化、后投递消息在进入消息队列之前先被写入本地存储数据库表或 NoSQL 集合以此保证任何环境或网络异常下消息不丢失详见 存储总览。内存存储则是 CAP 提供的一种不具备持久化能力的存储实现它把消息直接保存在进程内的ConcurrentDictionary中进程重启即全部丢失。因此它通常只用于开发和测试环境一旦切换到内存存储你就失去了本地事务消息的可靠性保证这一点在官方文档中已被明确警示见 in-memory-storage.md。从源码结构看DotNetCore.CAP.InMemoryStorage项目完整实现了 CAP 的存储抽象包含以下核心文件文件职责CAP.Options.Extensions.cs提供UseInMemoryStorage()扩展方法CAP.InMemoryCapOptionsExtension.cs注册内存存储相关服务到 DI 容器IDataStorage.InMemory.cs消息的增删改查与重试调度实现IMonitoringApi.InMemory.cs为 CAP Dashboard 提供统计与查询IStorageInitializer.InMemory.cs存储初始化内存模式为空操作ICapTransaction.InMemory.cs事务适配空实现MemoryMessage.cs内存消息实体二、安装与配置2.1 安装 NuGet 包在目标项目的包管理器控制台或使用dotnet add package执行PM Install-Package DotNetCore.CAP.InMemoryStorage2.2 在 ConfigureServices 中注册在Startup.cs或 Program.cs 的builder.Services中调用AddCap并在配置 lambda 里启用UseInMemoryStorage()public void ConfigureServices(IServiceCollection services) { // ... 其他服务注册 services.AddCap(x { x.UseInMemoryStorage(); // x.UseXXX ... 例如 x.UseRabbitMQ() / x.UseKafka() 等消息队列配置 }); }这段代码等价于在容器中注入内存存储实现。查看 CAP.InMemoryCapOptionsExtension.cs 可知其注册了三个关键服务ICapTransaction→InMemoryCapTransaction瞬时生命周期IDataStorage→InMemoryStorage单例消息容器全局唯一IStorageInitializer→InMemoryStorageInitializer单例。由于三个组件都是进程内对象注册为单例Singleton意味着所有发布/接收消息共享同一份内存字典这一点与数据库存储按表持久化完全不同。2.3 关于UseXXX组合x.UseInMemoryStorage()只解决消息存储问题传输层消息队列仍需单独配置例如UseRabbitMQ、UseKafka、UseRedisStreams等。以仓库中的示例项目 Sample.ConsoleApp 与 Sample.Dashboard.Auth 为参考内存存储常与测试脚手架搭配使用测试工程 DotNetCore.CAP.Test 也大量依赖内存化组件完成集成测试印证了它开发/测试首选的定位。三、5 分钟自动清理机制官方文档指出CAP 每 5 分钟会从内存中清理成功消息。这一行为由两个部分共同构成清理间隔配置全局选项CollectorCleaningInterval默认值为 300 秒5 分钟见 CAP.Options.cs。你可以通过x.CollectorCleaningInterval ...调整该间隔。清理执行器后台处理器 IProcessor.Collector.cs 周期性调用IDataStorage.DeleteExpiresAsync(table, time, ItemBatch, ...)每次批量删除 1000 条ItemBatch 1000删除完毕后等待下一个 300 秒周期。内存实现的删除逻辑在 IDataStorage.InMemory.cs它根据表名PublishedMessages或ReceivedMessages筛选出ExpiresAt timeout的消息通过ConcurrentDictionary.TryRemove逐条移除。需要特别说明的是只有进入终态如成功/失败且携带过期时间的消息才会被清理。以失败消息为例StoreReceivedExceptionMessageAsync 存储时会设置ExpiresAt DateTime.Now.AddSeconds(FailedMessageExpiredAfter)而FailedMessageExpiredAfter默认是 15 天见 CAP.Options.cs。换言之成功消息 5 分钟一轮即可回收失败消息则会保留到重试耗尽并过期后才被清理这与数据库存储的行为逻辑一致。四、内存数据结构的底层实现4.1 两个并发字典从 IDataStorage.InMemory.cs 可以看到全部消息存放在两个静态ConcurrentDictionarystring, MemoryMessage中PublishedMessages已发布消息Key 为消息 DbIdReceivedMessages已接收消息Key 为雪花 ID 字符串。使用ConcurrentDictionary是为了在多线程消费者线程、重试调度线程、Dashboard 查询线程并发访问下保证线程安全。由于是static字段同一进程内所有 CAP 实例共享这些消息这也意味着消息生命周期与进程绑定进程退出数据即消失。4.2 MemoryMessage 实体MemoryMessage.cs 继承自MediumMessageCAP 的中间消息模型并额外增加三个字段字段说明Name消息主题/名称对应发布时的 topicStatusName消息状态Scheduled / Succeeded / Failed / Delayed 等见 StatusName.csGroup消费组名仅接收消息使用发布/接收存储时的状态流转也体现了 Outbox 语义新消息以StatusName.Scheduled写入StoreMessageAsync发送成功后由调度器改为Succeeded失败则改为Failed。4.3 初始化与锁内存模式的空实现数据库存储通常需要建表、建索引、建锁表而内存模式全部是空操作IStorageInitializer.InMemory.csInitializeAsync直接返回完成锁表名返回空字符串。IDataStorage.InMemory.csAcquireLockAsync永远返回trueReleaseLockAsync/RenewLockAsync直接返回。也就是说内存存储不提供跨进程的存储锁能力——这与它在单进程内运行的定位相符。从 general.md 的说明可知存储锁表只有在启用UseStorageLock时才会创建内存模式天然不适用该特性。五、Publish with transaction为什么不支持事务发布官方文档明确In-memory storagedoes not supporttransactional message publishing.理解这一点需要回到 CAP 事务发布的本质CAP 依靠业务数据库事务 消息表写入同事务来保证消息与业务操作的一致性详见 存储总览。内存存储没有数据库事务的概念无法与外部业务事务协同。源码印证了这一结论ICapTransaction.InMemory.cs 中的Commit()仅仅调用Flush()Rollback()被注释为//IgnoreCommitAsync/RollbackAsync均为空实现。可以推断这是 CAP 为满足ICapTransaction接口而提供的占位实现真实的事务协调行为完全缺失。因此如果你的业务需要事务内发消息例如下单后必须在同一数据库事务里写订单并投递事件请改用支持事务的存储SQL ServerMySQLPostgreSqlMongoDB这些存储均与 CAP 官方文档中列出的支持事务的数据库清单一一对应见 general.md。内存存储适合只验证发布—投递—订阅链路本身、不依赖持久化的场景。六、与 Dashboard 监控的配合内存存储同样实现了 IMonitoringApi因此 CAP Dashboard 的统计图表在内存模式下依然可用GetStatisticsAsync统计 Succeeded / Failed / Delayed 等各状态数量IMonitoringApi.InMemory.csGetMessagesAsync支持按状态、名称、内容关键字分页查询发布与接收消息HourlyFailedJobs/HourlySucceededJobs提供过去 24 小时的按小时趋势数据。从源码看Dashboard 查询直接读取内存字典并做 LINQ 过滤响应极快但需注意其Version字段固定返回N/A因为内存中没有版本表这是与数据库实现的一个可见差异。七、适用场景与决策建议综合官方文档与源码实现给出如下选型建议场景建议单元/集成测试、本地快速验证 CAP 功能使用UseInMemoryStorage()无需任何外部依赖CI 流水线中的冒烟测试可用内存存储降低环境复杂度仓库测试工程即采用类似思路生产环境的分布式事务必须使用 SQL Server / MySQL / PostgreSQL / MongoDB 等持久化存储需要事务内发消息内存存储不适用请选用支持事务的存储需要跨实例/进程的数据可见性内存存储不适用数据在各自进程内总结内存存储是 CAP 中零依赖、上手最快的存储方案一条UseInMemoryStorage()即可在开发测试环境中跑通发布订阅全链路成功消息由CollectorProcessor每 5 分钟CollectorCleaningInterval 300秒批量清理底层以两个静态ConcurrentDictionary承载消息并完整支持 Dashboard 监控查询。它的代价同样清晰无持久化、无事务发布、无跨进程锁因此生产环境请务必切换到官方支持事务的持久化存储。赞分享后端消息队列微服务【免费下载链接】CAP基于最终一致性的微服务分布式事务解决方案也是一种采用 Outbox 模式的事件总线。项目地址https://gitcode.com/dotnetcore/CAP点击查看免费下载相关推荐如何免费解锁网盘直链下载8大网盘直链解析保姆级教程配合IDM与Aria2提速如何免费解锁网盘直链下载8大网盘直链解析保姆级教程配合IDM与Aria2提速 算一笔账200KB/s 下载 8GB 文件大约要 2 小时同样的文件走真前端WLED内存优化技巧解决ESP8266存储空间不足WLED内存优化技巧解决ESP8266存储空间不足 在使用ESP8266开发WLED项目时许多用户都会遇到 存储空间不足 的问题尤其是在添加自定义效果或传物联网嵌入式智能硬件CAP 事件总线 SQL Server 存储接入指南配置、Outbox 表结构与本地消息事务实战CAP 事件总线 SQL Server 存储接入指南配置、Outbox 表结构与本地消息事务实战 导读 本文是 DotNetCore.CAP 分布式事务框架使后端消息队列微服务消息路由上一篇kotaemon代码生成技术文档转代码下一篇Graphite图像扭曲创造性变形效果的实现方法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考