当前位置：首页 > news >正文

ABP VNext + Cosmos DB Change Feed：搭建实时数据变更流服务

news 2025/7/4 15:13:39

ABP VNext + Cosmos DB Change Feed：搭建实时数据变更流服务 🚀

📚 目录

ABP VNext + Cosmos DB Change Feed：搭建实时数据变更流服务 🚀
- - TL;DR ✨🚀
- 1. 环境与依赖 🏗️
- 2. 服务注册与依赖注入 🔌
- 3. 封装 Change Feed 为 IHostedService 🔧
- - 3.1 HostedService 生命周期流程图
  - 3.2 `ChangeFeedHostedService` 实现
- 4. 事务与幂等 🛡️
- 5. 发布到事件总线 📡
- - MassTransit 示例
- 6. 容错与监控 🛠️📊
- 7. 横向扩展 🌐
- 参考文档 📖

TL;DR ✨🚀

全托管 DI：CosmosClient 由容器单例管理，HostedService 构造注入，优雅释放。
作用域与事务：回调内创建新 Scope，结合 IUnitOfWorkManager 实现事务一致性🛡️。
Exactly-Once：通过（DocumentId, ETag）唯一索引 + 手动 Checkpoint，确保不漏不重✅。
容错重试：Polly 指数退避重试与熔断，处理启动与回调中的网络抖动🔄。
监控可扩展：日志、指标、Dead-Letter 容错，中控告警 + 多实例自动分片，助力弹性伸缩📊。

1. 环境与依赖 🏗️

.NET 平台：.NET 6 + / ABP VNext 6.x
Azure 资源：Cosmos DB Core API（Source 容器 + Lease 容器）

主要 NuGet 包

dotnet add package Microsoft.Azure.Cosmos
dotnet add package Volo.Abp.EventBus.MassTransit
dotnet add package Streamiz.Kafka.Net.Stream        # 可选
dotnet add package Volo.Abp.EntityFrameworkCore
dotnet add package Polly

appsettings.json 配置

{"Cosmos": {"ConnectionString": "<your-connection-string>","Database": "MyAppDb","SourceContainer": "Docs","LeaseContainer": "Leases"},"RabbitMq": { "Host": "rabbitmq://localhost" },"Kafka":   { "BootstrapServers": "localhost:9092" }
}

2. 服务注册与依赖注入 🔌

在 MyAppModule 的 ConfigureServices 中：

public override void ConfigureServices(ServiceConfigurationContext context)
{var configuration = context.Services.GetConfiguration();// CosmosClient 单例托管context.Services.AddSingleton(sp =>new CosmosClient(configuration["Cosmos:ConnectionString"]));// Polly 重试策略：3 次指数退避context.Services.AddSingleton(sp => Policy.Handle<Exception>().WaitAndRetryAsync(retryCount: 3,sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),onRetry: (ex, ts, retryCount, ctx) =>{var logger = sp.GetRequiredService<ILogger<ChangeFeedHostedService>>();logger.LogWarning(ex, "⚠️ ChangeFeed 启动重试，第 {RetryCount} 次", retryCount);}));// 注册 HostedServicecontext.Services.AddHostedService<ChangeFeedHostedService>();
}

💡 Tip：将 Cosmos、RabbitMQ、Kafka 等配置抽象到 SettingDefinition，支持动态变更。

3. 封装 Change Feed 为 IHostedService 🔧

3.1 HostedService 生命周期流程图

⚠️ “触发 StartAsync”更准确地反映了 ASP.NET Core Host 的启动流程。

3.2 `ChangeFeedHostedService` 实现

public class ChangeFeedHostedService : IHostedService, IDisposable
{private readonly CosmosClient _cosmosClient;private readonly IConfiguration _config;private readonly ILogger<ChangeFeedHostedService> _logger;private readonly IAsyncPolicy _retryPolicy;private readonly IServiceProvider _serviceProvider;private ChangeFeedProcessor _processor;public ChangeFeedHostedService(CosmosClient cosmosClient,IConfiguration config,ILogger<ChangeFeedHostedService> logger,IAsyncPolicy retryPolicy,IServiceProvider serviceProvider){_cosmosClient    = cosmosClient;_config          = config;_logger          = logger;_retryPolicy     = retryPolicy;_serviceProvider = serviceProvider;}public async Task StartAsync(CancellationToken ct){await _retryPolicy.ExecuteAsync(async () =>{_logger.LogInformation("🔄 ChangeFeedHostedService 正在启动...");var dbName = _config["Cosmos:Database"];var src    = _cosmosClient.GetContainer(dbName, _config["Cosmos:SourceContainer"]);var lease  = _cosmosClient.GetContainer(dbName, _config["Cosmos:LeaseContainer"]);_processor = src.GetChangeFeedProcessorBuilder<MyDocument>("abp-processor", HandleChangesAsync).WithInstanceName(Environment.MachineName).WithLeaseContainer(lease).WithStartTime(DateTime.MinValue.ToUniversalTime()).Build();await _processor.StartAsync(ct);_logger.LogInformation("✅ ChangeFeedProcessor 已启动");});}public async Task StopAsync(CancellationToken ct){if (_processor != null){_logger.LogInformation("🛑 ChangeFeedProcessor 正在停止...");await _processor.StopAsync(ct);_logger.LogInformation("✅ ChangeFeedProcessor 已停止");}}public void Dispose() => _processor = null;private async Task HandleChangesAsync(IReadOnlyCollection<MyDocument> docs,CancellationToken ct){if (docs == null || docs.Count == 0) return;_logger.LogInformation("📥 收到 {Count} 条文档变更", docs.Count);// 创建新的 DI Scopeusing var scope = _serviceProvider.CreateScope();var uowManager = scope.ServiceProvider.GetRequiredService<IUnitOfWorkManager>();var eventBus   = scope.ServiceProvider.GetRequiredService<IDistributedEventBus>();var auditRepo  = scope.ServiceProvider.GetRequiredService<IRepository<AuditEntry, Guid>>();// 开始事务using var uow = await uowManager.BeginAsync();foreach (var doc in docs){try{// 发布领域事件await eventBus.PublishAsync(new DocumentChangedEvent(doc.Id, doc), ct);// 审计写入，唯一索引保证幂等var entry = new AuditEntry{DocumentId = doc.Id,ETag       = doc.ETag,Operation  = doc.Operation,Timestamp  = DateTime.UtcNow,Payload    = JsonConvert.SerializeObject(doc)};await auditRepo.InsertAsync(entry, autoSave: true);}catch (DbUpdateException dbEx)when (dbEx.InnerException?.Message.Contains("UNIQUE") ?? false){_logger.LogWarning("⚠️ 文档 {DocumentId}@{ETag} 唯一索引冲突，跳过", doc.Id, doc.ETag);}catch (Exception ex){_logger.LogError(ex, "🔥 写审计失败，写入 Dead-Letter 容器");await WriteToDeadLetterAsync(doc, ex, ct);// 回滚本次事务await uow.RollbackAsync();// 跳过到下一文档continue;}}// 提交事务await uow.CompleteAsync();// 手动 Checkpointawait _processor.CheckpointAsync(ct);_logger.LogInformation("🗸 Checkpoint 完成，位置已记录");}private Task WriteToDeadLetterAsync(MyDocument doc, Exception ex, CancellationToken ct){// TODO: 实现将失败批次写入 Dead-Letter 容器或队列，用于离线补偿return Task.CompletedTask;}
}

4. 事务与幂等 🛡️

💡 Tip：在 AuditEntry 上建立 (DocumentId, ETag) 唯一索引，捕获 DbUpdateException 后跳过重复。

5. 发布到事件总线 📡

MassTransit 示例

services.AddMassTransit(cfg =>
{cfg.AddConsumer<DocumentChangedConsumer>();cfg.UsingRabbitMq((ctx, rc) =>{rc.Host(Configuration["RabbitMq:Host"]);rc.ReceiveEndpoint("change-feed-queue", e =>e.ConfigureConsumer<DocumentChangedConsumer>(ctx));});
});

public class DocumentChangedConsumer : IConsumer<DocumentChangedEvent>
{public async Task Consume(ConsumeContext<DocumentChangedEvent> ctx){// 下游业务逻辑…}
}