8.3 读写分离与读模型
数据库的读通常远多于写，于是几乎所有人都会做读写分离：写主库、读从库，主库再把变更异步复制到从库。它换来的读扩展能力很实在，但代价是一个绕不过去的问题：主从复制有延迟。
延迟意味着「写后立即读」可能读到旧值。玩家刚把装备升到 +10、一刷新还是 +9，或刚买的东西在背包里没出现——这类「我明明刚做过」的可见性问题，是读写分离最典型的副作用。本节先把它复现出来，再给三种解法，最后讲读模型。
主从复制延迟：写后立即读可能读到旧值
真实 MySQL 的主从延迟可能几十毫秒到几秒。本节用同一台 Redis 上的两组 key 模拟「主库 + 从库」：写落 master key，复制按需触发，从而精确控制从库「什么时候追上」。下面这段是 L2，真连本机 Redis（26379）：
// file: internal/cache/replica.go
// tier: L2
// verify: go test ./internal/cache/ -run TestReplicationLagStaleReadAndThreeFixes
package cache
import (
"context"
"fmt"
"strconv"
"strings"
"time"
"github.com/redis/go-redis/v9"
)
// Versioned 是带版本号的值：版本号单调递增，用于「写后读」可见性判断。
type Versioned struct {
Version int64
Data string
}
func encode(v Versioned) string { return strconv.FormatInt(v.Version, 10) + "|" + v.Data }
func decode(s string) (Versioned, error) {
i := strings.IndexByte(s, '|')
if i < 0 {
return Versioned{}, fmt.Errorf("cache: malformed payload %q", s)
}
ver, err := strconv.ParseInt(s[:i], 10, 64)
if err != nil {
return Versioned{}, err
}
return Versioned{Version: ver, Data: s[i+1:]}, nil
}
// Replica 用同一台 Redis 上的两组 key 模拟「主库 + 从库」：写落 master key，
// 复制按需触发（SyncOnce）或按 lag 节奏进行，复现真实的主从复制延迟。
type Replica struct {
rdb *redis.Client
ns string
lag time.Duration
}
// NewReplica 构造主从模拟器。
func NewReplica(rdb *redis.Client, ns string, lag time.Duration) *Replica {
return &Replica{rdb: rdb, ns: ns, lag: lag}
}
func (r *Replica) masterKey(id uint64) string { return fmt.Sprintf("%s:m:player:%d", r.ns, id) }
func (r *Replica) slaveKey(id uint64) string { return fmt.Sprintf("%s:s:player:%d", r.ns, id) }
func (r *Replica) verKey(id uint64) string { return fmt.Sprintf("%s:ver:player:%d", r.ns, id) }
// Write 写主库，返回本次写入的版本号（由版本计数器自增得到）。
func (r *Replica) Write(ctx context.Context, id uint64, data string) (int64, error) {
ver, err := r.rdb.Incr(ctx, r.verKey(id)).Result()
if err != nil {
return 0, err
}
if err := r.rdb.Set(ctx, r.masterKey(id), encode(Versioned{Version: ver, Data: data}), 0).Err(); err != nil {
return 0, err
}
return ver, nil
}
// SyncOnce 手动触发一次复制（测试里精确控制从库「什么时候追上」）。
func (r *Replica) SyncOnce(ctx context.Context, id uint64) error {
v, err := r.rdb.Get(ctx, r.masterKey(id)).Result()
if err != nil {
return err
}
return r.rdb.Set(ctx, r.slaveKey(id), v, 0).Err()
}
// ReadMaster 直接读主库（强一致，但压力大）。
func (r *Replica) ReadMaster(ctx context.Context, id uint64) (Versioned, error) {
s, err := r.rdb.Get(ctx, r.masterKey(id)).Result()
if err != nil {
return Versioned{}, err
}
return decode(s)
}
// ReadSlave 读从库（可能读到旧值）。
func (r *Replica) ReadSlave(ctx context.Context, id uint64) (Versioned, error) {
s, err := r.rdb.Get(ctx, r.slaveKey(id)).Result()
if err != nil {
return Versioned{}, err
}
return decode(s)
}
// ReadAfterWriteVersion 是解法二：读从库，若版本落后于本次写入要求，则回主库。
func (r *Replica) ReadAfterWriteVersion(ctx context.Context, id uint64, required int64) (Versioned, error) {
v, err := r.ReadSlave(ctx, id)
if err == nil && v.Version >= required {
return v, nil
}
return r.ReadMaster(ctx, id)
}
// ReadAfterWriteWait 是解法三：轮询从库直到版本追上，超时则回主库。
func (r *Replica) ReadAfterWriteWait(ctx context.Context, id uint64, required int64, timeout time.Duration) (Versioned, error) {
deadline := time.Now().Add(timeout)
for {
v, err := r.ReadSlave(ctx, id)
if err == nil && v.Version >= required {
return v, nil
}
if time.Now().After(deadline) {
return r.ReadMaster(ctx, id)
}
select {
case <-ctx.Done():
return Versioned{}, ctx.Err()
case <-time.After(5 * time.Millisecond):
}
}
}
三种解法对照
// file: internal/cache/replica_test.go
// tier: L2
// verify: go test ./internal/cache/ -run TestReplicationLagStaleReadAndThreeFixes -v
package cache
import (
"context"
"testing"
"time"
)
// TestReplicationLagStaleReadAndThreeFixes 复现「写后立即读可能读到旧值」，
// 并对照三种解法。这是 8.3 的核心可运行断言。
func TestReplicationLagStaleReadAndThreeFixes(t *testing.T) {
rdb := newTestClient(t)
ctx := context.Background()
const playerID = 5005
r := NewReplica(rdb, testNS, 100*time.Millisecond)
// 基线：写 lvl1 并复制到位。
if _, err := r.Write(ctx, playerID, "lvl1"); err != nil {
t.Fatalf("write lvl1: %v", err)
}
if err := r.SyncOnce(ctx, playerID); err != nil {
t.Fatalf("sync: %v", err)
}
// 写 lvl2 后立刻读从库：从库还没复制，读到旧值。
ver, err := r.Write(ctx, playerID, "lvl2")
if err != nil {
t.Fatalf("write lvl2: %v", err)
}
slave, err := r.ReadSlave(ctx, playerID)
if err != nil {
t.Fatalf("read slave: %v", err)
}
if slave.Data != "lvl1" {
t.Fatalf("expected stale lvl1, got %q", slave.Data)
}
t.Logf("reproduced: master=v%d(lvl2) but slave=%q(v%d)", ver, slave.Data, slave.Version)
// 解法一：写后强制读主。
if m, err := r.ReadMaster(ctx, playerID); err != nil || m.Data != "lvl2" {
t.Fatalf("fix1 read-master: err=%v data=%q", err, m.Data)
}
// 解法二：版本令牌，从库落后则回主。
if v, err := r.ReadAfterWriteVersion(ctx, playerID, ver); err != nil || v.Data != "lvl2" {
t.Fatalf("fix2 version-token: err=%v data=%q", err, v.Data)
}
// 解法三：等待从库追上（后台延迟复制一次）。
go func() {
time.Sleep(30 * time.Millisecond)
_ = r.SyncOnce(ctx, playerID)
}()
if v, err := r.ReadAfterWriteWait(ctx, playerID, ver, time.Second); err != nil || v.Data != "lvl2" {
t.Fatalf("fix3 wait: err=%v data=%q", err, v.Data)
}
}
本机实测输出：
reproduced: master=v2(lvl2) but slave="lvl1"(v1)
写 lvl2（版本 2）之后立刻读从库，从库还停在 lvl1（版本 1）——这就是「写后读旧」。三种解法：
| 解法 | 做法 | 代价 |
|---|---|---|
| 读主库 | 写后强制读主 | 主库压力大，但强一致 |
| 版本令牌 | 记住写返回的版本，读从库版本落后就回主 | 需把版本一路传到读请求 |
| 等待追上 | 轮询从库直到版本满足，超时回主 | 读延迟不可控 |
读模型：写侧落事件、读侧构建宽表
// file: internal/cache/readmodel.go
// tier: L2
// verify: go test ./internal/cache/ -run TestCQRSReadModelProjection
package cache
import (
"context"
"encoding/json"
"fmt"
"github.com/redis/go-redis/v9"
)
// Event 是写侧产生的事件。CQRS 轻量版里，写侧只追加事件，不做读优化。
type Event struct {
Type string `json:"type"`
Value string `json:"value"`
}
// WriteModel 是写侧：把事件追加到该角色的私有事件流。
type WriteModel struct {
rdb *redis.Client
ns string
}
// NewWriteModel 构造写侧。
func NewWriteModel(rdb *redis.Client, ns string) *WriteModel { return &WriteModel{rdb: rdb, ns: ns} }
func (w *WriteModel) streamKey(id uint64) string {
return fmt.Sprintf("%s:stream:player:%d", w.ns, id)
}
// Append 追加一个事件，返回事件流当前长度。
func (w *WriteModel) Append(ctx context.Context, id uint64, e Event) (int64, error) {
b, err := json.Marshal(e)
if err != nil {
return 0, err
}
return w.rdb.RPush(ctx, w.streamKey(id), b).Result()
}
// Events 按顺序读回该角色的事件流。
func (w *WriteModel) Events(ctx context.Context, id uint64) ([]Event, error) {
raw, err := w.rdb.LRange(ctx, w.streamKey(id), 0, -1).Result()
if err != nil {
return nil, err
}
out := make([]Event, 0, len(raw))
for _, s := range raw {
var e Event
if err := json.Unmarshal([]byte(s), &e); err != nil {
return nil, err
}
out = append(out, e)
}
return out, nil
}
// Projector 是读侧投影器：重放事件流，把它折叠成一张「宽表」（Redis Hash）。
type Projector struct {
rdb *redis.Client
wm *WriteModel
ns string
}
// NewProjector 构造投影器。
func NewProjector(rdb *redis.Client, wm *WriteModel, ns string) *Projector {
return &Projector{rdb: rdb, wm: wm, ns: ns}
}
// Project 重放事件流，重建宽表，返回应用的事件条数。
func (p *Projector) Project(ctx context.Context, id uint64) (int, error) {
events, err := p.wm.Events(ctx, id)
if err != nil {
return 0, err
}
key := fmt.Sprintf("%s:wide:player:%d", p.ns, id)
if err := p.rdb.Del(ctx, key).Err(); err != nil {
return 0, err
}
for _, e := range events {
if err := p.rdb.HSet(ctx, key, e.Type, e.Value).Err(); err != nil {
return 0, err
}
}
return len(events), nil
}
// ReadModel 是读侧查询接口：只读宽表，不碰事件流。
type ReadModel struct {
rdb *redis.Client
ns string
}
// NewReadModel 构造读侧。
func NewReadModel(rdb *redis.Client, ns string) *ReadModel { return &ReadModel{rdb: rdb, ns: ns} }
// Get 读取宽表全部字段。
func (rm *ReadModel) Get(ctx context.Context, id uint64) (map[string]string, error) {
return rm.rdb.HGetAll(ctx, fmt.Sprintf("%s:wide:player:%d", rm.ns, id)).Result()
}
WriteModel.Append 追加事件到角色私有事件流，Projector.Project 重放事件流折叠成宽表，ReadModel.Get 只读宽表。投影是幂等重建的——重放同一批事件得到同一张宽表，所以投影器可随时重跑。对应测试 TestCQRSReadModelProjection 断言：投影前宽表为空，投影后字段齐备，追加事件重投影后宽表更新（测试助手见 8.1 的 internal/cache/helpers_test.go）。
本节小结 / 实测结论
- 读写分离换读扩展能力，代价是主从复制延迟带来的「写后读旧」；本机实测（同机相对量级，绝对数值不可移植）写 lvl2 后立即读从库仍读到 lvl1。
- 三种解法：读主库（强一致、压主）、版本令牌（推荐）、等待追上（延迟不可控）。
- CQRS 轻量版：写侧落事件、读侧投影宽表，投影幂等可重建。
阅读导航：上一节：缓存预热与击穿防护 · 下一节：角色数据模型 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。