mcachedb

package module
v1.0.22 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 24, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

README

mcachedb

Go Reference Go Report Card License

mcachedb 是一个 Go 语言三级缓存库，面向读多写少、需要最终一致或强一致可选的业务场景。

整体架构

flowchart LR
    App[业务代码] --> MC[MultiCache]
    MC --> L1[L1 进程内内存 map]
    MC --> L2[L2 Redis 共享缓存]
    MC --> L3[L3 数据库 MySQL / 任意 DBStore]
    L2 -.->|故障自动降级| L3
  • 读路径：L1 → L2 → L3，逐级穿透并自动回填上层缓存。
  • 写路径：支持异步批量、同步直写、WriteThrough、CacheAside 四种模式。
  • L2 容错：Redis 故障自动降级为 L1 + L3，后台定时探测恢复。
  • 事务：单缓存本地事务 + 跨缓存最大努力分布式事务。

目录

安装

go get github.com/quietking0312/component/mcachedb

依赖：

  • github.com/jmoiron/sqlx
  • github.com/redis/go-redis/v9
  • github.com/stretchr/testify（仅测试）

快速开始

1. 定义实体

BaseEntity 只提供 CacheKey、Version、IsDeleted、Copy 等基础能力，不提供 Marshal/Unmarshal 的默认实现。业务实体必须自己实现序列化，否则编译不通过。

package main

import (
    "encoding/json"

    "github.com/quietking0312/component/mcachedb"
)

type Player struct {
    mcachedb.BaseEntity
    Name  string `json:"name"`
    Level int    `json:"level"`
}

func NewPlayer(id, name string, level int) *Player {
    return &Player{
        BaseEntity: *mcachedb.NewBaseEntity(id),
        Name:       name,
        Level:      level,
    }
}

func (p *Player) Copy() mcachedb.Entity {
    return &Player{
        BaseEntity: *p.BaseEntity.Copy(),
        Name:       p.Name,
        Level:      p.Level,
    }
}

func (p *Player) Marshal() ([]byte, error) { return json.Marshal(p) }
func (p *Player) Unmarshal(b []byte) error { return json.Unmarshal(b, p) }

如果你不想手写完整 Entity，可以直接使用内置的 泛型实体 GenericEntity[T]。

2. 准备 L3 存储

可以直接实现 DBStore 接口，也可以使用内置的 SQLXStore：

// 方式一：内置 SQLXStore（自动建表、通用 JSON 列）
l3, err := mcachedb.NewSQLXStore("root:pass@tcp(127.0.0.1:3306)/game", &Player{})
if err != nil {
    panic(err)
}

// 方式二：从已有 *sqlx.DB 创建
l3, err = mcachedb.NewSQLXStoreFromDB(db, &Player{})
3. 准备 L2 存储（可选）

可以直接实现 L2Store 接口，也可以使用内置的 RedisStore：

l2, err := mcachedb.NewRedisStore(&mcachedb.RedisConfig{
    Addr:       "127.0.0.1:6379",
    KeyPrefix:  "player:",
    DefaultTTL: 10 * time.Minute,
}, &Player{}) // entityType 用于反序列化时保持具体类型
if err != nil {
    panic(err)
}

传 nil 则退化为 L1 + L3 两级缓存。

4. 创建 MultiCache
cache, err := mcachedb.NewMultiCache(l3, l2, &mcachedb.MultiCacheConfig{
    L1MaxSize:      100_000,
    FlushInterval:  1 * time.Second,       // L1 -> L3 刷盘间隔
    SyncInterval:   500 * time.Millisecond, // L1 -> L2 同步间隔
    WriteToL2OnSet: false,                  // Set 时是否同步写 L2
})
if err != nil {
    panic(err)
}
defer cache.Close()
5. 读写删
// 写
cache.Set(NewPlayer("u1", "Alice", 10))

// 读：L1 → L2 → L3，命中后自动回填
entity, err := cache.Get("u1")
if err != nil {
    panic(err)
}
player := entity.(*Player)

// 批量读
m, err := cache.MGet([]string{"u1", "u2", "u3"})

// 删
cache.Delete("u1")

// 立即把所有脏数据刷到 L3
cache.Flush()

架构与数据流转

读路径
sequenceDiagram
    participant App
    participant L1 as L1 内存
    participant L2 as L2 Redis
    participant L3 as L3 数据库
    App->>L1: Get(key)
    alt L1 命中
        L1-->>App: 返回实体
    else L1 miss
        L1->>L2: Get(key)
        alt L2 命中
            L2-->>L1: 回填 L1
            L2-->>App: 返回实体
        else L2 miss / L2 故障
            L2->>L3: Get(key)
            L3-->>L2: 回填 L2
            L3-->>L1: 回填 L1
            L3-->>App: 返回实体
        end
    end
写路径：异步模式（默认）
sequenceDiagram
    participant App
    participant L1 as L1 内存
    participant L2 as L2 Redis
    participant L3 as L3 数据库
    App->>L1: Set(entity)
    L1-->>App: 立即返回
    Note over L1: 后台 syncToL2Loop
    L1->>L2: MSet(batch)
    Note over L1: 后台 flushToL3Loop
    L1->>L3: BatchInsert/Update(batch)
写路径：CacheAside 模式
sequenceDiagram
    participant App
    participant L1 as L1 内存
    participant L2 as L2 Redis
    participant L3 as L3 数据库
    App->>L3: Insert/Update(entity)
    L3-->>App: OK
    App->>L1: Remove(key)
    App->>L2: Delete(key)
    Note over L1,L2: 下次读取时从 L3 回填

写入模式

模式 常量 行为 适用场景
异步（默认） WriteModeAsync 写 L1 返回，后台批量 flush L3，L2 定期 sync 高吞吐、可接受秒级丢失
同步 WriteModeSync 写 L1 同时同步写 L3 需要每次写都落库
直写 WriteModeWriteThrough 先写 L3，成功后再写 L1 写少读多，L1 与 L3 强一致
旁路 WriteModeCacheAside 先写 L3，成功后删除 L1/L2；下次读回填 业务最常用的强一致模式 
cache, _ := mcachedb.NewMultiCache(l3, l2, &mcachedb.MultiCacheConfig{
    WriteMode: mcachedb.WriteModeCacheAside,
})
异步模式的数据安全边界

默认配置下，极端情况（进程崩溃）最多丢失 FlushInterval 内的数据。若不能容忍，请选择 WriteModeSync、WriteModeWriteThrough 或 WriteModeCacheAside，亦或在关键写后手动调用 cache.Flush()。

自定义 L2/L3 实现

mcachedb 通过接口解耦 L2/L3，你可以完全自定义：

// 自定义 L3：例如对接已有分库分表、MongoDB 等
type MyDBStore struct{}

func (s *MyDBStore) Get(ctx context.Context, key string) (mcachedb.Entity, error) { ... }
func (s *MyDBStore) MGet(ctx context.Context, keys []string) (map[string]mcachedb.Entity, error) { ... }
func (s *MyDBStore) Insert(ctx context.Context, e mcachedb.Entity) error { ... }
func (s *MyDBStore) Update(ctx context.Context, e mcachedb.Entity) error { ... }
func (s *MyDBStore) Delete(ctx context.Context, key string) error { ... }
func (s *MyDBStore) BatchInsert(ctx context.Context, es []mcachedb.Entity) error { ... }
func (s *MyDBStore) BatchUpdate(ctx context.Context, es []mcachedb.Entity) error { ... }
func (s *MyDBStore) BatchDelete(ctx context.Context, keys []string) error { ... }
func (s *MyDBStore) Close() error { ... }

// 自定义 L2：例如本地 RocksDB、Memcached 等
type MyL2Store struct{}

func (s *MyL2Store) Get(ctx context.Context, key string) (mcachedb.Entity, error) { ... }
func (s *MyL2Store) MGet(ctx context.Context, keys []string) (map[string]mcachedb.Entity, error) { ... }
func (s *MyL2Store) Set(ctx context.Context, e mcachedb.Entity) error { ... }
func (s *MyL2Store) MSet(ctx context.Context, es []mcachedb.Entity) error { ... }
func (s *MyL2Store) Delete(ctx context.Context, key string) error { ... }
func (s *MyL2Store) Ping(ctx context.Context) error { ... }
func (s *MyL2Store) Close() error { ... }

然后直接传入：

cache, _ := mcachedb.NewMultiCache(&MyDBStore{}, &MyL2Store{}, &mcachedb.MultiCacheConfig{})

L2 Redis 配置

type RedisConfig struct {
    Addr         string        // 单机地址，默认 localhost:6379
    Addrs        []string      // 集群节点列表，len > 1 时自动切换集群模式
    Password     string
    DB           int
    KeyPrefix    string        // Key 前缀，默认 "mcachedb:"
    DefaultTTL   time.Duration // 0 表示不过期
    PoolSize     int
    MinIdleConns int
    ClusterMode  bool          // 显式开启集群模式
}

集群模式下 MGet/MSet/MDelete 由 go-redis 自动按 slot 分片；Subscribe 仅在单机模式下可用。

L2 故障降级与自动恢复
  • L2 出现网络错误或超时后，自动标记 l2Down，读写跳过 L2 直通 L3。
  • 后台每 30s 探测一次，连续 2 次 Ping 成功后恢复 L2。
  • 默认 WriteToL2OnSet: false，L2 故障不会影响写成功率。

辅助构建器

SimpleConfig
cache, err := mcachedb.NewSimpleCache(&Player{}, &mcachedb.SimpleConfig{
    DB:          db,
    RedisAddr:   "127.0.0.1:6379",
    RedisPrefix: "player:",
    L1Size:      50_000,
})
Builder 链式
cache, err := mcachedb.NewCacheBuilder(db, &Player{}).
    WithRedis("127.0.0.1:6379", "", 0).
    WithRedisPrefix("player:").
    WithL1Size(50_000).
    WithFlushInterval(500 * time.Millisecond).
    Build()
预设高可用模式
// 写 L1 时同步写 L2，适合读多写少的共享缓存
cache, err := mcachedb.NewHighReliabilityCache(db, "127.0.0.1:6379", &Player{})

// 同上，且 flush 间隔缩短到 100ms，适合交易/充值等强一致场景
cache, err := mcachedb.NewUltraReliabilityCache(db, "127.0.0.1:6379", &Player{})

事务

单缓存事务 Tx

对同一个 Cache 的多个 key 原子提交，失败时逆序恢复内存旧值。

tx, err := cache.Begin()
if err != nil {
    panic(err)
}

tx.Set(playerA)
tx.Delete("old-key")

if err := tx.Commit(); err != nil {
    _ = tx.Rollback()
}
跨缓存分布式事务 DistTx

协调多个 MultiCache 实例，采用"预读旧值 + 顺序提交 + 失败补偿"策略。

tx := mcachedb.NewDistTx()
tx.AddSet(goldCache, newGold)
tx.AddSet(bagCache, newItem)

if err := tx.Commit(); err != nil {
    // 已执行的步骤会尽最大努力回滚
    log.Println(err)
}

// 提交后立即把涉及缓存的脏数据 flush 到 L3（适合交易场景）
if err := tx.CommitAndFlush(); err != nil {
    log.Println(err)
}

DistTx 不提供跨进程的严格原子性（无两阶段提交），适用于同进程内多缓存的最大努力一致性。

泛型实体

不想手写完整 Entity 实现时，可使用内置泛型包装：

type PlayerData struct {
    Name  string `json:"name"`
    Level int    `json:"level"`
}

entity := mcachedb.NewGenericEntity("u1", &PlayerData{Name: "Alice", Level: 10})
cache.Set(entity)

GenericEntity[T] 内置 JSON 序列化；若需要 Protobuf 等其他格式，需要自行子类化并覆盖 Marshal/Unmarshal。

SQLXStore 通用存储

内置通用数据库存储，自动建表（MySQL），表结构：

列 类型 说明
cache_key VARCHAR(255) PK Entity.CacheKey()
cache_data JSON 实体序列化数据
version BIGINT 乐观锁版本号
is_deleted TINYINT 软删除标记
updated_at TIMESTAMP 自动更新时间

删除为软删除，可调用 CleanExpired 定期物理清理：

store.CleanExpired(ctx, time.Now().Add(-24*time.Hour))

如果业务已有数据库表，直接实现 DBStore 接口即可，不必使用 SQLXStore。

统计与监控

stats := cache.Stats()
fmt.Printf("L1=%d L2=%d L3=%d Miss=%d HitRate=%.2f%%\n",
    stats.L1Hits, stats.L2Hits, stats.L3Hits, stats.Misses,
    stats.HitRate()*100,
)

MultiCacheStats 提供：

字段 说明
L1Hits L1 命中次数
L2Hits L2 命中次数
L3Hits L3 命中次数
Misses 三级均未命中次数
HitRate() 总命中率
对接 Prometheus 示例
import "github.com/prometheus/client_golang/prometheus"

var (
    cacheHit = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "mcachedb_hit_total",
        Help: "mcachedb hit counter",
    }, []string{"cache", "level"})

    cacheMiss = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "mcachedb_miss_total",
        Help: "mcachedb miss counter",
    }, []string{"cache"})
)

func recordStats(name string, c *mcachedb.MultiCache) {
    s := c.Stats()
    cacheHit.WithLabelValues(name, "l1").Add(float64(s.L1Hits))
    cacheHit.WithLabelValues(name, "l2").Add(float64(s.L2Hits))
    cacheHit.WithLabelValues(name, "l3").Add(float64(s.L3Hits))
    cacheMiss.WithLabelValues(name).Add(float64(s.Misses))
}

注意：Stats() 返回的是累计值，建议按固定间隔采样后做差值或 Counter 直接累加。

性能参考

以下数据为本地开发机（8C16G / 本机 Redis / 本机 MySQL）的粗略参考，实际性能强烈建议结合业务实体大小、网络延迟、数据库负载自行压测。

指标 L1 L2 Redis L3 MySQL
单次读取延迟 ~300 ns ~ 1 μs ~0.5 ms ~ 2 ms ~5 ms ~ 20 ms
单次写入延迟 ~300 ns ~ 1 μs ~0.5 ms ~ 2 ms ~5 ms ~ 20 ms
不同写入模式吞吐对比
模式 大致吞吐（ops/s） 说明
WriteModeAsync 5w ~ 15w 写 L1 即返回，批量 flush
WriteModeSync 2k ~ 8k 每次写同步落库
WriteModeWriteThrough 2k ~ 8k 先写库再写缓存
WriteModeCacheAside 1k ~ 5k 先写库再删缓存，读时回填
WriteModeAsync + WriteToL2OnSet 3w ~ 8w 同步写 L2，性能有所下降
命中率建议
  • L1 命中率建议在 70% ~ 90%。
  • L2 命中率建议在 20% ~ 50%。
  • L3 + Miss 控制在 10% 以下。

如果 L1 命中率偏低，可适当调大 L1MaxSize 或缩短 CleanupInterval。

接口总览

// 实体
type Entity interface {
    CacheKey() string
    IsDeleted() bool
    SetDeleted(bool)
    Version() int64
    IncrementVersion()
    Copy() Entity
    Marshal() ([]byte, error)
    Unmarshal([]byte) error
}

// L3 数据库存储
type DBStore interface {
    Get(ctx context.Context, key string) (Entity, error)
    MGet(ctx context.Context, keys []string) (map[string]Entity, error)
    Insert(ctx context.Context, entity Entity) error
    Update(ctx context.Context, entity Entity) error
    Delete(ctx context.Context, key string) error
    BatchInsert(ctx context.Context, entities []Entity) error
    BatchUpdate(ctx context.Context, entities []Entity) error
    BatchDelete(ctx context.Context, keys []string) error
    Close() error
}

// L2 缓存存储（RedisStore 实现此接口）
type L2Store interface {
    Get(ctx context.Context, key string) (Entity, error)
    MGet(ctx context.Context, keys []string) (map[string]Entity, error)
    Set(ctx context.Context, entity Entity) error
    MSet(ctx context.Context, entities []Entity) error
    Delete(ctx context.Context, key string) error
    Ping(ctx context.Context) error
    Close() error
}

选型建议

场景 推荐模式 / 函数
读多写少，允许秒级数据丢失 WriteModeAsync（默认）
配置/字典类强一致读 WriteModeCacheAside
交易、充值、库存扣减 WriteModeCacheAside + DistTx.CommitAndFlush()，或 NewUltraReliabilityCache
无 Redis，单机或单元测试 NewMultiCache(l3, nil, config)
跨进程共享热数据 NewHighReliabilityCache（WriteToL2OnSet: true）
对接已有数据库/缓存 自定义 DBStore / L2Store

License

MIT

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func BuyItemExample 

func BuyItemExample(goldCache, bagCache *MultiCache, uid int64, itemID int64, cost int64) error

BuyItemExample 用 DistTx 实现扣金币 + 加背包道具 goldCache 和 bagCache 是两个独立的 MultiCache

Types

type BagDBStore 

type BagDBStore struct {
	// contains filtered or unexported fields
}

func NewBagDBStore 

func NewBagDBStore(db *sqlx.DB) *BagDBStore

func (*BagDBStore) BatchDelete 

func (s *BagDBStore) BatchDelete(ctx context.Context, keys []string) error

func (*BagDBStore) BatchInsert 

func (s *BagDBStore) BatchInsert(ctx context.Context, entities []Entity) error

func (*BagDBStore) BatchUpdate 

func (s *BagDBStore) BatchUpdate(ctx context.Context, entities []Entity) error

func (*BagDBStore) Close 

func (s *BagDBStore) Close() error

func (*BagDBStore) Delete 

func (s *BagDBStore) Delete(ctx context.Context, key string) error

func (*BagDBStore) Get 

func (s *BagDBStore) Get(ctx context.Context, key string) (Entity, error)

func (*BagDBStore) Insert 

func (s *BagDBStore) Insert(ctx context.Context, entity Entity) error

func (*BagDBStore) MGet 

func (s *BagDBStore) MGet(ctx context.Context, keys []string) (map[string]Entity, error)

func (*BagDBStore) Update 

func (s *BagDBStore) Update(ctx context.Context, entity Entity) error

type BagItemEntity 

type BagItemEntity struct {
	BaseEntity
	UID    int64 `json:"uid"    db:"uid"`
	ItemID int64 `json:"item_id" db:"item_id"`
	Count  int64 `json:"count"  db:"count"`
}

func NewBagItemEntity 

func NewBagItemEntity(uid, itemID, count int64) *BagItemEntity

func (*BagItemEntity) Copy 

func (e *BagItemEntity) Copy() Entity

func (*BagItemEntity) Marshal added in v1.0.17 

func (e *BagItemEntity) Marshal() ([]byte, error)

func (*BagItemEntity) Unmarshal added in v1.0.17 

func (e *BagItemEntity) Unmarshal(data []byte) error

type BagService 

type BagService struct {
	// contains filtered or unexported fields
}

func NewBagService 

func NewBagService(cache *MultiCache) *BagService

func (*BagService) ChangeCount 

func (s *BagService) ChangeCount(uid, itemID, delta int64) error

ChangeCount 只修改单个道具，数据库仅更新一行

func (*BagService) DeleteItem 

func (s *BagService) DeleteItem(uid, itemID int64) error

DeleteItem 删除单个道具

func (*BagService) GetAll 

func (s *BagService) GetAll(uid int64, itemIDs []int64) (map[int64]*BagItemEntity, error)

GetAll 使用 MGet 批量读取整个背包 注意：实际项目中 itemIDs 可以从索引表或 Redis Set 中获取

func (*BagService) SetItem 

func (s *BagService) SetItem(item *BagItemEntity) error

SetItem 直接设置单个道具

type BaseEntity 

type BaseEntity struct {
	ID         string    `json:"id" db:"id"`
	Ver        int64     `json:"ver" db:"ver"`
	DelFlag    bool      `json:"del_flag" db:"del_flag"`
	CreateTime time.Time `json:"create_time" db:"create_time"`
	UpdateTime time.Time `json:"update_time" db:"update_time"`
}

BaseEntity 基础实体实现

func NewBaseEntity 

func NewBaseEntity(id string) *BaseEntity

NewBaseEntity 创建基础实体

func (*BaseEntity) CacheKey 

func (e *BaseEntity) CacheKey() string

CacheKey 返回缓存键

func (*BaseEntity) Copy 

func (e *BaseEntity) Copy() *BaseEntity

Copy 复制（深拷贝）

func (*BaseEntity) IncrementVersion 

func (e *BaseEntity) IncrementVersion()

IncrementVersion 增加版本号

func (*BaseEntity) IsDeleted 

func (e *BaseEntity) IsDeleted() bool

IsDeleted 是否已删除

func (*BaseEntity) SetDeleted 

func (e *BaseEntity) SetDeleted(deleted bool)

SetDeleted 设置删除标记

func (*BaseEntity) Version 

func (e *BaseEntity) Version() int64

Version 获取版本号

type Cache 

type Cache struct {
	// contains filtered or unexported fields
}

Cache 带数据库持久化的缓存

func New 

func New(store DBStore, opts ...Option) (*Cache, error)

New 创建缓存

func NewUserCache 

func NewUserCache(store DBStore, opts ...Option) (*Cache, error)

CacheWithUser 带有用户实体类型的缓存

func (*Cache) Begin 

func (c *Cache) Begin() (*Tx, error)

Begin 开始事务

func (*Cache) Clear 

func (c *Cache) Clear()

Clear 清空缓存

func (*Cache) Close 

func (c *Cache) Close() error

Close 关闭缓存

func (*Cache) Delete 

func (c *Cache) Delete(key string) error

Delete 删除

func (*Cache) DirtyCount 

func (c *Cache) DirtyCount() int

DirtyCount 获取脏数据数量

func (*Cache) Flush 

func (c *Cache) Flush() error

Flush 立即刷新到数据库

func (*Cache) Get 

func (c *Cache) Get(key string) (Entity, error)

Get 获取实体 1. 先查缓存 2. 缓存未命中查数据库 3. 回填缓存

func (*Cache) IsDeleted added in v1.0.15 

func (c *Cache) IsDeleted(key string) bool

IsDeleted 返回 key 在 L1 中是否存在且已被标记删除

func (*Cache) IsDirty added in v1.0.15 

func (c *Cache) IsDirty(key string) bool

IsDirty 返回 key 在 L1 中是否存在且未 flush 的脏数据

func (*Cache) Keys 

func (c *Cache) Keys() []string

Keys 获取所有key

func (*Cache) L2DirtyCount added in v1.0.18 

func (c *Cache) L2DirtyCount() int

L2DirtyCount 获取待同步到 L2 的条目数

func (*Cache) Len 

func (c *Cache) Len() int

Len 获取缓存大小

func (*Cache) Load added in v1.0.15 

func (c *Cache) Load(entity Entity)

Load 将实体加载到 L1 内存，不标记 dirty，不触发后台刷盘 适用于从 L2/L3 回填等只读场景 跳过条件：L1 中已有脏数据/删除标记，或 L1 版本号 >= 传入版本（避免旧数据覆盖新数据）

func (*Cache) MGet 

func (c *Cache) MGet(keys []string) (map[string]Entity, error)

MGet 批量获取

func (*Cache) Pipeline 

func (c *Cache) Pipeline() *Pipeline

Pipeline 创建管道

func (*Cache) Remove added in v1.0.15 

func (c *Cache) Remove(key string)

Remove 直接从内存中移除指定 key，不触发持久化，也不标记脏数据

func (*Cache) Set 

func (c *Cache) Set(entity Entity) error

Set 设置实体

func (*Cache) Stats 

func (c *Cache) Stats() Stats

Stats 获取统计信息

type CacheBuilder 

type CacheBuilder struct {
	// contains filtered or unexported fields
}

CacheBuilder 缓存构建器

func NewCacheBuilder 

func NewCacheBuilder(db *sqlx.DB, entityType Entity) *CacheBuilder

NewCacheBuilder 创建构建器

func (*CacheBuilder) Build 

func (b *CacheBuilder) Build() (*MultiCache, error)

Build 构建缓存

func (*CacheBuilder) MustBuild 

func (b *CacheBuilder) MustBuild() *MultiCache

MustBuild 构建缓存，失败则panic

func (*CacheBuilder) WithFlushInterval 

func (b *CacheBuilder) WithFlushInterval(d time.Duration) *CacheBuilder

WithFlushInterval 设置刷盘间隔

func (*CacheBuilder) WithL1Size 

func (b *CacheBuilder) WithL1Size(size int) *CacheBuilder

WithL1Size 设置L1大小

func (*CacheBuilder) WithRedis 

func (b *CacheBuilder) WithRedis(addr, password string, db int) *CacheBuilder

WithRedis 设置Redis

func (*CacheBuilder) WithRedisPrefix 

func (b *CacheBuilder) WithRedisPrefix(prefix string) *CacheBuilder

WithRedisPrefix 设置Redis前缀

func (*CacheBuilder) WithSyncInterval 

func (b *CacheBuilder) WithSyncInterval(d time.Duration) *CacheBuilder

WithSyncInterval 设置同步间隔

type Config 

type Config struct {
	// 写入模式
	WriteMode WriteMode

	// 刷新模式
	FlushMode FlushMode

	// 刷新间隔（定时刷新模式下有效）
	FlushInterval time.Duration

	// 批量大小，达到此数量触发写入
	BatchSize int

	// 最大缓存条目数
	MaxCacheSize int

	// 默认过期时间，0 表示永不过期
	DefaultExpiration time.Duration

	// 清理过期数据间隔
	CleanupInterval time.Duration

	// 写入失败重试次数
	RetryCount int

	// 重试间隔
	RetryInterval time.Duration

	// 并发写入协程数
	WriteWorkers int

	// 数据库连接池配置
	DBMaxOpenConns int
	DBMaxIdleConns int
	DBMaxLifetime  time.Duration

	// 回调函数
	OnFlushStart   func(dirtyCount int)
	OnFlushSuccess func(count int, duration time.Duration)
	OnFlushError   func(err error, entities []Entity)
	OnCacheMiss    func(key string)
	OnCacheHit     func(key string)
}

Config 配置

func DefaultConfig 

func DefaultConfig() *Config

DefaultConfig 返回默认配置

type DBStore 

type DBStore interface {
	// Get 从数据库获取实体
	Get(ctx context.Context, key string) (Entity, error)

	// MGet 批量获取
	MGet(ctx context.Context, keys []string) (map[string]Entity, error)

	// Insert 插入新记录
	Insert(ctx context.Context, entity Entity) error

	// Update 更新记录
	Update(ctx context.Context, entity Entity) error

	// Delete 删除记录
	Delete(ctx context.Context, key string) error

	// BatchInsert 批量插入
	BatchInsert(ctx context.Context, entities []Entity) error

	// BatchUpdate 批量更新
	BatchUpdate(ctx context.Context, entities []Entity) error

	// BatchDelete 批量删除
	BatchDelete(ctx context.Context, keys []string) error

	// Close 关闭连接
	Close() error
}

DBStore 数据库存储接口

type DistTx 

type DistTx struct {
	// contains filtered or unexported fields
}

DistTx 跨 MultiCache 的分布式事务协调器 采用"预读旧值 + 顺序提交 + 失败补偿"策略

func NewDistTx 

func NewDistTx() *DistTx

NewDistTx 创建跨缓存事务

func (*DistTx) AddDelete 

func (dt *DistTx) AddDelete(cache *MultiCache, key string) error

AddDelete 注册 Delete 操作，同时预读旧值用于回滚

func (*DistTx) AddSet 

func (dt *DistTx) AddSet(cache *MultiCache, entity Entity) error

AddSet 注册 Set 操作，同时预读旧值用于回滚

func (*DistTx) Commit 

func (dt *DistTx) Commit() error

Commit 提交事务：顺序执行，失败时自动尽最大努力回滚

func (*DistTx) CommitAndFlush 

func (dt *DistTx) CommitAndFlush() error

CommitAndFlush 提交事务，并立即把所有涉及缓存的脏数据刷到数据库 适用于充值、交易等必须立即落盘的场景

func (*DistTx) Rollback 

func (dt *DistTx) Rollback() error

Rollback 手动回滚（只能在 Commit 前调用）

type DistTxOpType 

type DistTxOpType string

DistTxOpType 操作类型 

const (
	DistTxSet    DistTxOpType = "set"
	DistTxDelete DistTxOpType = "delete"
)

type Entity 

type Entity interface {
	// CacheKey 返回缓存键
	CacheKey() string
	// IsDeleted 是否标记删除
	IsDeleted() bool
	// SetDeleted 标记删除
	SetDeleted(deleted bool)
	// Version 获取版本号（用于乐观锁）
	Version() int64
	// IncrementVersion 增加版本号
	IncrementVersion()
	// Copy 复制实体（深拷贝）
	Copy() Entity

	// Marshal 将实体序列化为字节流，由用户决定具体格式（JSON、Protobuf 等）
	Marshal() ([]byte, error)
	// Unmarshal 从字节流反序列化到当前实体，格式需与 Marshal 保持一致
	Unmarshal([]byte) error
}

Entity 实体接口，定义了缓存实体的基本契约

type EntityWrapper 

type EntityWrapper struct {
	BaseEntity
	Data interface{} `json:"data"`
}

EntityWrapper 实体包装器，用于将任意结构体转为 Entity

func NewEntityWrapper 

func NewEntityWrapper(key string, data interface{}) *EntityWrapper

NewEntityWrapper 创建实体包装器

func (*EntityWrapper) Copy 

func (e *EntityWrapper) Copy() Entity

Copy 复制

func (*EntityWrapper) Marshal added in v1.0.16 

func (e *EntityWrapper) Marshal() ([]byte, error)

Marshal JSON 序列化整个 wrapper

func (*EntityWrapper) Unmarshal added in v1.0.16 

func (e *EntityWrapper) Unmarshal(data []byte) error

Unmarshal JSON 反序列化整个 wrapper

type FlushMode 

type FlushMode int

FlushMode 刷新模式 

const (
	// FlushModeInterval 定时刷新（默认）
	FlushModeInterval FlushMode = iota
	// FlushModeImmediate 立即刷新（每次操作都触发检查）
	FlushModeImmediate
	// FlushModeManual 手动刷新（需要手动调用 Flush）
	FlushModeManual
)

type GenericEntity 

type GenericEntity[T any] struct {
	BaseEntity
	Payload T `json:"payload"`
}

GenericEntity 泛型实体

func NewGenericEntity 

func NewGenericEntity[T any](key string, payload T) *GenericEntity[T]

NewGenericEntity 创建泛型实体

func (*GenericEntity[T]) Copy 

func (e *GenericEntity[T]) Copy() Entity

Copy 复制泛型实体

func (*GenericEntity[T]) Marshal added in v1.0.16 

func (e *GenericEntity[T]) Marshal() ([]byte, error)

Marshal JSON 序列化完整泛型实体

func (*GenericEntity[T]) Unmarshal added in v1.0.16 

func (e *GenericEntity[T]) Unmarshal(data []byte) error

Unmarshal JSON 反序列化完整泛型实体

type GoldEntity 

type GoldEntity struct {
	BaseEntity
	UID  int64 `json:"uid"`
	Gold int64 `json:"gold"`
}

GoldEntity 金币实体（简化示例）

func NewGoldEntity 

func NewGoldEntity(uid, gold int64) *GoldEntity

func (*GoldEntity) Copy 

func (e *GoldEntity) Copy() Entity

func (*GoldEntity) Marshal added in v1.0.17 

func (e *GoldEntity) Marshal() ([]byte, error)

func (*GoldEntity) Unmarshal added in v1.0.17 

func (e *GoldEntity) Unmarshal(data []byte) error

type L2Store added in v1.0.15 

type L2Store interface {
	Get(ctx context.Context, key string) (Entity, error)
	MGet(ctx context.Context, keys []string) (map[string]Entity, error)
	Set(ctx context.Context, entity Entity) error
	MSet(ctx context.Context, entities []Entity) error
	Delete(ctx context.Context, key string) error
	Ping(ctx context.Context) error
	Close() error
}

L2Store L2 存储接口（内部使用，便于测试和扩展）

type MultiCache 

type MultiCache struct {
	// contains filtered or unexported fields
}

MultiCache 多级缓存 (L1内存 -> L2Redis -> L3数据库)

func MustNewSimpleCache 

func MustNewSimpleCache(entityType Entity, cfg *SimpleConfig) *MultiCache

MustNewSimpleCache 创建缓存，失败则panic

func NewHighReliabilityCache 

func NewHighReliabilityCache(db *sqlx.DB, redisAddr string, entityType Entity) (*MultiCache, error)

NewHighReliabilityCache 创建高可靠性缓存（写L1时同步写L2 Redis）

func NewMultiCache 

func NewMultiCache(dbStore DBStore, l2Store L2Store, config *MultiCacheConfig) (*MultiCache, error)

NewMultiCache 创建多级缓存

func NewSimpleCache 

func NewSimpleCache(entityType Entity, cfg *SimpleConfig) (*MultiCache, error)

NewSimpleCache 创建简化版多级缓存

func NewUltraReliabilityCache 

func NewUltraReliabilityCache(db *sqlx.DB, redisAddr string, entityType Entity) (*MultiCache, error)

NewUltraReliabilityCache 创建超高可靠性缓存（写L1同步写L2，且快速刷L3）

func (*MultiCache) Close 

func (mc *MultiCache) Close() error

Close 关闭

func (*MultiCache) Delete 

func (mc *MultiCache) Delete(key string) error

Delete 删除

func (*MultiCache) Flush 

func (mc *MultiCache) Flush() error

Flush 立即刷盘到数据库

func (*MultiCache) FlushToL3 added in v1.0.18 

func (mc *MultiCache) FlushToL3() error

FlushToL3 立即将 L1 中所有脏数据刷盘到 L3，不等待后台定时器

func (*MultiCache) Get 

func (mc *MultiCache) Get(key string) (Entity, error)

Get 获取（L1 -> L2 -> L3）

func (*MultiCache) MGet 

func (mc *MultiCache) MGet(keys []string) (map[string]Entity, error)

MGet 批量获取（L1 -> L2 -> L3），汇总返回

func (*MultiCache) Set 

func (mc *MultiCache) Set(entity Entity) error

Set 设置

func (*MultiCache) Stats 

func (mc *MultiCache) Stats() MultiCacheStats

Stats 获取统计

func (*MultiCache) SyncToL2 added in v1.0.18 

func (mc *MultiCache) SyncToL2()

SyncToL2 立即将 L1 中待同步的变更推送到 L2，不等待后台定时器

type MultiCacheConfig 

type MultiCacheConfig struct {
	L1MaxSize         int
	L1CleanupInterval time.Duration
	WriteToL2OnSet    bool
	SyncInterval      time.Duration
	FlushInterval     time.Duration
	WriteMode         WriteMode // L1 写入模式
}

MultiCacheConfig 多级缓存配置

func DefaultMultiCacheConfig 

func DefaultMultiCacheConfig() *MultiCacheConfig

DefaultMultiCacheConfig 默认配置

type MultiCacheStats 

type MultiCacheStats struct {
	// 命中计数
	L1Hits int64
	L2Hits int64
	L3Hits int64
	Misses int64

	// 当前内存状态
	L1Size       int   // L1 中当前存活的条目数（含未 flush 的脏数据）
	L1DirtyCount int64 // 待 flush 到 L3 的脏条目数
	L2DirtyCount int64 // 待同步到 L2 的条目数（自上次 syncToL2 以来新增的变更）

	// L2 状态
	L2Down bool // L2 当前是否处于降级状态
}

MultiCacheStats 统计

func (*MultiCacheStats) HitRate 

func (s *MultiCacheStats) HitRate() float64

HitRate 总命中率

func (*MultiCacheStats) L1HitRate added in v1.0.18 

func (s *MultiCacheStats) L1HitRate() float64

L1HitRate L1 命中率

type Option 

type Option func(*Config)

Option 配置选项

func WithBatchSize 

func WithBatchSize(size int) Option

WithBatchSize 设置批量大小

func WithCleanupInterval 

func WithCleanupInterval(d time.Duration) Option

WithCleanupInterval 设置清理过期数据间隔

func WithDefaultExpiration 

func WithDefaultExpiration(d time.Duration) Option

WithDefaultExpiration 设置默认过期时间

func WithFlushInterval 

func WithFlushInterval(d time.Duration) Option

WithFlushInterval 设置刷新间隔

func WithFlushMode 

func WithFlushMode(mode FlushMode) Option

WithFlushMode 设置刷新模式

func WithMaxCacheSize 

func WithMaxCacheSize(size int) Option

WithMaxCacheSize 设置最大缓存大小

func WithRetryCount 

func WithRetryCount(count int) Option

WithRetryCount 设置重试次数

func WithWriteMode 

func WithWriteMode(mode WriteMode) Option

WithWriteMode 设置写入模式

func WithWriteWorkers 

func WithWriteWorkers(n int) Option

WithWriteWorkers 设置写入工作协程数

type Order 

type Order struct {
	BaseEntity
	UserID    string      `json:"user_id"`
	ProductID string      `json:"product_id"`
	Amount    float64     `json:"amount"`
	Status    string      `json:"status"`
	Items     []OrderItem `json:"items"`
}

Order 订单实体示例

func NewOrder 

func NewOrder(id, userID, productID string, amount float64) *Order

NewOrder 创建订单

func (*Order) Copy 

func (o *Order) Copy() Entity

Copy 深拷贝

func (*Order) Marshal added in v1.0.16 

func (o *Order) Marshal() ([]byte, error)

Marshal JSON 序列化完整 Order 结构体

func (*Order) Unmarshal added in v1.0.16 

func (o *Order) Unmarshal(data []byte) error

Unmarshal JSON 反序列化到 Order

type OrderItem 

type OrderItem struct {
	ProductID string  `json:"product_id"`
	Quantity  int     `json:"quantity"`
	Price     float64 `json:"price"`
}

OrderItem 订单项

type Pipeline 

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline 管道批量操作

func (*Pipeline) Delete 

func (p *Pipeline) Delete(key string) *Pipeline

Delete 管道删除

func (*Pipeline) Discard 

func (p *Pipeline) Discard()

Discard 丢弃管道操作

func (*Pipeline) Exec 

func (p *Pipeline) Exec() error

Exec 执行管道

func (*Pipeline) Len 

func (p *Pipeline) Len() int

Len 返回管道长度

func (*Pipeline) Set 

func (p *Pipeline) Set(entity Entity) *Pipeline

Set 管道设置

type RedisConfig 

type RedisConfig struct {
	Addr         string
	Addrs        []string // 集群地址列表（如 ["127.0.0.1:7000", "127.0.0.1:7001"]）
	Password     string
	DB           int
	KeyPrefix    string
	DefaultTTL   time.Duration
	PoolSize     int
	MinIdleConns int
	ClusterMode  bool // 显式开启集群模式
}

RedisConfig Redis配置

func DefaultRedisConfig 

func DefaultRedisConfig() *RedisConfig

DefaultRedisConfig 默认Redis配置

type RedisStore 

type RedisStore struct {
	// contains filtered or unexported fields
}

RedisStore Redis存储实现（支持单机和集群模式）

func NewRedisStore 

func NewRedisStore(config *RedisConfig, entityType ...Entity) (*RedisStore, error)

NewRedisStore 创建Redis存储（自动识别单机/集群模式）

func (*RedisStore) Close 

func (s *RedisStore) Close() error

Close 关闭连接

func (*RedisStore) Delete 

func (s *RedisStore) Delete(ctx context.Context, key string) error

Delete 从Redis删除

func (*RedisStore) Get 

func (s *RedisStore) Get(ctx context.Context, key string) (Entity, error)

Get 从Redis获取

func (*RedisStore) MDelete 

func (s *RedisStore) MDelete(ctx context.Context, keys []string) error

MDelete 批量删除（集群模式下 go-redis 会自动按 slot 分片执行）

func (*RedisStore) MGet 

func (s *RedisStore) MGet(ctx context.Context, keys []string) (map[string]Entity, error)

MGet 批量获取（集群模式下 go-redis 会自动按 slot 分片执行）

func (*RedisStore) MSet 

func (s *RedisStore) MSet(ctx context.Context, entities []Entity) error

MSet 批量设置（集群模式下 pipeline 会自动按 slot 分片到各节点执行）

func (*RedisStore) Ping added in v1.0.15 

func (s *RedisStore) Ping(ctx context.Context) error

Ping 探测 Redis 是否可用

func (*RedisStore) Publish 

func (s *RedisStore) Publish(ctx context.Context, channel string, message string) error

Publish 发布消息

func (*RedisStore) Set 

func (s *RedisStore) Set(ctx context.Context, entity Entity) error

Set 设置到Redis

func (*RedisStore) Subscribe 

func (s *RedisStore) Subscribe(ctx context.Context, channels ...string) *redis.PubSub

Subscribe 订阅消息（集群模式下暂不支持，返回 nil）

type SQLXStore 

type SQLXStore struct {
	// contains filtered or unexported fields
}

SQLXStore SQLX存储实现

func NewSQLXStore 

func NewSQLXStore(dsn string, entityType Entity, configs ...*SQLXStoreConfig) (*SQLXStore, error)

NewSQLXStore 创建SQLX存储

func NewSQLXStoreFromDB 

func NewSQLXStoreFromDB(db *sqlx.DB, entityType Entity) (*SQLXStore, error)

NewSQLXStoreFromDB 从现有 sqlx.DB 创建存储

func (*SQLXStore) BatchDelete 

func (s *SQLXStore) BatchDelete(ctx context.Context, keys []string) error

func (*SQLXStore) BatchInsert 

func (s *SQLXStore) BatchInsert(ctx context.Context, entities []Entity) error

func (*SQLXStore) BatchUpdate 

func (s *SQLXStore) BatchUpdate(ctx context.Context, entities []Entity) error

func (*SQLXStore) CleanExpired 

func (s *SQLXStore) CleanExpired(ctx context.Context, before time.Time) error

CleanExpired 清理过期（软删除）数据

func (*SQLXStore) Close 

func (s *SQLXStore) Close() error

func (*SQLXStore) Delete 

func (s *SQLXStore) Delete(ctx context.Context, key string) error

func (*SQLXStore) Get 

func (s *SQLXStore) Get(ctx context.Context, key string) (Entity, error)

func (*SQLXStore) Insert 

func (s *SQLXStore) Insert(ctx context.Context, entity Entity) error

func (*SQLXStore) MGet 

func (s *SQLXStore) MGet(ctx context.Context, keys []string) (map[string]Entity, error)

func (*SQLXStore) Update 

func (s *SQLXStore) Update(ctx context.Context, entity Entity) error

type SQLXStoreConfig 

type SQLXStoreConfig struct {
	DSN          string
	TableName    string
	KeyColumn    string
	DataColumn   string
	VerColumn    string
	DelColumn    string
	TimeColumn   string
	MaxOpenConns int
	MaxIdleConns int
	MaxLifetime  time.Duration
}

SQLXStoreConfig SQLX存储配置

func DefaultSQLXStoreConfig 

func DefaultSQLXStoreConfig() *SQLXStoreConfig

DefaultSQLXStoreConfig 默认配置

type ShardFunc 

type ShardFunc func(key string) int

ShardFunc 分片函数：根据 key 返回分片索引

func DefaultShardFunc 

func DefaultShardFunc(shardCount int) ShardFunc

DefaultShardFunc 默认 CRC32 取模分片

type ShardedCache 

type ShardedCache struct {
	// contains filtered or unexported fields
}

ShardedCache 分片多级缓存（L1/L2/L3 按 shard 隔离）

func NewShardedCache 

func NewShardedCache(shardCount int, newShard func(idx int) (*MultiCache, error), shardFn ShardFunc) (*ShardedCache, error)

NewShardedCache 创建分片缓存池 

shardCount: 分片数量（建议与 MySQL 分表数一致，如 16/32/64/128）
newShard:   工厂函数，传入分片索引 idx，返回该分片对应的 *MultiCache
shardFn:    可选自定义分片函数，nil 则使用 CRC32 取模

func (*ShardedCache) Close 

func (sc *ShardedCache) Close() error

Close 关闭所有分片

func (*ShardedCache) Delete 

func (sc *ShardedCache) Delete(key string) error

Delete 删除

func (*ShardedCache) Flush 

func (sc *ShardedCache) Flush() error

Flush 强制所有分片立即刷盘

func (*ShardedCache) Get 

func (sc *ShardedCache) Get(key string) (Entity, error)

Get 获取单个实体

func (*ShardedCache) MGet 

func (sc *ShardedCache) MGet(keys []string) (map[string]Entity, error)

MGet 批量获取（按分片并发查询后聚合结果）

func (*ShardedCache) Set 

func (sc *ShardedCache) Set(entity Entity) error

Set 写入（自动路由到对应分片）

func (*ShardedCache) Stats 

func (sc *ShardedCache) Stats() MultiCacheStats

Stats 聚合所有分片的命中统计

type SimpleConfig 

type SimpleConfig struct {
	// 数据库
	DB *sqlx.DB

	// Redis（可选，为空则不使用L2）
	RedisAddr   string
	RedisPass   string
	RedisDB     int
	RedisPrefix string

	// 性能配置
	L1Size        int           // L1最大条目数，默认10万
	FlushInterval time.Duration // 刷盘间隔，默认1秒
	SyncInterval  time.Duration // L1->L2同步间隔，默认500ms
}

SimpleConfig 简化配置

type Stats 

type Stats struct {
	// 缓存统计
	CacheHits   int64
	CacheMisses int64
	CacheSize   int
	DirtyCount  int64

	// 数据库统计
	DBReads       int64
	DBWrites      int64
	DBWriteErrors int64

	// 刷新统计
	FlushCount     int64
	FlushTotalTime time.Duration
	LastFlushTime  time.Time

	// 当前状态
	IsFlushing bool
}

Stats 统计信息

func (*Stats) HitRate 

func (s *Stats) HitRate() float64

HitRate 命中率

type Tx 

type Tx struct {
	// contains filtered or unexported fields
}

Tx 事务

func (*Tx) Commit 

func (tx *Tx) Commit() error

Commit 提交事务

func (*Tx) Delete 

func (tx *Tx) Delete(key string) error

Delete 事务中删除

func (*Tx) Get 

func (tx *Tx) Get(key string) (Entity, error)

Get 事务中获取

func (*Tx) Rollback 

func (tx *Tx) Rollback() error

Rollback 回滚事务

func (*Tx) Set 

func (tx *Tx) Set(entity Entity) error

Set 事务中设置

type User 

type User struct {
	BaseEntity
	Username string   `json:"username"`
	Email    string   `json:"email"`
	Age      int      `json:"age"`
	Tags     []string `json:"tags"`
}

User 用户实体示例

func NewUser 

func NewUser(id, username, email string, age int) *User

NewUser 创建用户

func (*User) Copy 

func (u *User) Copy() Entity

Copy 深拷贝

func (*User) Marshal added in v1.0.16 

func (u *User) Marshal() ([]byte, error)

Marshal JSON 序列化完整 User 结构体

func (*User) Unmarshal added in v1.0.16 

func (u *User) Unmarshal(data []byte) error

Unmarshal JSON 反序列化到 User

type UserService 

type UserService struct {
	// contains filtered or unexported fields
}

UserService 用户服务示例

func NewUserService 

func NewUserService(cache *Cache) *UserService

NewUserService 创建用户服务

func (*UserService) BatchUpdateUsers 

func (s *UserService) BatchUpdateUsers(users []*User) error

BatchUpdateUsers 批量更新用户

func (*UserService) DeleteUser 

func (s *UserService) DeleteUser(userID string) error

DeleteUser 删除用户

func (*UserService) GetUser 

func (s *UserService) GetUser(userID string) (*User, error)

GetUser 获取用户

func (*UserService) SaveUser 

func (s *UserService) SaveUser(user *User) error

SaveUser 保存用户

type WriteMode 

type WriteMode int

WriteMode 写入模式 

const (
	// WriteModeAsync 异步写入（默认），写入缓存后立即返回，后台批量刷盘
	WriteModeAsync WriteMode = iota
	// WriteModeSync 同步写入，写入缓存同时写入数据库
	WriteModeSync
	// WriteModeWriteThrough 直写模式，先写数据库成功后，再更新缓存
	WriteModeWriteThrough
	// WriteModeCacheAside 缓存旁路模式，先写数据库成功后，再删除缓存
	WriteModeCacheAside
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.