CQRS 模式实战：从理论到落地，构建可扩展的订单系统

By | 2026年10月6日
0 Comment

为什么你需要关注 CQRS？

在传统的 CRUD 架构中，我们通常使用同一个模型来处理读写操作。但随着业务复杂度增加，读写需求往往不对称：读操作可能远多于写操作，且查询逻辑复杂多变（如多维度筛选、聚合统计），而写操作则注重业务规则和一致性。强行用同一模型应对，会导致代码臃肿、性能瓶颈和难以维护。

CQRS（Command Query Responsibility Segregation，命令查询职责分离）正是解决这一痛点的模式。它主张将命令（写操作）和查询（读操作）分离，使用不同的模型甚至不同的存储，从而让系统更灵活、可扩展。

本文将以一个简单的订单系统为例，带你从零实现 CQRS。你将学到：

  • CQRS 的核心概念与适用场景
  • 如何设计命令与查询模型
  • 使用事件驱动同步读写存储
  • 实战中常见的坑与解决方案
  • 何时不该使用 CQRS

案例背景：订单系统

假设我们有一个订单服务，需求如下：

  • 用户可创建订单、支付订单、取消订单。
  • 后台需要查询订单列表，支持按状态、用户、时间范围过滤，并展示商品详情。
  • 订单量较大，读请求远多于写请求。

传统做法是用一个 Order 实体，同时处理读写。但查询时需要关联商品、用户等信息，写操作需要校验业务规则。随着时间推移，代码会变得混乱。我们决定采用 CQRS。

第一步：定义命令与查询

命令表示“做什么”，通常返回 void 或简单结果；查询表示“要什么”，返回数据。我们使用 TypeScript 来演示，但思想适用于任何语言。


// 命令接口
interface Command {
  readonly type: string;
}

// 查询接口
interface Query {
  readonly type: string;
}

// 具体命令
export class CreateOrderCommand implements Command {
  readonly type = 'CreateOrder';
  constructor(
    public readonly orderId: string,
    public readonly userId: string,
    public readonly items: Array<{ productId: string; quantity: number; price: number }>
  ) {}
}

export class PayOrderCommand implements Command {
  readonly type = 'PayOrder';
  constructor(public readonly orderId: string) {}
}

// 具体查询
export class GetOrdersQuery implements Query {
  readonly type = 'GetOrders';
  constructor(
    public readonly userId?: string,
    public readonly status?: string,
    public readonly startDate?: Date,
    public readonly endDate?: Date
  ) {}
}

注意：命令和查询应该是不可变的数据对象，仅包含执行所需的信息，不包含业务逻辑。

第二步：设计写模型（命令端）

写模型负责处理业务规则，维护领域状态。我们使用领域驱动设计（DDD）中的聚合根概念。


// 订单状态
export enum OrderStatus {
  Created = 'created',
  Paid = 'paid',
  Cancelled = 'cancelled',
}

// 订单聚合根
export class Order {
  private _status: OrderStatus = OrderStatus.Created;
  private _items: Array<{ productId: string; quantity: number; price: number }> = [];

  constructor(public readonly id: string, public readonly userId: string) {}

  get status() { return this._status; }
  get items() { return this._items; }

  addItem(productId: string, quantity: number, price: number) {
    if (this._status !== OrderStatus.Created) {
      throw new Error('只能在创建状态下添加商品');
    }
    if (quantity <= 0) throw new Error('数量必须大于0');
    this._items.push({ productId, quantity, price });
  }

  pay() {
    if (this._status !== OrderStatus.Created) {
      throw new Error('订单状态不允许支付');
    }
    if (this._items.length === 0) {
      throw new Error('订单不能为空');
    }
    this._status = OrderStatus.Paid;
  }

  cancel() {
    if (this._status === OrderStatus.Paid) {
      throw new Error('已支付订单不能取消');
    }
    this._status = OrderStatus.Cancelled;
  }
}

接下来，实现命令处理器。每个命令对应一个处理器，负责加载聚合、执行业务逻辑、保存并发布事件。


// 事件存储接口（简化版）
interface EventStore {
  saveEvents(aggregateId: string, events: any[], expectedVersion: number): Promise<void>;
  getEvents(aggregateId: string): Promise<any[]>;
}

// 事件发布器
interface EventPublisher {
  publish(event: any): Promise<void>;
}

// 命令处理器
export class CreateOrderHandler {
  constructor(
    private eventStore: EventStore,
    private publisher: EventPublisher
  ) {}

  async handle(command: CreateOrderCommand): Promise<void> {
    const order = new Order(command.orderId, command.userId);
    command.items.forEach(item => order.addItem(item.productId, item.quantity, item.price));

    const events = [
      {
        type: 'OrderCreated',
        aggregateId: order.id,
        userId: order.userId,
        items: order.items,
        timestamp: new Date(),
      },
    ];

    await this.eventStore.saveEvents(order.id, events, -1); // -1 表示新聚合
    for (const event of events) {
      await this.publisher.publish(event);
    }
  }
}

💡 提示：这里我们简单地将聚合状态直接保存为事件，实际项目中可以使用事件溯源（Event Sourcing）持久化所有状态变化。为了简化，本文不深入事件溯源。

第三步：设计读模型（查询端）

读模型专注于查询性能，通常使用反范式化的数据表或文档。我们为订单列表创建一个读模型，包含订单基本信息和商品详情。


// 读模型数据库表（这里用内存模拟）
export class OrderReadModel {
  private orders: Map<string, any> = new Map();

  async upsert(order: any) {
    this.orders.set(order.id, order);
  }

  async find(query: GetOrdersQuery): Promise<any[]> {
    let results = Array.from(this.orders.values());
    if (query.userId) {
      results = results.filter(o => o.userId === query.userId);
    }
    if (query.status) {
      results = results.filter(o => o.status === query.status);
    }
    if (query.startDate) {
      results = results.filter(o => o.createdAt >= query.startDate);
    }
    if (query.endDate) {
      results = results.filter(o => o.createdAt <= query.endDate);
    }
    return results;
  }
}

查询处理器直接调用读模型：


export class GetOrdersHandler {
  constructor(private readModel: OrderReadModel) {}

  async handle(query: GetOrdersQuery): Promise<any[]> {
    return this.readModel.find(query);
  }
}

第四步：同步读写模型（事件驱动）

写模型产生事件后，需要更新读模型。我们使用一个事件订阅者来监听事件并更新读模型。


// 事件订阅者
export class OrderProjection {
  constructor(private readModel: OrderReadModel) {}

  async handle(event: any): Promise<void> {
    switch (event.type) {
      case 'OrderCreated':
        await this.readModel.upsert({
          id: event.aggregateId,
          userId: event.userId,
          status: 'created',
          items: event.items,
          createdAt: event.timestamp,
          updatedAt: event.timestamp,
        });
        break;
      case 'OrderPaid':
        const order = await this.readModel.find({} as GetOrdersQuery).then(orders => orders.find(o => o.id === event.aggregateId));
        if (order) {
          order.status = 'paid';
          order.updatedAt = event.timestamp;
          await this.readModel.upsert(order);
        }
        break;
      case 'OrderCancelled':
        // 类似处理
        break;
    }
  }
}

注意：读模型更新可能失败，需要重试机制和幂等处理。事件应至少投递一次，因此订阅者必须幂等。

第五步：组装与测试

现在我们把所有组件组装起来，模拟一次完整的订单流程。


// 简单的内存事件存储和发布器
class InMemoryEventStore implements EventStore {
  private events: Map<string, any[]> = new Map();
  async saveEvents(aggregateId: string, events: any[], expectedVersion: number) {
    const existing = this.events.get(aggregateId) || [];
    if (existing.length !== expectedVersion + 1) {
      throw new Error('并发冲突');
    }
    this.events.set(aggregateId, [...existing, ...events]);
  }
  async getEvents(aggregateId: string) {
    return this.events.get(aggregateId) || [];
  }
}

class SimpleEventPublisher implements EventPublisher {
  private subscribers: Array<(event: any) => void> = [];
  subscribe(handler: (event: any) => void) {
    this.subscribers.push(handler);
  }
  async publish(event: any) {
    for (const sub of this.subscribers) {
      await sub(event);
    }
  }
}

// 组装
const eventStore = new InMemoryEventStore();
const publisher = new SimpleEventPublisher();
const readModel = new OrderReadModel();
const projection = new OrderProjection(readModel);
publisher.subscribe(event => projection.handle(event));

const createOrderHandler = new CreateOrderHandler(eventStore, publisher);
const getOrdersHandler = new GetOrdersHandler(readModel);

// 执行命令
(async () => {
  await createOrderHandler.handle(
    new CreateOrderCommand('order-1', 'user-1', [
      { productId: 'p1', quantity: 2, price: 100 },
    ])
  );

  // 查询
  const orders = await getOrdersHandler.handle(new GetOrdersQuery('user-1'));
  console.log(orders);
})();

运行后，你会看到查询结果包含了刚创建的订单。这证明了读写分离的流程是通的。

常见坑与最佳实践

  1. 最终一致性：CQRS 通常导致读写延迟。对于需要强一致的场景，可以在命令处理后同步更新读模型，但会失去解耦优势。建议在 UI 上做适当处理，如显示“处理中”。
  1. 事件版本管理：事件结构可能变化，需要版本化。建议在事件中包含版本号，并编写向上兼容的转换器。
  1. 读模型重建：当读模型结构变更时，需要从事件存储重放事件来重建。设计时应支持重放，并考虑性能。
  1. 不要过度设计：CQRS 增加复杂度，仅适用于读写负载差异大、查询复杂或需要多视图的场景。简单 CRUD 系统慎用。
  1. 命令校验：命令应在处理器中校验，而不是在命令对象中。命令对象保持纯净。

总结

CQRS 是一种强大的模式，通过分离读写职责，让系统更易于扩展和维护。本文通过订单案例，展示了从命令定义、写模型、读模型到事件同步的完整实现。关键点包括：

  • 命令与查询分离，各司其职
  • 写模型维护业务规则，读模型优化查询
  • 使用事件驱动同步读写模型
  • 注意最终一致性和幂等性

下一步，你可以探索事件溯源（Event Sourcing）与 CQRS 的结合，或使用消息队列（如 Kafka）实现可靠的事件传递。

延伸阅读

  • 领域驱动设计（DDD）中的聚合与领域事件
  • 事件溯源（Event Sourcing）模式
  • 使用 Axon Framework 或 EventStoreDB 实现 CQRS