《手把手教你用 .NET Core 搭建高并发、可扩展的 CQRS 与 DDD 架构》

MyApp.Solution├── MyApp.Domain         // 领域层：实体、值对象、聚合、领域服务、仓储接口├── MyApp.Application    // 应用层：应用服务、DTO、CQRS 命令/查询、中介处理├── MyApp.Infrastructure // 基础设施层：EF Core 实现仓储、事件总线、外部服务调用├── MyApp.WebApi         // 表现层：ASP.NET Core Web API，接收请求，返回响应└── MyApp.UnitTests      // 单元测试

// Domain/Entities/Order.cspublic class Order : AggregateRoot{    public Guid CustomerId { get; private set; }    public decimal TotalAmount { get; private set; }    public OrderStatus Status { get; private set; }    private readonly List<OrderItem> _items = new();    public IReadOnlyCollection<OrderItem> Items => _items.AsReadOnly();    // 工厂方法    public static Order Create(Guid customerId, List<OrderItemDto> items)    {        var order = new Order        {            Id = Guid.NewGuid(),            CustomerId = customerId,            Status = OrderStatus.Pending        };        foreach (var item in items)        {            order.AddItem(item.ProductId, item.Quantity, item.UnitPrice);        }        // 添加领域事件        order.AddDomainEvent(new OrderCreatedEvent(order.Id));
        return order;    }    private void AddItem(Guid productId, int quantity, decimal unitPrice)    {        // 业务规则校验        if (quantity <= 0) throw new DomainException("数量必须大于0");        var item = new OrderItem(productId, quantity, unitPrice);        _items.Add(item);        TotalAmount += item.Amount;    }}

// Domain/Entities/OrderItem.cspublic class OrderItem : Entity{    public Guid ProductId { get; private set; }    public int Quantity { get; private set; }    public decimal UnitPrice { get; private set; }    public decimal Amount => Quantity * UnitPrice;    public OrderItem(Guid productId, int quantity, decimal unitPrice)    {        ProductId = productId;        Quantity = quantity;        UnitPrice = unitPrice;    }}

// Domain/Repositories/IOrderRepository.cspublic interface IOrderRepository : IRepository<Order, Guid>{    Task<Order> GetByOrderNumberAsync(string orderNumber);    Task<List<Order>> GetByCustomerIdAsync(Guid customerId);}

客户端   ↓Web API Controller   ↓MediatR (中介者)   ↙       ↘Command Handler          Query Handler   ↓                       ↓Domain Layer (业务逻辑)     Read Model (DTO/ViewModel)   ↓                       ↓Write Database (EF Core)   Read Database (SQL View / NoSQL / Cache)

<PackageReference Include="MediatR" Version="12.2.0" /><PackageReference Include="MediatR.Extensions.Microsoft.DependencyInjection" Version="12.1.0" />

// WebApi/Program.csbuilder.Services.AddMediatR(cfg => cfg.RegisterServicesFromAssembly(typeof(CreateOrderCommand).Assembly));

// Application/Commands/CreateOrderCommand.cspublic record CreateOrderCommand(    Guid CustomerId,    List<OrderItemDto> Items) : IRequest<Guid>; // 返回订单ID// DTO 用于传输public record OrderItemDto(Guid ProductId, int Quantity, decimal UnitPrice);

// Application/Commands/CreateOrderCommandHandler.cspublic class CreateOrderCommandHandler : IRequestHandler<CreateOrderCommand, Guid>{    private readonly IOrderRepository _orderRepository;    private readonly IUnitOfWork _unitOfWork;    public CreateOrderCommandHandler(IOrderRepository orderRepository, IUnitOfWork unitOfWork)    {        _orderRepository = orderRepository;        _unitOfWork = unitOfWork;    }    public async Task<Guid> Handle(CreateOrderCommand request, CancellationToken ct)    {        // 1. 使用领域模型创建订单        var order = Order.Create(request.CustomerId, request.Items);        // 2. 持久化        await _orderRepository.AddAsync(order, ct);        await _unitOfWork.CommitAsync(ct);        // 3. 可选：发布领域事件（用于后续处理，如发送邮件、库存扣减）        // 事件处理将在后续章节讲解        return order.Id;    }}

// Application/Queries/GetOrderQuery.cspublic record GetOrderQuery(Guid OrderId) : IRequest<OrderDto>;// Application/Queries/GetOrderQueryHandler.cspublic class GetOrderQueryHandler : IRequestHandler<GetOrderQuery, OrderDto>{    private readonly DapperContext _dapperContext; // 使用 Dapper 高效查询    public GetOrderQueryHandler(DapperContext dapperContext)    {        _dapperContext = dapperContext;    }    public async Task<OrderDto> Handle(GetOrderQuery request, CancellationToken ct)    {        const string sql = @"            SELECT o.Id, o.OrderNumber, o.TotalAmount, o.Status,                   c.Name as CustomerName,                   oi.ProductId, oi.Quantity, oi.UnitPrice            FROM Orders o            JOIN Customers c ON o.CustomerId = c.Id            LEFT JOIN OrderItems oi ON o.Id = oi.OrderId            WHERE o.Id = @OrderId";        using var connection = _dapperContext.CreateConnection();        var lookup = new Dictionary<Guid, OrderDto>();        await connection.QueryAsync<OrderDto, OrderItemDto, OrderDto>(            sql,            (order, item) =>            {                if (!lookup.TryGetValue(order.Id, out var existingOrder))                {                    existingOrder = order;                    existingOrder.Items = new List<OrderItemDto>();                    lookup.Add(order.Id, existingOrder);                }                if (item != null) existingOrder.Items.Add(item);                return existingOrder;            },            request,            splitOn: "ProductId"        );        return lookup.Values.FirstOrDefault();    }}

// DTOspublic record OrderDto(    Guid Id,    string OrderNumber,    decimal TotalAmount,    OrderStatus Status,    string CustomerName,    List<OrderItemDto> Items);

// WebApi/Controllers/OrdersController.cs[ApiController][Route("api/[controller]")]public class OrdersController : ControllerBase{    private readonly IMediator _mediator;    public OrdersController(IMediator mediator)    {        _mediator = mediator;    }    [HttpPost]    public async Task<IActionResult> CreateOrder([FromBody] CreateOrderCommand command)    {        var orderId = await _mediator.Send(command);        return CreatedAtAction(nameof(GetOrder), new { id = orderId }, orderId);    }    [HttpGet("{id}")]    public async Task<IActionResult> GetOrder(Guid id)    {        var query = new GetOrderQuery(id);        var order = await _mediator.Send(query);        return Ok(order);    }}

// Domain/Events/OrderCreatedEvent.cspublic record OrderCreatedEvent(Guid OrderId) : INotification;

// Domain/Entities/Order.cspublic class Order : AggregateRoot{    // ... 其他代码    public static Order Create(Guid customerId, List<OrderItemDto> items)    {        var order = new Order        {            Id = Guid.NewGuid(),            CustomerId = customerId,            Status = OrderStatus.Pending        };        foreach (var item in items)        {            order.AddItem(item.ProductId, item.Quantity, item.UnitPrice);        }        // 发布事件        order.AddDomainEvent(new OrderCreatedEvent(order.Id));
        return order;    }}

// Infrastructure/Data/UnitOfWork.cspublic class UnitOfWork : IUnitOfWork{    private readonly AppDbContext _context;    private readonly IPublisher _mediator; // MediatR 的发布者    public UnitOfWork(AppDbContext context, IPublisher mediator)    {        _context = context;        _mediator = mediator;    }    public async Task<bool> CommitAsync(CancellationToken ct)    {        // 1. 保存实体变更        var result = await _context.SaveChangesAsync(ct);        // 2. 发布所有聚合根的领域事件        await PublishDomainEvents(ct);        return result > 0;    }    private async Task PublishDomainEvents(CancellationToken ct)    {        var aggregates = _context.ChangeTracker            .Entries<IAggregateRoot>()            .Where(x => x.Entity.DomainEvents.Any())            .Select(x => x.Entity)            .ToList();        foreach (var aggregate in aggregates)        {            var events = aggregate.DomainEvents.ToList();            aggregate.ClearDomainEvents(); // 清空已发布的事件            foreach (var @event in events)            {                await _mediator.Publish(@event, ct);            }        }    }}

// Application/Handlers/OrderCreatedEventHandler.cspublic class OrderCreatedEventHandler : INotificationHandler<OrderCreatedEvent>{    private readonly ILogger<OrderCreatedEventHandler> _logger;    private readonly IEmailService _emailService;    private readonly IInventoryService _inventoryService; // 可能是 gRPC 或 HTTP 客户端    public OrderCreatedEventHandler(        ILogger<OrderCreatedEventHandler> logger,        IEmailService emailService,        IInventoryService inventoryService)    {        _logger = logger;        _emailService = emailService;        _inventoryService = inventoryService;    }    public async Task Handle(OrderCreatedEvent notification, CancellationToken ct)    {        _logger.LogInformation("订单 {OrderId} 已创建，正在处理后续操作...", notification.OrderId);        // 1. 发送订单确认邮件        await _emailService.SendOrderConfirmationAsync(notification.OrderId, ct);        // 2. 调用库存服务扣减库存（可能是异步消息）        await _inventoryService.DeductInventoryAsync(notification.OrderId, ct);        // 3. 更新用户积分（领域服务）        // await _pointsService.AddPointsAsync(...);    }}

Command → [Aggregate] → Events → Event Store → (Replay) → Current State                                 ↓                             Projections → Read Models

// Domain/Events/OrderPlacedEvent.cspublic record OrderPlacedEvent(    Guid OrderId,    Guid CustomerId,    List<OrderItem> Items,    DateTime PlacedAt) : IEvent;// Infrastructure/EventStore/IEventStore.cspublic interface IEventStore{    Task SaveEventsAsync<T>(Guid aggregateId, IEnumerable<IEvent> events, int expectedVersion);    Task<List<IEvent>> GetEventsAsync(Guid aggregateId);}// 简单实现（实际可用 EventStoreDB、Cosmos DB 等）public class SqlEventStore : IEventStore{    public async Task SaveEventsAsync<T>(Guid aggregateId, IEnumerable<IEvent> events, int expectedVersion)    {        // 将事件保存到数据库表        // 包含 AggregateId, AggregateType, Version, EventType, Data (JSON)    }    public async Task<List<IEvent>> GetEventsAsync(Guid aggregateId)    {        // 从数据库加载所有事件并反序列化    }}

HTTP Request (WebApi)       ↓MediatR Request (Command / Query)       ↙                    ↘[Command Handler]        [Query Handler]       ↓                         ↓Domain Layer:              Read-Optimized:- Load Aggregate           - Direct SQL/Dapper- Execute Business Logic   - Cache (Redis)- Validate Rules           - Elasticsearch- Emit Domain Events       - Return DTO       ↓Persist via Repository & UnitOfWork       ↓Publish Domain Events → Event Handlers (Async)

// 在命令处理器中直接调用库存服务await _inventoryService.DeductAsync(orderId); // 失败则订单创建失败

// OrderCreatedEventHandler.cspublic async Task Handle(OrderCreatedEvent notification, CancellationToken ct){    // 发布一个集成事件到消息队列（如 RabbitMQ/Kafka）    await _messageBus.PublishAsync(new InventoryDeductionRequestedEvent(        OrderId: notification.OrderId,        Items: orderItems // 可以从仓储加载    ), ct);}

// GetOrderQueryHandler.cspublic async Task<OrderDto> Handle(GetOrderQuery request, CancellationToken ct){    var cacheKey = $"order:{request.OrderId}";
    // 先查缓存    var cached = await _cache.GetStringAsync(cacheKey, ct);    if (!string.IsNullOrEmpty(cached))    {        return JsonSerializer.Deserialize<OrderDto>(cached);    }    // 缓存未命中，查数据库    var order = await LoadFromDatabase(request.OrderId, ct);
    // 写入缓存（设置过期时间）    await _cache.SetStringAsync(        cacheKey,         JsonSerializer.Serialize(order),         TimeSpan.FromMinutes(10),         ct);    return order;}

eShop.Solution├── eShop.Ordering                # 订单上下文（微服务）│   ├── Ordering.Domain│   ├── Ordering.Application│   ├── Ordering.Infrastructure│   └── Ordering.WebApi│├── eShop.Payment                 # 支付上下文│   ├── Payment.Domain│   ├── Payment.Application│   └── ...│├── eShop.SharedKernel            # 共享内核（ID、时间、异常基类）├── eShop.EventBus                # 事件总线抽象与实现└── eShop.UnitTests               # 共享测试基类

// Features/Orders/CreateOrder/├── CreateOrderCommand.cs├── CreateOrderCommandHandler.cs├── CreateOrderValidator.cs      // FluentValidation└── CreateOrderResponse.cs// Features/Orders/GetOrder/├── GetOrderQuery.cs├── GetOrderQueryHandler.cs└── GetOrderValidator.cs

// Ordering.Domain.Tests/OrderTests.cspublic class OrderTests{    [Fact]    public void CreateOrder_WithValidItems_ShouldSucceed()    {        // Arrange        var customerId = Guid.NewGuid();        var items = new List<OrderItemDto>        {            new(Guid.NewGuid(), 2, 100m)        };        // Act        var order = Order.Create(customerId, items);        // Assert        Assert.Equal(200m, order.TotalAmount);        Assert.Single(order.Items);        Assert.Contains(order.DomainEvents, e => e is OrderCreatedEvent);    }    [Fact]    public void AddItem_WithInvalidQuantity_ShouldThrow()    {        // Arrange        var order = Order.Create(Guid.NewGuid(), new List<OrderItemDto>());        // Act & Assert        Assert.Throws<DomainException>(() =>             order.AddItem(Guid.NewGuid(), 0, 100m));    }}

// Ordering.Application.Tests/CreateOrderCommandHandlerTests.cspublic class CreateOrderCommandHandlerTests : IClassFixture<DatabaseFixture>{    private readonly DatabaseFixture _fixture;    public CreateOrderCommandHandlerTests(DatabaseFixture fixture)    {        _fixture = fixture;    }    [Fact]    public async Task Handle_ValidCommand_ShouldPersistOrder()    {        // Arrange        var handler = _fixture.GetService<IRequestHandler<CreateOrderCommand, Guid>>();        var command = new CreateOrderCommand(            CustomerId: Guid.NewGuid(),            Items: new List<OrderItemDto> { /* ... */ }        );        // Act        var orderId = await handler.Handle(command, CancellationToken.None);        // Assert        var order = await _fixture.OrderRepository.GetByIdAsync(orderId);        Assert.NotNull(order);        Assert.Equal(OrderStatus.Pending, order.Status);    }}

// Command Handler 返回完整 OrderDtopublic async Task<OrderDto> Handle(CreateOrderCommand request, ...){    var order = Order.Create(...);    await _repo.AddAsync(order);    await _uow.CommitAsync();    // 直接返回，避免查询延迟    return _mapper.Map<OrderDto>(order);}

// 当 OrderCreatedEvent 发生时，更新 OrderSummary 表
public class OrderCreatedProjection : INotificationHandler<OrderCreatedEvent>{    private readonly AppDbContext _readContext; // 专用的读库上下文    public async Task Handle(OrderCreatedEvent e, CancellationToken ct)    {        var summary = new OrderSummary        {            OrderId = e.OrderId,            Status = "Pending",            CreatedAt = DateTime.UtcNow,            // ... 其他预计算字段        };        await _readContext.OrderSummaries.AddAsync(summary, ct);        await _readContext.SaveChangesAsync(ct);    }}

// CreateOrderCommand.cs
public record CreateOrderCommand(
    Guid CommandId, // 唯一标识
    Guid CustomerId,
    List<OrderItemDto> Items) : IRequest<Guid>;
// Command Handler
public async Task<Guid> 
Handle(CreateOrderCommand request, CancellationToken ct){
    // 1. 检查该 CommandId 是否已处理
    if (await _commandLog.IsProcessedAsync(request.CommandId, ct))    {        // 返回上次的结果（幂等）        return await _commandLog.GetResultAsync<Guid>(request.CommandId, ct);    }    // 2. 正常处理逻辑    var order = Order.Create(request.CustomerId, request.Items);    await _repo.AddAsync(order);    await _uow.CommitAsync();    
// 3. 记录命令已处理
    await _commandLog.LogProcessedAsync(request.CommandId, order.Id, ct);
    return order.Id;}

// RabbitMQ Consumer 示例
try{
    await HandleEvent(event);
    await model.BasicAck(deliveryTag, false);
 // 确认}
catch (Exception ex){
    
// 记录日志    
// 消息将自动重回队列或进入DLQ
    throw; 
// 不确认，触发重试}

MyApp.Solution

├── MyApp.Ordering          // 订单上下文├── MyApp.Customer          // 客户上下文├── MyApp.Catalog           // 商品目录上下文├── MyApp.SharedKernel

└── MyApp.WebApi            // API 网关入口

public class CatalogAcl : ICatalogService{    private readonly HttpClient _client;    public async Task<ProductDto> GetProductAsync(GuidproductId)    {

        // 调用 Catalog 微服务的 HTTP API        var response = await _client.GetAsync($"/api/products/{productId}");        // 将外部模型转换为内部模型        var external = await response.Content.ReadFromJsonAsync<ExternalProductDto>();        return _mapper.Map<ProductDto>(external);    }}