参考

Snapshotting

JSON 快照：实现 SnapshotVersion、MarshalSnapshot 与 UnmarshalSnapshot，Repository 如何加载，后台 Snapshotter，以及版本演进。

Snapshotting

包：snapshot、eventstore

快照是聚合在某个序号时状态的 JSON 副本。Repository 先加载最新快照，再只加载其后的事件。它用来加速长生命周期聚合的加载——例如事件流特别长的账户。快照只是缓存，事件仍然是事实来源。

让聚合支持快照

实现三个方法：

type Snapshotable interface {
    SnapshotVersion() int
    MarshalSnapshot() (json.RawMessage, error)
    UnmarshalSnapshot(data json.RawMessage) error
}

bank 的 Account：

const accountSnapshotVersion = 1

type accountSnap struct {
    Owner    string      `json:"owner"`
    Currency string      `json:"currency"`
    Balance  types.Money `json:"balance"`
    Open     bool        `json:"open"`
}

func (a *Account) SnapshotVersion() int { return accountSnapshotVersion }

func (a *Account) MarshalSnapshot() (json.RawMessage, error) {
    return json.Marshal(accountSnap{
        Owner: a.Owner, Currency: a.Currency, Balance: a.Balance, Open: a.Open,
    })
}

func (a *Account) UnmarshalSnapshot(data json.RawMessage) error {
    var s accountSnap
    if err := json.Unmarshal(data, &s); err != nil {
        return err
    }
    a.Owner, a.Currency, a.Balance, a.Open = s.Owner, s.Currency, s.Balance, s.Open
    if a.Open {
        a.refreshUniqueKeys()
    }
    return nil
}

请用专门的快照结构体，不要直接序列化聚合本身。这样存储形状是显式的，也可以按自己的节奏演进。要恢复 applier 通常会推导的内容（例如唯一键），因为 UnmarshalSnapshot 取代了对早期事件的回放。

快照格式只支持 JSON。

store 保存什么

type Snapshot struct {
    AggregateID     uuid.UUID
    SequenceNumber  int64
    SnapshotType    string // 通常是聚合类型名
    SnapshotVersion int
    JSON            json.RawMessage
    CreatedAt       time.Time
}

eventstore.SnapshotStore 提供 SaveSnapshot、LoadLatestSnapshot、MarkSnapshotTaken 与 ListNeedingSnapshots。memory 与 Postgres store 都实现了它。Postgres 使用 snapshot_records 和 aggregates_that_need_snapshots 两张表。

加载

把 Repo.Snapshots 设为快照存储。Load 时，Repository 会：

用工厂构造空聚合。
查询 SnapshotType 与 SnapshotVersion 都匹配的最新快照。
找到则调用 UnmarshalSnapshot，并把聚合版本设为快照序号。
加载事件流，只回放该序号之后的事件。

版本不同的快照会被忽略，聚合回退为加载全部事件。

生成快照

追加使事件流超过阈值时，store 会把该聚合标记为需要快照。Postgres 里阈值是聚合行上的 snapshot_threshold，默认 100；memory store 用 SetSnapshotThreshold(n) 设置。

snapshot.Snapshotter 在后台生成快照：

sn := &snapshot.Snapshotter{
    Store:     store,
    Snapshots: store,
    Factories: map[string]aggregate.Factory{"bank.Account": func(id uuid.UUID) aggregate.Root { return NewAccount(id) }},
    ResolveType: func(ctx context.Context, id uuid.UUID) (string, error) { /* 返回 id 对应的聚合类型 */ },
    Interval:  time.Minute,
    BatchSize: 10,
}
n, err := sn.RunOnce(ctx) // 或 sn.Run(ctx)

对每个需要快照的聚合，它加载事件流、重建聚合、调用 snapshot.Take、保存结果并标记完成。聚合不是 Snapshotable 时，Take 返回错误。

聚合形状变化时

快照的存储形状或状态含义发生不兼容变化时：

提升 SnapshotVersion()。
旧快照不再匹配，会被忽略。
Snapshotter 会逐步重新生成。

也可以直接删除旧的 snapshot_records 行。快照总能由事件重新生成，所以可疑或被污染的快照不是致命问题。从备份恢复之后，也要清掉不兼容的快照。见备份与恢复与事件演进。

何时使用

先测量到长事件流加载变慢，再加快照。短事件流本来就很快，而快照会多出一条需要保持正确的代码路径。

Snapshotting

让聚合支持快照

store 保存什么

加载

生成快照

聚合形状变化时

何时使用

相关