4. CQRS 架构模式：命令与查询分离、读模型与写模型、在 Go 中构建 CQRS 的实践

CQRS（Command Query Responsibility Segregation，命令查询职责分离）是事件溯源架构中最核心的配套模式。它将系统的数据操作接口拆分为两个独立的通道：命令端（写模型）负责处理变更，查询端（读模型）负责返回数据。这种分离使得 Go 并发模型能够针对读写两端分别优化，避免单一数据模型在并发场景下的锁竞争和性能瓶颈。

4.1 核心概念：命令与查询的分离原则

CQRS 的核心理念源于 Bertrand Meyer 在《面向对象软件构造》中提出的“命令-查询分离”（CQS）原则：一个方法要么是修改状态的命令，要么是返回数据的查询，不能同时兼具两者。CQRS 将这一原则从方法级别提升到架构级别，形成两条独立的数据管道。

4.2 读模型与写模型的职责边界

在 Go 的 CQRS 实现中，写模型和读模型各自拥有独立的数据结构和存储策略，互不干扰。

写模型（Write Model）

• 聚合根（Aggregate Root）：封装业务逻辑，保证聚合内的数据一致性。
• 事件存储（Event Store）：仅追加（append-only）的事件日志，是写模型的唯一事实来源。
• 命令处理器（Command Handler）：验证命令、调用聚合方法、生成事件并持久化。
• 并发控制：每个聚合实例在 Go 中通常由一个独立的 goroutine 或带版本号的乐观锁保护。

读模型（Read Model）

• 投影器（Projector）：订阅事件流，将事件转换为查询友好的视图。
• 视图存储（View Store）：可以是关系表、Redis 缓存、Elasticsearch 索引或内存 map。
• 查询处理器（Query Handler）：无状态、只读，直接从视图存储返回数据。
• 并发安全：投影器通常使用单 goroutine 顺序消费事件，或使用分片 key 保证同一聚合的事件顺序。

4.3 在 Go 中构建 CQRS 的实践

以下是一个基于 Go 的轻量级 CQRS 实现示例，包含命令端、事件存储和查询端的核心骨架。

4.3.1 定义命令与事件

// command.go
type CreateOrderCommand struct {
    OrderID string
    UserID  string
    Items   []OrderItem
    Total   float64
}

type OrderItem struct {
    ProductID string
    Quantity  int
    Price     float64
}

// event.go
type OrderCreatedEvent struct {
    OrderID   string
    UserID    string
    Items     []OrderItem
    Total     float64
    CreatedAt time.Time
}

type OrderShippedEvent struct {
    OrderID   string
    ShippedAt time.Time
}

4.3.2 写模型：聚合根与命令处理器

// aggregate.go
type OrderAggregate struct {
    ID      string
    Version int
    Status  string
    Events  []interface{}
}

func (a *OrderAggregate) HandleCreate(cmd CreateOrderCommand) error {
    if a.Version != 0 {
        return errors.New("order already exists")
    }
    event := OrderCreatedEvent{
        OrderID:   cmd.OrderID,
        UserID:    cmd.UserID,
        Items:     cmd.Items,
        Total:     cmd.Total,
        CreatedAt: time.Now(),
    }
    a.apply(event)
    return nil
}

func (a *OrderAggregate) apply(event interface{}) {
    switch e := event.(type) {
    case OrderCreatedEvent:
        a.ID = e.OrderID
        a.Status = "created"
    }
    a.Version++
    a.Events = append(a.Events, event)
}

// command_handler.go
type OrderCommandHandler struct {
    store EventStore
}

func (h *OrderCommandHandler) Handle(cmd CreateOrderCommand) error {
    agg := &OrderAggregate{ID: cmd.OrderID}
    if err := agg.HandleCreate(cmd); err != nil {
        return err
    }
    // 乐观锁：期望版本号为 0
    return h.store.Save(cmd.OrderID, 0, agg.Events)
}

4.3.3 事件存储：仅追加的并发安全实现

// event_store.go
type EventStore interface {
    Save(aggregateID string, expectedVersion int, events []interface{}) error
    Load(aggregateID string) ([]interface{}, error)
}

type InMemoryEventStore struct {
    mu       sync.RWMutex
    events   map[string][]interface{} // aggregateID -> events
    versions map[string]int           // aggregateID -> current version
}

func (s *InMemoryEventStore) Save(aggregateID string, expectedVersion int, events []interface{}) error {
    s.mu.Lock()
    defer s.mu.Unlock()

    currentVersion := s.versions[aggregateID]
    if currentVersion != expectedVersion {
        return fmt.Errorf("concurrency conflict: expected version %d, got %d", expectedVersion, currentVersion)
    }

    s.events[aggregateID] = append(s.events[aggregateID], events...)
    s.versions[aggregateID] = currentVersion + len(events)
    return nil
}

func (s *InMemoryEventStore) Load(aggregateID string) ([]interface{}, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    return s.events[aggregateID], nil
}

4.3.4 读模型：投影器与查询处理器

// projector.go
type OrderProjector struct {
    viewStore *sync.Map // orderID -> OrderView
}

type OrderView struct {
    OrderID string
    UserID  string
    Status  string
    Total   float64
    Items   []OrderItem
}

func (p *OrderProjector) Project(event interface{}) {
    switch e := event.(type) {
    case OrderCreatedEvent:
        view := OrderView{
            OrderID: e.OrderID,
            UserID:  e.UserID,
            Status:  "created",
            Total:   e.Total,
            Items:   e.Items,
        }
        p.viewStore.Store(e.OrderID, view)
    case OrderShippedEvent:
        if v, ok := p.viewStore.Load(e.OrderID); ok {
            view := v.(OrderView)
            view.Status = "shipped"
            p.viewStore.Store(e.OrderID, view)
        }
    }
}

// query_handler.go
type OrderQueryHandler struct {
    viewStore *sync.Map
}

func (h *OrderQueryHandler) GetOrder(orderID string) (OrderView, error) {
    v, ok := h.viewStore.Load(orderID)
    if !ok {
        return OrderView{}, errors.New("order not found")
    }
    return v.(OrderView), nil
}

func (h *OrderQueryHandler) ListOrdersByUser(userID string) []OrderView {
    var result []OrderView
    h.viewStore.Range(func(key, value interface{}) bool {
        view := value.(OrderView)
        if view.UserID == userID {
            result = append(result, view)
        }
        return true
    })
    return result
}

4.4 并发回放与投影优化

在 Go 中实现 CQRS 时，投影器的并发回放性能直接影响查询端的延迟。以下是三种常见的投影优化策略：

4.5 最终一致性处理与补偿

CQRS 的读写分离必然引入最终一致性窗口。在 Go 中处理这种不一致性需要明确的策略：

• 命令端返回事件 ID：客户端在收到命令成功响应后，通过事件 ID 轮询查询端，直到视图更新。
• 投影延迟监控：在投影器中记录当前已处理的事件序列号，暴露为 Prometheus 指标，用于监控延迟。
• 补偿命令：当投影失败或业务规则被违反时，发送补偿命令（如 CancelOrderCommand）回写模型，投影器消费补偿事件后更新视图。

// 投影延迟监控示例
type MonitoredProjector struct {
    projector     *OrderProjector
    processedSeq  int64
    lastProcessed time.Time
}

func (m *MonitoredProjector) Project(event interface{}, seq int64) {
    m.projector.Project(event)
    atomic.StoreInt64(&m.processedSeq, seq)
    atomic.StoreInt64((*int64)(&m.lastProcessed), time.Now().UnixNano())
}

// 暴露指标（伪代码）
// prometheus.NewGauge("projector_lag_events", "事件滞后数量")
// prometheus.NewGauge("projector_lag_seconds", "时间滞后秒数")

4.6 本章小结