🧠 基于 ABP vNext + CQRS + MediatR 构建高可用与高性能微服务系统:从架构设计到落地实战
目录
- 🧠 基于 ABP vNext + CQRS + MediatR 构建高可用与高性能微服务系统:从架构设计到落地实战
- 🧰 模块结构概览
- 📦 各模块注册代码示例
- DomainModule
- ApplicationModule
- HttpApiModule
- DbMigratorModule
- 🧱 架构图 & 概念说明
- ✍️ 核心代码实践
- 1. 聚合根 & 仓储接口
- 2. CacheKeys 常量
- 3. 命令 + 验证器 + 幂等处理
- 4. Pipeline 行为示意图
- 5. ValidationBehavior + LoggingBehavior
- 6. 查询缓存 + 布隆过滤器
- 部署架构示意图
- 🧪 自动化测试示例(Testcontainers)
-
- 📦 Docker Compose
🧰 模块结构概览
Abp.CqrsDemo
├── src
│ ├── Abp.CqrsDemo.Domain
│ ├── Abp.CqrsDemo.Application
│ ├── Abp.CqrsDemo.HttpApi
│ └── Abp.CqrsDemo.DbMigrator
└── tests└── Abp.CqrsDemo.Tests
- Domain:实体、聚合根、领域事件、仓储接口
- Application:命令、查询、MediatR 处理器、DTO、Pipeline Behaviors、CacheKeys
- HttpApi:控制器、外部配置绑定
- DbMigrator:数据库初始化与迁移工具
📦 各模块注册代码示例
DomainModule
[DependsOn(typeof(AbpDddDomainModule))]
public class CqrsDemoDomainModule : AbpModule
{
}
ApplicationModule
[DependsOn(typeof(CqrsDemoDomainModule))]
public class CqrsDemoApplicationModule : AbpModule
{public override void ConfigureServices(ServiceConfigurationContext context){context.Services.AddMediatR(typeof(CqrsDemoApplicationModule).Assembly);context.Services.AddValidatorsFromAssembly(typeof(CqrsDemoApplicationModule).Assembly);context.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));context.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));context.Services.AddDistributedCache<OrderDto>();}
}
HttpApiModule
[DependsOn(typeof(CqrsDemoApplicationModule))]
public class CqrsDemoHttpApiModule : AbpModule
{public override void ConfigureServices(ServiceConfigurationContext context){var configuration = context.Services.GetConfiguration();context.Services.AddStackExchangeRedisCache(options =>{options.Configuration = configuration["Redis:Connection"];});context.Services.AddControllers();}
}
DbMigratorModule
[DependsOn(typeof(AbpEntityFrameworkCoreModule))]
public class CqrsDemoDbMigratorModule : AbpModule
{public override void ConfigureServices(ServiceConfigurationContext context){context.Services.AddAbpDbContext<CqrsDemoDbContext>(options =>{options.AddDefaultRepositories(includeAllEntities: true);});Configure<AbpDbContextOptions>(options =>{options.UseSqlServer(); });}
}
🧱 架构图 & 概念说明
- 命令 负责状态更改
- 查询 可接入 Redis 或 读库
- Pipeline:Validation → Logging → Handler
✍️ 核心代码实践
1. 聚合根 & 仓储接口
public class Order : AggregateRoot<Guid>
{public string ProductId { get; private set; }public int Quantity { get; private set; }protected Order() { } public Order(Guid id, string productId, int quantity){Id = id;ProductId = productId;Quantity = quantity;}public void ChangeQuantity(int newQty){if (newQty <= 0) throw new BusinessException("Quantity must be positive");Quantity = newQty;}
}public interface IOrderRepository : IRepository<Order, Guid>
{
}
2. CacheKeys 常量
public static class CacheKeys
{public const string OrderById = "order:{0}";
}
3. 命令 + 验证器 + 幂等处理
public record CreateOrderCommand(Guid OrderId, string ProductId, int Quantity) : IRequest<Guid>;public class CreateOrderValidator : AbstractValidator<CreateOrderCommand>
{public CreateOrderValidator(){RuleFor(x => x.ProductId).NotEmpty();RuleFor(x => x.Quantity).GreaterThan(0);}
}public class CreateOrderHandler : IRequestHandler<CreateOrderCommand, Guid>
{private readonly IOrderRepository _repo;private readonly IDistributedIdempotencyService _idempotency;public CreateOrderHandler(IOrderRepository repo, IDistributedIdempotencyService idempotency){_repo = repo;_idempotency = idempotency;}public async Task<Guid> Handle(CreateOrderCommand request, CancellationToken ct){return await _idempotency.ExecuteAsync(request.OrderId.ToString(), async () =>{var order = new Order(request.OrderId, request.ProductId, request.Quantity);await _repo.InsertAsync(order, cancellationToken: ct);return order.Id;});}
}
4. Pipeline 行为示意图
5. ValidationBehavior + LoggingBehavior
public class ValidationBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>where TRequest : notnull
{private readonly IEnumerable<IValidator<TRequest>> _validators;public ValidationBehavior(IEnumerable<IValidator<TRequest>> validators) => _validators = validators;public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken ct){var context = new ValidationContext<TRequest>(request);var results = await Task.WhenAll(_validators.Select(v => v.ValidateAsync(context, ct)));var failures = results.SelectMany(r => r.Errors).Where(f => f != null).ToList();if (failures.Any()) throw new ValidationException(failures);return await next();}
}public class LoggingBehavior<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
{private readonly ILogger<LoggingBehavior<TRequest, TResponse>> _logger;public LoggingBehavior(ILogger<LoggingBehavior<TRequest, TResponse>> logger) => _logger = logger;public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, CancellationToken ct){_logger.LogInformation("Handling {Request} with payload: {@Payload}", typeof(TRequest).Name, request);var sw = Stopwatch.StartNew();var response = await next();sw.Stop();_logger.LogInformation("Handled {Request} in {Elapsed}ms with response: {@Response}", typeof(TRequest).Name, sw.ElapsedMilliseconds, response);return response;}
}
6. 查询缓存 + 布隆过滤器
public class GetOrderByIdHandler : IRequestHandler<GetOrderByIdQuery, OrderDto>
{private readonly IOrderRepository _repo;private readonly IDistributedCache<OrderDto> _cache;private readonly IBloomFilter _bloomFilter;public GetOrderByIdHandler(IOrderRepository repo,IDistributedCache<OrderDto> cache,IBloomFilter bloomFilter){_repo = repo;_cache = cache;_bloomFilter = bloomFilter;}public async Task<OrderDto> Handle(GetOrderByIdQuery request, CancellationToken ct){if (!_bloomFilter.Contains(request.OrderId))throw new EntityNotFoundException(typeof(Order), request.OrderId);var cacheKey = string.Format(CacheKeys.OrderById, request.OrderId);return await _cache.GetOrAddAsync(cacheKey,async () =>{var order = await _repo.FindAsync(request.OrderId, ct);if (order == null) throw new EntityNotFoundException(typeof(Order), request.OrderId);return new OrderDto(order.Id, order.ProductId, order.Quantity);},options =>{options.SlidingExpiration = TimeSpan.FromMinutes(5);options.AbsoluteExpirationRelativeToNow = TimeSpan.FromHours(1);});}
}```markdown---## 🚀 部署与运行(Program.cs)```csharp
var builder = WebApplication.CreateBuilder(args);
var config = builder.Configuration;
builder.Host.ConfigureAppConfiguration((ctx, cfg) =>
{cfg.AddJsonFile("appsettings.json", optional: false, reloadOnChange: true).AddEnvironmentVariables();
});
builder.Services.AddAbpDbContext<CqrsDemoDbContext>(options =>
{options.AddDefaultRepositories();
});
builder.Services.Configure<AbpDbContextOptions>(options =>options.UseSqlServer(config.GetConnectionString("Default")));
builder.Services.AddStackExchangeRedisCache(opts =>opts.Configuration = config["Redis:Connection"]);
builder.Services.AddSingleton<IBloomFilter>(_ =>new BloomFilter(expectedItemCount: 10000, falsePositiveRate: 0.01));
builder.Services.AddHealthChecks().AddDbContextCheck<CqrsDemoDbContext>("Database").AddRedis(config["Redis:Connection"], name: "Redis");
builder.Services.AddHttpClient("order").AddTransientHttpErrorPolicy(p => p.WaitAndRetryAsync(3, i => TimeSpan.FromSeconds(Math.Pow(2, i)))).AddTransientHttpErrorPolicy(p => p.CircuitBreakerAsync(5, TimeSpan.FromSeconds(30)));var app = builder.Build();
app.MapHealthChecks("/health/live", new HealthCheckOptions { Predicate = _ => false });
app.MapHealthChecks("/health/ready");app.MapControllers();
app.Run();
部署架构示意图
🧪 自动化测试示例(Testcontainers)
using DotNet.Testcontainers.Builders;
using DotNet.Testcontainers.Containers;public class CustomWebAppFactory : WebApplicationFactory<Program>, IAsyncLifetime
{private readonly PostgreSqlTestcontainer _postgres;public CustomWebAppFactory(){_postgres = new TestcontainersBuilder<PostgreSqlTestcontainer>().WithDatabase(new PostgreSqlTestcontainerConfiguration{Database = "demo",Username = "sa",Password = "sa"}).Build();}public async Task InitializeAsync() => await _postgres.StartAsync();public async Task DisposeAsync() => await _postgres.DisposeAsync();protected override void ConfigureWebHost(IWebHostBuilder builder){builder.ConfigureAppConfiguration((ctx, cfg) =>{cfg.AddInMemoryCollection(new Dictionary<string, string>{["ConnectionStrings:Default"] = _postgres.ConnectionString});});}
}public class OrderApiTests : IClassFixture<CustomWebAppFactory>
{private readonly HttpClient _client;public OrderApiTests(CustomWebAppFactory factory) => _client = factory.CreateClient();[Fact]public async Task Should_Create_Order_Successfully(){var command = new { orderId = Guid.NewGuid(), productId = "A001", quantity = 1 };var response = await _client.PostAsJsonAsync("/api/orders", command);response.EnsureSuccessStatusCode();}
}
测试流程图
📦 Docker Compose
version: "3.9"
services:api:build: .ports:- "5000:80"db:image: postgresenvironment:POSTGRES_DB: demoPOSTGRES_USER: saPOSTGRES_PASSWORD: saredis:image: redis