CQRS架构下读模型数据预计算逻辑的优化实践

先讲个故事。我认识一个做电商的朋友，他们的订单查询页面，一到双十一就卡成PPT，后来查日志发现，一个订单详情要关联十几张表，数据库CPU直接爆掉。后来改用了CQRS，把读和写拆开，查询确实快了不少，但新的问题又来了：那些预计算好的读模型数据，经常要等很久才更新，而且一旦数据量大了，预计算本身就成了新的瓶颈。今天咱们就围绕这个“读模型的数据预计算”聊一聊到底该怎么优化。

一、查询慢，到底慢在哪？

大多数业务系统一开始都是单体应用，写和读共用一套数据库表。订单表、订单明细表、商品表、用户表，一张查询要做一堆join。数据量小的时候无所谓，数据量一上来，光join就能拖垮数据库。

这个时候有人想到了CQRS：把“改数据”和“查数据”分开。改数据走命令模型，负责忠实记录每一次变化；查数据走读模型，专门为查询服务，可以建索引、冗余字段，甚至把多张表的数据合并成一张宽表。这样读模型就能做到“一条SQL搞定”，响应自然快。

这里的“预计算”指的是什么呢？简单说，就是提前把结果算好存起来，而不是等到用户查询那一刻才现算。比如统计某个月的总销售额，与其每次从几千张订单明细里sum，不如每天凌晨更新一个“销售月汇总表”，查询时直接读这张表。

二、CQRS里的读模型和预计算

打个比方。你把CQRS想象成一个餐厅。写模型就是后厨，负责接单、炒菜。读模型就是前厅的菜单牌，上面写着“今日推荐”。如果没有预计算，就相当于每次客人问“今天有什么菜”，服务员还得跑进后厨问一圈才能答。预计算就是提前把菜单牌写好，客人随时看，秒回。

但问题来了：菜单牌写的不对怎么办？后厨菜变了，菜单牌没改，客人就会抱怨。这就是CQRS里说的“最终一致性”——读模型的数据可能暂时不是最新，但过一会儿会追平。而预计算逻辑的优化，主要就是解决两个问题：一是怎么算得快，二是怎么保持数据不会太旧。

三、预计算逻辑的几个“坑”

先说几个我见过的实际坑，很多人都踩过。

第一个坑：在写请求里同步做预计算。假设每下一个新订单，就直接在事务里更新“订单合计表”。订单量小的时候没事，订单量一大，写请求就得等预计算完成才能返回，写链路变长，下单体验变差。

第二个坑：查询时还是现算。有的系统表面上建了“汇总表”，但查询接口里没走汇总表，还是去扫原始订单表。这就等于没预计算，白搭。

第三个坑：更新时机想当然。比如以为每天凌晨跑批就行，但业务是24小时在线，半夜的订单第二天才能看到汇总，老板一看数据不对，立刻炸毛。

第四个坑：全量重算。有时候为了简单，直接truncate汇总表再重新统计所有订单。数据量一大，这个任务跑一两个小时，期间读模型数据还是空的，查询更慢。

四、优化实践：从“伪预计算”到“真预计算”

说了这么多坑，那正确的姿势是什么？我的经验是四个字：按需预计算。别想着把所有数据都算好，只算那些真正高频使用的统计值。然后，用事件驱动来更新预计算结果，别再用定时任务硬扫。

4.1 同步触发预计算（简单但不完美）

先从最基础的开始。我们有一个订单服务，保存订单的同时，更新一个“每日销售统计表”。示例统一使用 C# 12 + .NET 8。

public class OrderService
{
    private readonly AppDbContext _db;

    public OrderService(AppDbContext db)
    {
        _db = db;
    }

    /// <summary>
    /// 创建订单，同时更新每日统计（同步方式）
    /// </summary>
    public async Task CreateOrderAsync(Order order, CancellationToken ct)
    {
        // 先保存订单
        _db.Orders.Add(order);
        await _db.SaveChangesAsync(ct);

        // 再更新统计表 —— 这就是预计算
        var today = DateTime.Today;
        var stat = await _db.DailyStatistics
            .FirstOrDefaultAsync(x => x.Date == today, ct);

        if (stat is null)
        {
            stat = new DailyStatistic { Date = today, OrderCount = 0, TotalAmount = 0m };
            _db.DailyStatistics.Add(stat);
        }

        // 只更新当天的订单数量和金额，这就是增量计算
        stat.OrderCount += 1;
        stat.TotalAmount += order.Amount;

        await _db.SaveChangesAsync(ct);
    }
}

这段代码看起来没问题吧？但它就是最典型的“同步预计算”。订单多的时候，数据库每次要加锁更新同一天的行，并发一上来，锁等待特别严重。所以这种方式适合写入量很小的后台系统，不适合真实电商大流量场景。

4.2 异步事件驱动预计算（推荐）

更合理的做法是把预计算和写操作隔离开。写模型接受命令，保存订单，然后发布一个“订单已创建”事件。读模型订阅这个事件，异步去更新统计表。这样写操作不用等预计算完成，响应自然更快。

我们用RabbitMQ做消息队列，示例代码如下。

// 命令侧：订单创建后，发布事件
public class OrderCommandHandler
{
    private readonly AppDbContext _db;
    private readonly IMessageBus _bus;

    public OrderCommandHandler(AppDbContext db, IMessageBus bus)
    {
        _db = db;
        _bus = bus;
    }

    public async Task Handle(CreateOrderCommand command, CancellationToken ct)
    {
        // 1. 保存订单（命令模型只碰自己的写库）
        var order = new Order
        {
            Id = Guid.NewGuid(),
            CustomerId = command.CustomerId,
            Amount = command.Amount,
            CreatedAt = DateTime.UtcNow
        };
        _db.Orders.Add(order);
        await _db.SaveChangesAsync(ct);

        // 2. 发布事件，通知读模型去预计算
        await _bus.PublishAsync(new OrderCreatedEvent
        {
            OrderId = order.Id,
            Amount = order.Amount,
            CreatedDate = order.CreatedAt.Date
        });
    }
}

// 读模型侧：订阅事件，更新统计表
public class DailyStatisticProjector : IEventHandler<OrderCreatedEvent>
{
    private readonly ReadDbContext _readDb;

    public DailyStatisticProjector(ReadDbContext readDb)
    {
        _readDb = readDb;
    }

    public async Task Handle(OrderCreatedEvent message, CancellationToken ct)
    {
        var today = message.CreatedDate;

        // 读模型库里的统计表
        var stat = await _readDb.DailyStatistics
            .FirstOrDefaultAsync(x => x.Date == today, ct);

        if (stat is null)
        {
            stat = new DailyStatistic { Date = today, OrderCount = 0, TotalAmount = 0m };
            _readDb.DailyStatistics.Add(stat);
        }

        // 增量更新：只加这一单的金额和数量
        stat.OrderCount += 1;
        stat.TotalAmount += message.Amount;

        await _readDb.SaveChangesAsync(ct);
    }
}

你看，左边写订单，右边更新统计，两者不在同一个事务里，靠消息传递。这样做有几点好处：写请求只需要保证自己那一步成功，统计稍有延迟可以接受；万一统计失败，还可以从消息队列里重新消费，不会影响用户下单。

4.3 增量计算，别动不动全量算

上面的代码里面其实已经体现了增量思想：我们读到了当天的统计行，然后只加一单，并没有去刷新整张表。但很多人改造的时候，还是会习惯写成“删掉今天的，再重新汇总今天所有订单”。这就是所谓的全量重算，数据量大时会非常慢。

那什么时候会用到全量重算呢？一般是历史数据修复，或者发现投影出错的时候。正常流程里我们应该使用“幂等”的增量更新方式。什么意思？就是同样的消息消费两次，结果不会错。还是上面的例子，如果消息重复投递，OrderCount会增加两次。解决办法是在事件里带上一个唯一的消息ID，在投影的时候去重。下面是添加了幂等处理的示例：

public class DailyStatisticProjector : IEventHandler<OrderCreatedEvent>
{
    private readonly ReadDbContext _readDb;

    public async Task Handle(OrderCreatedEvent message, CancellationToken ct)
    {
        // 同一个事件消费过就不再重复处理
        var processed = await _readDb.ProcessedMessages
            .AnyAsync(x => x.MessageId == message.MessageId, ct);

        if (processed) return;

        var today = message.CreatedDate;
        var stat = await _readDb.DailyStatistics
            .FirstOrDefaultAsync(x => x.Date == today, ct);

        if (stat is null)
        {
            stat = new DailyStatistic { Date = today, OrderCount = 0, TotalAmount = 0m };
            _readDb.DailyStatistics.Add(stat);
        }

        stat.OrderCount += 1;
        stat.TotalAmount += message.Amount;

        // 记录这条消息已经处理过了
        _readDb.ProcessedMessages.Add(new ProcessedMessage { MessageId = message.MessageId });

        await _readDb.SaveChangesAsync(ct);
    }
}

五、再配个缓存，锦上添花

预计算好的数据，虽然比原始表快，但如果每天都很大，比如统计粒度到了“每小时成交量”，那读模型表本身也会膨胀。这时候可以再加一层缓存，把最常查的预计算结果放到Redis里，设置过期时间，比如30秒。查询时先看缓存，缓存没有再看数据库。

我们用一个简单的接口示例：

public class OrderQueryService
{
    private readonly IRedisClient _redis;
    private readonly ReadDbContext _readDb;

    public OrderQueryService(IRedisClient redis, ReadDbContext readDb)
    {
        _redis = redis;
        _readDb = readDb;
    }

    /// <summary>
    /// 获取某天的销售额，优先走Redis缓存
    /// </summary>
    public async Task<decimal> GetDailySalesAsync(DateTime date, CancellationToken ct)
    {
        var cacheKey = $"sales:daily:{date:yyyyMMdd}";

        // 尝试从缓存读取
        var cached = await _redis.GetAsync<decimal?>(cacheKey);
        if (cached.HasValue) return cached.Value;

        // 缓存没命中，读预计算好的读模型表
        var total = await _readDb.DailyStatistics
            .Where(x => x.Date == date)
            .Select(x => x.TotalAmount)
            .FirstOrDefaultAsync(ct);

        // 写回缓存，30秒过期
        await _redis.SetAsync(cacheKey, total, TimeSpan.FromSeconds(30));

        return total;
    }
}

这样读多写少的场景，Redis大大降低了数据库的压力。注意缓存过期时间不要太长，否则预计算结果更新了，缓存还是旧的，体验会很差。30秒对于大多数报表型数据足够了。

六、聊聊优缺点和适用场景

聊了这么多，那用这套CQRS预计算方案到底图什么？肯定先看优点。

优点很明显：查询性能高，因为读模型是专门为查询设计的宽表，加上预计算和缓存，首屏响应轻轻松松。其次，读写各自独立扩展，比如大促时给读模型多挂几个只读副本，即使写库扛不住，读库也能撑。再次，业务逻辑更清晰，写模型只管命令，读模型只管投影，代码好维护许多。

但缺点也躲不掉。第一个就是最终一致性，刚下的订单不会立刻出现在统计里，可能有秒级甚至分钟级延迟。对延迟敏感的强一致场景，这个方案不适合。第二是系统复杂度上升，消息队列、事件总线、多个数据库，这些都是要花钱花精力运维的。第三是数据修正麻烦，一旦预计算逻辑有bug，你不仅仅要改代码，还要重建历史投影数据，这个成本可能非常高。

所以说，CQRS预计算适合什么样的应用呢？主要是写少读多、查询复杂、对一致性要求不那么苛刻的系统，比如报表统计、用户中心、订单列表这类。如果你是做一个简单的博客留言板，那完全没必要，直接用关系数据库查就好了。

七、这些坑你千万注意

我最后再强调几个注意事项。

第一，别过度设计。预计算不是银弹。如果你的查询只有几千条数据，直接带索引查就很快，非要去搞一套消息队列，反而是给自己找麻烦。

第二，要关注消息的可靠性和顺序。异步预计算依赖消息队列，如果消息丢了，数据就对不上。生产环境必须开启持久化，并且要监控消费积压。有些业务还要求顺序处理，比如同一个用户的订单必须按时间顺序投影，这时可以按用户ID路由到同一个队列分区。

第三，预计算结果要有版本号或时间戳。当投影逻辑升级的时候，老版本的数据要能识别并触发重算。我在项目里习惯给统计表加一个 SchemaVersion 字段，每次逻辑改变就升级版本，然后跑一个一次性迁移任务。

第四，别忘了补偿机制。如果消息消费失败多次，可以放进死信队列，并配套一个定时巡检任务，找出不一致的地方然后重新计算。

八、最后说两句

读模型数据预计算这件事，说难不难，说简单也不简单。核心就是两条：第一，只算高频用的数据；第二，用事件驱动去更新它，别用定时全量跑。在我的经验里，绝大多数查询性能问题都是因为“把所有计算留到了查询时”，而CQRS给了我们一个很好的思路——把计算提前到写入时。后面你再碰到查询慢，不妨先想一想：这些结果能不能提前算好？用什么方式更新合适？希望这篇文章能帮你理清思路。