《MMO 服务器开发》8.1 Redis 使用与失效策略

// file: internal/cache/key.go
// tier: L1
// verify: go test ./internal/cache/ -run TestKeyBuilder
package cache

import "fmt"

// DefaultNamespace 是本项目缓存键的统一前缀，避免与其它系统共用 Redis 时互相污染。
const DefaultNamespace = "aegis"

// KeyBuilder 把实体映射成「命名空间 : 版本 : 类型 : ID」四段式缓存键。
// 版本号用于整体失效：结构变更时版本 +1，所有读路径自动落到新键，旧键靠 TTL 淘汰。
type KeyBuilder struct {
	ns      string
	version int
}

// NewKeyBuilder 构造键构建器；ns 为空回退默认，version <= 0 归一为 1。
func NewKeyBuilder(ns string, version int) *KeyBuilder {
	if ns == "" {
		ns = DefaultNamespace
	}
	if version <= 0 {
		version = 1
	}
	return &KeyBuilder{ns: ns, version: version}
}

// Key 生成形如 aegis:v1:player:1001 的缓存键。
func (b *KeyBuilder) Key(kind string, id uint64) string {
	return fmt.Sprintf("%s:v%d:%s:%d", b.ns, b.version, kind, id)
}

// Version 返回当前版本号。
func (b *KeyBuilder) Version() int { return b.version }

// Bump 返回版本号 +1 的新构建器（整体失效）。
func (b *KeyBuilder) Bump() *KeyBuilder {
	return &KeyBuilder{ns: b.ns, version: b.version + 1}
}

// file: internal/cache/invalidate.go
// tier: L2
// verify: go test ./internal/cache/ -run TestConcurrentEventuallyConsistent
package cache

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// Store 抽象「源」（线上是 MySQL/Postgres）。本节用另一个 Redis key 模拟，
// 让缓存与源落在同一台机器上，便于复现并发交错。
type Store interface {
	Load(ctx context.Context, id uint64) (string, error)
	Save(ctx context.Context, id uint64, val string) error
}

// RedisStore 用独立命名空间下的另一个 key 模拟数据库表。
type RedisStore struct {
	rdb *redis.Client
	ns  string
}

// NewRedisStore 构造一个以 Redis 为后端的模拟源。
func NewRedisStore(rdb *redis.Client, ns string) *RedisStore {
	return &RedisStore{rdb: rdb, ns: ns}
}

func (s *RedisStore) srcKey(id uint64) string { return fmt.Sprintf("%s:src:player:%d", s.ns, id) }

// Load 读取源；源不存在时返回 redis.Nil。
func (s *RedisStore) Load(ctx context.Context, id uint64) (string, error) {
	return s.rdb.Get(ctx, s.srcKey(id)).Result()
}

// Save 写入源。
func (s *RedisStore) Save(ctx context.Context, id uint64, val string) error {
	return s.rdb.Set(ctx, s.srcKey(id), val, 0).Err()
}

// Cache 是 cache-aside 模式的最小实现。
type Cache struct {
	rdb *redis.Client
	kb  *KeyBuilder
	ttl time.Duration
}

// NewCache 构造缓存；ttl 为回填值的存活时间。
func NewCache(rdb *redis.Client, kb *KeyBuilder, ttl time.Duration) *Cache {
	return &Cache{rdb: rdb, kb: kb, ttl: ttl}
}

// ErrNotFound 表示源里也没有这条数据。
var ErrNotFound = errors.New("cache: not found")

// GetOrLoad 是读路径：先查缓存，未命中回源并回填。
func (c *Cache) GetOrLoad(ctx context.Context, s Store, id uint64) (string, error) {
	key := c.kb.Key("player", id)
	v, err := c.rdb.Get(ctx, key).Result()
	if err == nil {
		return v, nil
	}
	if err != redis.Nil {
		return "", err
	}
	val, err := s.Load(ctx, id)
	if err != nil {
		if err == redis.Nil {
			return "", ErrNotFound
		}
		return "", err
	}
	_ = c.rdb.Set(ctx, key, val, c.ttl).Err()
	return val, nil
}

// WriteThenEvict：先写源、再删缓存。绝大多数场景推荐的顺序。
func (c *Cache) WriteThenEvict(ctx context.Context, s Store, id uint64, val string) error {
	if err := s.Save(ctx, id, val); err != nil {
		return err
	}
	return c.rdb.Del(ctx, c.kb.Key("player", id)).Err()
}

// EvictThenWrite：先删缓存、再写源。
func (c *Cache) EvictThenWrite(ctx context.Context, s Store, id uint64, val string) error {
	if err := c.rdb.Del(ctx, c.kb.Key("player", id)).Err(); err != nil {
		return err
	}
	return s.Save(ctx, id, val)
}

// DelayedDoubleDelete：写源前后各删一次，第二次延迟 delay 执行，覆盖
// 「读者在第一次删之后、写源之前把旧值回填」的窗口。
func (c *Cache) DelayedDoubleDelete(ctx context.Context, s Store, id uint64, val string, delay time.Duration) error {
	key := c.kb.Key("player", id)
	if err := c.rdb.Del(ctx, key).Err(); err != nil {
		return err
	}
	if err := s.Save(ctx, id, val); err != nil {
		return err
	}
	go func() {
		t := time.NewTimer(delay)
		defer t.Stop()
		<-t.C
		_ = c.rdb.Del(context.Background(), key).Err()
	}()
	return nil
}

// file: internal/cache/invalidate_test.go
// tier: L2
// verify: go test ./internal/cache/ -run TestConcurrentEventuallyConsistent -v
package cache

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func TestKeyBuilder(t *testing.T) {
	kb := NewKeyBuilder(testNS, 3)
	if got := kb.Key("player", 1001); got != testNS+":v3:player:1001" {
		t.Fatalf("unexpected key: %s", got)
	}
	if kb.Bump().Version() != 4 {
		t.Fatalf("bump version: got %d want 4", kb.Bump().Version())
	}
}

// TestConcurrentEventuallyConsistent：并发读写停止后，经一次「写源 + 延迟双删」，
// 缓存与源最终一致。这是 8.1 的核心可运行断言。
func TestConcurrentEventuallyConsistent(t *testing.T) {
	rdb := newTestClient(t)
	ctx := context.Background()
	kb := NewKeyBuilder(testNS, 1)
	c := NewCache(rdb, kb, 30*time.Second)
	s := NewRedisStore(rdb, testNS)

	const playerID = 1001
	if err := s.Save(ctx, playerID, "v0"); err != nil {
		t.Fatalf("seed: %v", err)
	}

	stop := make(chan struct{})
	var readers, writers sync.WaitGroup
	var reads, writes int64

	for i := 0; i < 8; i++ {
		readers.Add(1)
		go func() {
			defer readers.Done()
			for {
				select {
				case <-stop:
					return
				default:
				}
				if _, err := c.GetOrLoad(ctx, s, playerID); err == nil {
					atomic.AddInt64(&reads, 1)
				}
			}
		}()
	}

	for w := 0; w < 4; w++ {
		writers.Add(1)
		go func(base int) {
			defer writers.Done()
			for i := 0; i < 50; i++ {
				v := fmt.Sprintf("w%d-%d", base, i)
				if err := c.WriteThenEvict(ctx, s, playerID, v); err == nil {
					atomic.AddInt64(&writes, 1)
				}
			}
		}(w)
	}

	writers.Wait()
	close(stop)
	readers.Wait()

	// 收尾：写终值 + 延迟双删，等第二次删除生效后再读，断言缓存与源一致。
	const final = "final-value"
	if err := c.DelayedDoubleDelete(ctx, s, playerID, final, 50*time.Millisecond); err != nil {
		t.Fatalf("final write: %v", err)
	}
	time.Sleep(150 * time.Millisecond)

	got, err := c.GetOrLoad(ctx, s, playerID)
	if err != nil {
		t.Fatalf("read final: %v", err)
	}
	if got != final {
		t.Fatalf("cache/source inconsistent: got %q want %q", got, final)
	}
	t.Logf("reads=%d writes=%d final=%s", atomic.LoadInt64(&reads), atomic.LoadInt64(&writes), got)
}

reads=355 writes=200 final=final-value