基于 ABP vNext + CQRS + MediatR 构建高可用与高性能微服务系统：从架构设计到落地实战

🧠 基于 ABP vNext + CQRS + MediatR 构建高可用与高性能微服务系统：从架构设计到落地实战

---

---

🧰 模块结构概览

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){// CQRS & Validationcontext.Services.AddMediatR(typeof(CqrsDemoApplicationModule).Assembly);context.Services.AddValidatorsFromAssembly(typeof(CqrsDemoApplicationModule).Assembly);// Pipeline Behaviorscontext.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));context.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));// 分布式缓存注册（Application 层即可依赖）context.Services.AddDistributedCache<OrderDto>();}
}

HttpApiModule

[DependsOn(typeof(CqrsDemoApplicationModule))]
public class CqrsDemoHttpApiModule : AbpModule
{public override void ConfigureServices(ServiceConfigurationContext context){var configuration = context.Services.GetConfiguration();// Redis 连接与缓存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);});// 设置 EF Core 数据库提供者Configure<AbpDbContextOptions>(options =>{options.UseSqlServer(); // 或 UseNpgsql()});}
}

---

🧱 架构图 & 概念说明

• 命令 负责状态更改
• 查询 可接入 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() { } // EF Core 兼容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();
});// EF Core & DbContext
builder.Services.AddAbpDbContext<CqrsDemoDbContext>(options =>
{options.AddDefaultRepositories();
});
builder.Services.Configure<AbpDbContextOptions>(options =>options.UseSqlServer(config.GetConnectionString("Default")));// Redis 缓存 & 布隆过滤器
builder.Services.AddStackExchangeRedisCache(opts =>opts.Configuration = config["Redis:Connection"]);
builder.Services.AddSingleton<IBloomFilter>(_ =>new BloomFilter(expectedItemCount: 10000, falsePositiveRate: 0.01));// Health Checks
builder.Services.AddHealthChecks().AddDbContextCheck<CqrsDemoDbContext>("Database").AddRedis(config["Redis:Connection"], name: "Redis");// Polly HTTP 客户端
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

---