缓存实战
Redis 最广泛的应用场景就是缓存。本章从缓存更新策略入手,覆盖缓存三大经典问题(穿透、雪崩、击穿)、分布式锁的完整实现、秒杀系统实战、Feed 流、GEO/Bitmap/HyperLogLog 高级数据结构,以及多级缓存架构,力求每一步都有完整可运行的代码示例。
1. 缓存基础
1.1 缓存的作用
在典型的 Web 架构中,数据库通常是最慢的一环。缓存的引入带来三个核心收益:
| 收益 | 说明 |
|---|---|
| 降低后端负载 | 热点数据缓存在内存中,减少 DB 查询次数 |
| 提高读写效率 | Redis 单机 QPS 可达 10w+,远超 MySQL |
| 降低响应时间 | 内存读取在微秒级,数据库查询在毫秒级 |
成本与代价:
| 代价 | 说明 |
|---|---|
| 数据一致性 | 缓存中的数据可能与 DB 不一致,需要额外机制保证 |
| 代码维护 | 缓存逻辑侵入业务代码,增加复杂度 |
| 运维复杂度 | 缓存集群需要额外运维,宕机风险需要预案 |
1.2 缓存更新策略
| 策略 | 说明 | 一致性 | 维护成本 | 适用场景 |
|---|---|---|---|---|
| 内存淘汰 | Redis 自动淘汰(LRU/LFU),下次查询时更新 | 差 | 无 | 低一致性要求(如文章浏览计数) |
| 超时剔除 | 设置 TTL,到期自动删除 | 一般 | 低 | 一般业务要求的兜底方案 |
| 主动更新 | 修改数据库时同步更新/删除缓存 | 好 | 高 | 高一致性要求(如库存、余额) |
💡 最佳实践:主动更新为主,超时剔除为兜底。即业务代码负责更新/删除缓存,同时所有缓存都设置合理的 TTL 作为最后防线。
1.3 三种缓存模式
Cache Aside Pattern(旁路缓存)—— 最常用
应用直接与缓存和数据库交互,缓存层不自动同步。
读取流程:
客户端 → 查缓存 → 命中?→ 返回
↓ 未命中
查DB → 写缓存 → 返回写入流程:
客户端 → 更新DB → 删除缓存(而非更新缓存)func GetUser(ctx context.Context, id int64) (*User, error) {
key := fmt.Sprintf("user:%d", id)
// 1. 查缓存
data, err := rdb.Get(ctx, key).Result()
if err == nil {
var user User
json.Unmarshal([]byte(data), &user)
return &user, nil
}
// 2. 查数据库
user, err := db.QueryUser(ctx, id)
if err != nil {
return nil, err
}
// 3. 写缓存
userBytes, _ := json.Marshal(user)
rdb.Set(ctx, key, userBytes, 30*time.Minute)
return user, nil
}
func UpdateUser(ctx context.Context, user *User) error {
// 1. 更新数据库
if err := db.UpdateUser(ctx, user); err != nil {
return err
}
// 2. 删除缓存
key := fmt.Sprintf("user:%d", user.ID)
rdb.Del(ctx, key)
return nil
}Read/Write Through(读写穿透)
应用只与缓存层交互,缓存层负责与数据库同步。适合有专门缓存中间件的场景(如 Apache Ignite)。Go 中通常需要自己封装缓存抽象层。
// 伪代码:缓存层自动同步
type CacheService struct {
rdb *redis.Client
db *sql.DB
}
func (s *CacheService) Get(ctx context.Context, key string) (string, error) {
val, err := s.rdb.Get(ctx, key).Result()
if err == nil {
return val, nil
}
// 同步加载到缓存
val, err = s.db.Load(ctx, key)
if err != nil {
return "", err
}
s.rdb.Set(ctx, key, val, 0)
return val, nil
}Write Behind Caching(异步缓存写入)
应用只写缓存,缓存层异步批量写入数据库。适合超高频率写入、低一致性要求的场景(如浏览量、点赞数)。
写请求 → 只写 Redis(TTL 永久或很长)
↓ 异步(定时/批量)
MySQL// 示例:文章阅读数计数
func IncrViewCount(ctx context.Context, articleID int64) error {
key := fmt.Sprintf("article:view:%d", articleID)
return rdb.Incr(ctx, key).Err()
}
// 定时任务:每5分钟批量同步到MySQL
func SyncViewCounts(ctx context.Context) error {
keys, _ := rdb.Keys(ctx, "article:view:*").Result()
for _, key := range keys {
count, _ := rdb.GetSet(ctx, key, "0").Int64()
articleID := extractID(key)
db.UpdateViewCount(ctx, articleID, count)
}
return nil
}1.4 主动更新的三个关键问题
问题一:删除缓存还是更新缓存?
💡 选择删除缓存,而非更新缓存。
原因:更新缓存可能导致"写多读少"的无效写操作。如果缓存是一个复杂的计算结果(如关联查询、聚合统计数据),更新缓存成本更高。删除缓存后,下次读取自然会重建最新数据。
问题二:先操作缓存还是数据库?
💡 先操作数据库,后删除缓存。
两种顺序的并发问题分析:
方案 A:先删除缓存,后更新数据库
线程1:删除缓存
线程2:查缓存(miss) → 查DB(旧值) → 写缓存(旧值)
线程1:更新DB(新值)
→ 结果:缓存中是旧值,DB 是新值,不一致!方案 B:先更新数据库,后删除缓存
线程1:查缓存(miss) → 查DB(旧值)
线程2:更新DB(新值) → 删除缓存
线程1:写缓存(旧值)
→ 结果:缓存中是旧值,但这种窗口极短(需要线程1在写缓存前被阻塞很久)🔬 深入原理:方案 B 的并发窗口更小,因为数据库写入通常快于缓存的读写延迟,而且缓存 miss 后还需要网络 IO 查 DB。但理论上仍有极小概率不一致。解决方案:
- 设置 TTL:即使发生不一致,TTL 到期后也会自动修复
- 延迟双删:更新 DB 后立即删除缓存,延迟几百毫秒后再次删除缓存
// 延迟双删
func UpdateWithDelayDoubleDelete(ctx context.Context, user *User) error {
key := fmt.Sprintf("user:%d", user.ID)
// 第一删
rdb.Del(ctx, key)
// 更新数据库
if err := db.UpdateUser(ctx, user); err != nil {
return err
}
// 第二删(延迟执行)
time.AfterFunc(500*time.Millisecond, func() {
rdb.Del(context.Background(), key)
})
return nil
}问题三:如何保证一致性?
| 场景 | 方案 |
|---|---|
| 单体应用 | 将 DB 更新和缓存删除放在同一个本地事务中 |
| 分布式应用 | 使用分布式事务(TCC、Saga)、Canal 监听 binlog 异步更新缓存 |
1.5 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 先删缓存再更新 DB | 并发窗口期大,读请求会把旧数据写回缓存 |
| 🚨 修改缓存而非删除 | 写多读少的场景浪费写入、增加不一致风险 |
| 🚨 不设 TTL | 极端情况下脏数据永不过期 |
| 🚨 缓存与 DB 操作不在同一事务 | 可能一个成功一个失败 |
2. 缓存三大问题
2.1 缓存穿透
定义:请求的数据既不在缓存也不在数据库(通常为恶意攻击或查询不存在的 ID),所有请求直接打到数据库上。
请求不存在的数据(id=-1)
↓
查缓存: miss
↓
查DB: miss
↓
返回空 → 下一次请求依然走全流程解决方案
方案一:缓存空对象
对于不存在的数据,缓存一个 null 标记,并设置较短的 TTL。
func GetItem(ctx context.Context, id int64) (*Item, error) {
key := fmt.Sprintf("item:%d", id)
data, err := rdb.Get(ctx, key).Result()
if err == nil {
if data == "null" {
return nil, ErrNotFound
}
var item Item
json.Unmarshal([]byte(data), &item)
return &item, nil
}
item, err := db.QueryItem(ctx, id)
if errors.Is(err, sql.ErrNoRows) {
// 缓存空对象,短 TTL
rdb.Set(ctx, key, "null", 5*time.Minute)
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
itemBytes, _ := json.Marshal(item)
rdb.Set(ctx, key, itemBytes, 30*time.Minute)
return item, nil
}| 说明 | |
|---|---|
| 优点 | 实现简单,一行 Set 解决 |
| 缺点 | 占用额外内存;短期不一致(新建同名数据后可能短暂返回不存在) |
方案二:布隆过滤器
🔬 深入原理:布隆过滤器使用 k 个独立的哈希函数,将每个 key 映射到一个长度为 m 的位数组。查询时所有 k 个位置都为 1 才认为"可能存在";任一位置为 0 则"一定不存在"。误判率取决于 m、k 和元素数量 n。
添加 key "user_123":
hash1 → bit[3] = 1
hash2 → bit[8] = 1
hash3 → bit[15] = 1
查询 key "user_456":
hash1 → bit[3] = 1 ✓
hash2 → bit[8] = 1 ✓
hash3 → bit[15] = 0 ✗ → 一定不存在!使用 RedisBloom 模块:
# 创建布隆过滤器
BF.RESERVE user_filter 0.01 1000000
# 添加元素
BF.ADD user_filter user:1001
BF.MADD user_filter user:1002 user:1003
# 检查存在
BF.EXISTS user_filter user:1001 # 返回 1
BF.EXISTS user_filter user:9999 # 返回 0(一定不存在)Go 中使用 Redisson/go-redis 配合布隆过滤器(不依赖模块的纯 Redis 实现):
// 使用多个 hash 函数 + SETBIT/GETBIT 实现简易布隆过滤器
const (
bloomKey = "bloom:user"
hashCount = 3
)
func AddToBloom(ctx context.Context, value string) error {
pipe := rdb.Pipeline()
for i := 0; i < hashCount; i++ {
offset := hash(value, i) % (1 << 20) // 位数组大小 2^20
pipe.SetBit(ctx, bloomKey, int64(offset), 1)
}
_, err := pipe.Exec(ctx)
return err
}
func ExistInBloom(ctx context.Context, value string) bool {
pipe := rdb.Pipeline()
cmds := make([]*redis.IntCmd, hashCount)
for i := 0; i < hashCount; i++ {
offset := hash(value, i) % (1 << 20)
cmds[i] = pipe.GetBit(ctx, bloomKey, int64(offset))
}
pipe.Exec(ctx)
for _, cmd := range cmds {
if cmd.Val() == 0 {
return false
}
}
return true // 可能存在
}| 说明 | |
|---|---|
| 优点 | 内存占用极小(1 亿数据约 114MB),不存在漏判 |
| 缺点 | 有误判率(可接受)、实现复杂、删除元素困难 |
| 适用 | 数据量大、key 相对固定(如用户 ID、商品 ID) |
方案三:参数校验
在入口处对请求参数做基础校验,拦截明显非法的请求。
func GetItemHandler(w http.ResponseWriter, r *http.Request) {
id, err := strconv.ParseInt(r.URL.Query().Get("id"), 10, 64)
if err != nil || id <= 0 {
http.Error(w, "invalid id", http.StatusBadRequest)
return
}
// ... 继续处理
}方案四:权限校验 + 限流
在网关层对用户认证和鉴权,对异常高频的无效请求进行限流。
2.2 缓存雪崩
定义:大量 key 在同一时刻过期,或 Redis 服务宕机,导致所有请求直接打到数据库,造成数据库崩溃。
解决方案
方案一:TTL 加随机值
为每个 key 的过期时间加上一个随机偏移,避免集中过期。
const baseTTL = 30 * time.Minute
func SetWithJitter(ctx context.Context, key string, value interface{}) error {
data, _ := json.Marshal(value)
// 在基础 TTL 基础上加 ±5 分钟的随机偏移
ttl := baseTTL + time.Duration(rand.Int63n(600))*time.Second
return rdb.Set(ctx, key, data, ttl).Err()
}方案二:Redis 集群 + 哨兵
# sentinel.conf
sentinel monitor mymaster 192.168.1.10 6379 2
sentinel down-after-milliseconds mymaster 30000
sentinel parallel-syncs mymaster 1
sentinel failover-timeout mymaster 180000方案三:降级限流
使用 sentinel 限流组件或熔断器(如 Hystrix / Sentinel),在缓存层不可用时直接返回降级结果。
// 简单的本地限流器
var limiter = rate.NewLimiter(100, 200) // 100 QPS, burst 200
func GetItemWithLimiter(ctx context.Context, id int64) (*Item, error) {
if !limiter.Allow() {
return nil, errors.New("rate limited")
}
return GetItem(ctx, id)
}方案四:多级缓存
请求 → Nginx/OpenResty 本地缓存 (L1)
↓ miss
Redis 集群 (L2)
↓ miss
MySQL (L3)详见 第 10 节:多级缓存。
2.3 缓存击穿
定义:一个热点 key 在过期瞬间,大量并发请求同时打到数据库。
热点 key "hot:item:100" 过期
↓
1000 个并发请求同时 cache miss
↓
1000 个请求全部查询 DB
↓
DB 过载注意区分:缓存穿透是 key 根本不存在,缓存击穿是 热点 key 恰好过期。
解决方案一:互斥锁
核心思想:只让一个线程去查 DB 重建缓存,其他线程等待结果。
请求1: 获取锁成功 → 查DB → 写缓存 → 释放锁 → 返回
请求2: 获取锁失败 → 等待/重试 → 从缓存读取 → 返回
请求3: 获取锁失败 → 等待/重试 → 从缓存读取 → 返回func GetHotItem(ctx context.Context, id int64) (*Item, error) {
cacheKey := fmt.Sprintf("item:%d", id)
lockKey := fmt.Sprintf("lock:item:%d", id)
// 1. 查缓存
data, err := rdb.Get(ctx, cacheKey).Result()
if err == nil {
var item Item
json.Unmarshal([]byte(data), &item)
return &item, nil
}
// 2. 尝试获取互斥锁
locked, err := rdb.SetNX(ctx, lockKey, "1", 10*time.Second).Result()
if err != nil {
return nil, err
}
if locked {
// 3. 获取锁成功:Double Check + 查 DB + 写缓存 + 释放锁
defer rdb.Del(ctx, lockKey)
// Double Check
data, err = rdb.Get(ctx, cacheKey).Result()
if err == nil {
var item Item
json.Unmarshal([]byte(data), &item)
return &item, nil
}
// 查数据库
item, err := db.QueryItem(ctx, id)
if err != nil {
return nil, err
}
// 写缓存
itemBytes, _ := json.Marshal(item)
rdb.Set(ctx, cacheKey, itemBytes, 30*time.Minute)
return item, nil
}
// 4. 获取锁失败:等待并重试(最多重试 10 次)
for i := 0; i < 10; i++ {
time.Sleep(50 * time.Millisecond)
data, err = rdb.Get(ctx, cacheKey).Result()
if err == nil {
var item Item
json.Unmarshal([]byte(data), &item)
return &item, nil
}
}
return nil, errors.New("failed to get item after retries")
}| 说明 | |
|---|---|
| 优点 | 一致性好,实现简单 |
| 缺点 | 线程等待增加响应延迟;有死锁风险(必须设锁超时) |
解决方案二:逻辑过期
核心思想:缓存永不过期(或设置很长 TTL),value 中包含一个逻辑过期时间字段。读取时判断是否逻辑过期:
- 未过期 → 直接返回
- 已过期 → 获取锁 → 启动新 goroutine 异步重建 → 当前请求返回旧数据
请求1: 发现逻辑过期 → 获得锁 → 开启异步线程重建 → 立即返回旧数据
请求2: 发现逻辑过期 → 未获得锁 → 立即返回旧数据
请求3: 未过期 → 直接返回数据结构:
type ItemWithExpire struct {
Data Item `json:"data"`
ExpireAt int64 `json:"expire_at"` // 逻辑过期时间戳(毫秒)
}完整实现:
const (
cacheTTL = 24 * time.Hour // 物理过期时间(很长,或设 -1 永不物理过期)
logicTTL = 30 * time.Minute // 逻辑过期时间
lockTTL = 10 * time.Second
)
func GetHotItemLogical(ctx context.Context, id int64) (*Item, error) {
cacheKey := fmt.Sprintf("item:logical:%d", id)
lockKey := fmt.Sprintf("lock:item:logical:%d", id)
data, err := rdb.Get(ctx, cacheKey).Result()
if err != nil {
// 缓存中完全没有:正常查 DB 并写入(首次加载)
return loadAndCache(ctx, id, cacheKey)
}
var cached ItemWithExpire
if err := json.Unmarshal([]byte(data), &cached); err != nil {
return nil, err
}
// 判断是否逻辑过期
if time.Now().UnixMilli() < cached.ExpireAt {
// 未过期,直接返回
return &cached.Data, nil
}
// 已过期,尝试获取锁
locked, _ := rdb.SetNX(ctx, lockKey, "1", lockTTL).Result()
if locked {
// 获得锁:异步重建
go rebuildCache(context.Background(), id, cacheKey, lockKey)
}
// 无论是否获得锁,都返回旧数据
return &cached.Data, nil
}
func loadAndCache(ctx context.Context, id int64, cacheKey string) (*Item, error) {
item, err := db.QueryItem(ctx, id)
if err != nil {
return nil, err
}
wrapped := ItemWithExpire{
Data: *item,
ExpireAt: time.Now().Add(logicTTL).UnixMilli(),
}
data, _ := json.Marshal(wrapped)
rdb.Set(ctx, cacheKey, data, cacheTTL)
return item, nil
}
func rebuildCache(ctx context.Context, id int64, cacheKey, lockKey string) {
defer rdb.Del(ctx, lockKey)
// Double Check:可能其他线程已经重建过
data, _ := rdb.Get(ctx, cacheKey).Result()
if data != "" {
var cached ItemWithExpire
json.Unmarshal([]byte(data), &cached)
if time.Now().UnixMilli() < cached.ExpireAt {
return // 已被其他线程重建,无需再次重建
}
}
item, err := db.QueryItem(ctx, id)
if err != nil {
return
}
wrapped := ItemWithExpire{
Data: *item,
ExpireAt: time.Now().Add(logicTTL).UnixMilli(),
}
newData, _ := json.Marshal(wrapped)
rdb.Set(ctx, cacheKey, newData, cacheTTL)
}| 说明 | |
|---|---|
| 优点 | 几乎无等待,用户始终能拿到数据(即使是旧数据),性能极佳 |
| 缺点 | 短期数据不一致(返回旧数据)、实现稍复杂、统计信息需要逻辑过期判断 |
对比总结:
| 维度 | 互斥锁方案 | 逻辑过期方案 |
|---|---|---|
| 响应延迟 | 高(等待线程排队) | 低(直接返回旧数据) |
| 数据一致性 | 好(始终拿到最新数据) | 短期不一致 |
| 实现复杂度 | 低 | 中等 |
| 适用场景 | 一致性要求高(库存、余额) | 性能要求高(热门新闻、排行榜) |
2.4 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 缓存空对象忘记设 TTL | 数据已入库但缓存一直返回 null |
| 🚨 互斥锁忘记设超时 | 进程崩溃后锁永不释放,所有请求等待超时 |
| 🚨 逻辑过期中忘记 Double Check | 多个线程重复查 DB 重建缓存 |
| 🚨 布隆过滤器误判导致少量穿透 | 可接受,配合缓存空对象兜底 |
| 🚨 TTL 随机值范围过大 | 在合理范围内波动(如 5% ~ 20%),过大影响缓存命中率 |
3. 分布式锁
3.1 为什么需要分布式锁
在单机应用中,sync.Mutex 可以保证同一进程内只有一个 goroutine 操作共享资源。但在集群环境下,多个服务实例运行在不同 JVM/进程上,进程内锁无法互斥。
分布式锁方案对比:
| 方案 | 互斥原理 | 高可用 | 高性能 | 安全性 |
|---|---|---|---|---|
| MySQL | 行锁 / 表锁 / 唯一索引 | 好 | 一般 | 连接断开自动释放 |
| Redis | SETNX + 超时 | 好(集群) | 好 | 超时释放(需看门狗) |
| Zookeeper | 临时顺序节点 + Watch | 好 | 一般 | 连接断开自动释放 |
| etcd | MVCC + Lease 租约 | 极好 | 好 | 租约到期自动释放 |
Redis 方案因部署简单、性能高而最常用。
3.2 基础版本:SETNX + EX
# 获取锁:NX(不存在才设置)+ EX(超时时间,防止死锁)
SET lock:order:1001 uuid-12345 NX EX 30- NX:只有当 key 不存在时才设置成功,保证互斥
- EX 30:锁 30 秒后自动释放,防止客户端崩溃后死锁
- uuid-12345:唯一标识,释放锁时校验"锁是否还是我的"
释放锁
🚨 错误做法:直接
DEL lock:order:1001—— 可能删除别人持有的锁!
✅ 正确做法:先校验 value 是否匹配,匹配才删除。两步操作必须原子执行(Lua 脚本)。
-- unlock.lua
-- KEYS[1]: 锁的 key
-- ARGV[1]: 锁的 value(唯一标识)
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('DEL', KEYS[1])
else
return 0
end完整 Go 实现
package lock
import (
"context"
"errors"
"time"
"github.com/google/uuid"
"github.com/redis/go-redis/v9"
)
type RedisLock struct {
rdb *redis.Client
key string
value string // 唯一标识,释放时校验
expireSec int
}
// NewRedisLock 创建一个 Redis 分布式锁实例
func NewRedisLock(rdb *redis.Client, key string, expireSec int) *RedisLock {
return &RedisLock{
rdb: rdb,
key: key,
value: uuid.New().String(),
expireSec: expireSec,
}
}
// TryLock 尝试获取锁,返回是否成功
func (l *RedisLock) TryLock(ctx context.Context) (bool, error) {
return l.rdb.SetNX(ctx, l.key, l.value, time.Duration(l.expireSec)*time.Second).Result()
}
// Lock 阻塞获取锁,支持重试
func (l *RedisLock) Lock(ctx context.Context, retryTimes int, retryInterval time.Duration) (bool, error) {
for i := 0; i < retryTimes; i++ {
ok, err := l.TryLock(ctx)
if err != nil {
return false, err
}
if ok {
return true, nil
}
time.Sleep(retryInterval)
}
return false, errors.New("failed to acquire lock after retries")
}
// Unlock 释放锁(Lua 脚本保证原子性)
func (l *RedisLock) Unlock(ctx context.Context) error {
script := `
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('DEL', KEYS[1])
else
return 0
end
`
result, err := l.rdb.Eval(ctx, script, []string{l.key}, l.value).Result()
if err != nil {
return err
}
if result.(int64) == 0 {
return errors.New("lock already released by others or expired")
}
return nil
}
// 使用 go:embed 嵌入 Lua 脚本(推荐方式)
//go:embed scripts/unlock.lua
var unlockScript string
func (l *RedisLock) UnlockWithEmbed(ctx context.Context) error {
result, err := l.rdb.Eval(ctx, unlockScript, []string{l.key}, l.value).Result()
if err != nil {
return err
}
if result.(int64) == 0 {
return errors.New("lock already released by others or expired")
}
return nil
}使用示例:
func main() {
ctx := context.Background()
lock := NewRedisLock(rdb, "lock:order:1001", 30)
ok, err := lock.TryLock(ctx)
if err != nil || !ok {
log.Fatal("failed to acquire lock")
}
defer lock.Unlock(ctx)
// 执行业务逻辑
processOrder(ctx, 1001)
}3.3 分布式锁的问题与优化
问题一:不可重入
同一线程重复获取同一把锁会失败(SETNX 已有 key)。
解决方案:使用 Hash 结构,key 为锁名,field 为线程标识,value 为重入次数。
-- reentrant_lock.lua
local key = KEYS[1]
local threadId = ARGV[1]
local ttl = ARGV[2]
-- 如果锁不存在,或当前线程持有锁
if redis.call('EXISTS', key) == 0 or redis.call('HEXISTS', key, threadId) == 1 then
redis.call('HINCRBY', key, threadId, 1)
redis.call('EXPIRE', key, ttl)
return 1
end
return 0问题二:超时自动释放
业务执行时间超过锁的超时时间,锁自动释放,其他线程获取锁,导致并发问题。
解决方案:看门狗(Watch Dog)机制 —— 持有锁的客户端定期续期。
// WatchDog 自动续期
func (l *RedisLock) StartWatchDog(ctx context.Context) {
ticker := time.NewTicker(time.Duration(l.expireSec/3) * time.Second)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
// 续期:把锁的超时时间重置
script := `
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('EXPIRE', KEYS[1], ARGV[2])
end
return 0
`
result, _ := l.rdb.Eval(ctx, script, []string{l.key}, l.value, l.expireSec).Result()
if result.(int64) == 0 {
return // 锁已不属于当前线程,停止续期
}
}
}
}
// 使用示例
func ProcessWithWatchDog(ctx context.Context) error {
lock := NewRedisLock(rdb, "lock:task", 10)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go lock.StartWatchDog(ctx) // 启动看门狗
ok, _ := lock.TryLock(ctx)
if !ok {
return errors.New("lock failed")
}
defer lock.Unlock(ctx)
// 执行长时间任务...
time.Sleep(30 * time.Second)
return nil
}问题三:主从一致性问题
Redis 主从异步复制时:主节点写入锁后宕机,锁数据未同步到从节点,新主节点没有锁信息,其他客户端可以再次获取锁。
解决方案:Redlock 算法(见下一节)。
3.4 Redlock 算法
🔬 深入原理:Redlock 由 Redis 作者 antirez 提出。核心思想是向 N 个完全独立的 Redis 实例(非主从关系)依次获取锁,超过半数成功才算获取成功。
算法流程:
1. 获取当前时间 t1
2. 依次向 N 个 Redis 实例尝试获取锁(SET NX EX)
3. 计算总耗时 = 当前时间 - t1
4. 成功条件:
- 成功实例数 > N/2(过半数)
- 总耗时 < 锁的有效期
5. 实际有效时间 = 锁有效期 - 总耗时
6. 若失败:向所有实例发送释放锁命令争议与讨论:
| 观点 | 代表 | 论点 |
|---|---|---|
| 反对 | Martin Kleppmann | 依赖时钟单调性的假设不完全准确;GC 暂停可能导致锁超时后仍在执行 |
| 支持 | antirez | 锁本身就有自动过期作为安全网;实际 GC 暂停远小于锁有效期 |
💡 实际建议:大多数业务场景下单实例 Redis 分布式锁 + 主从哨兵足够可靠。Redlock 增加了运维复杂度和延迟,只有在对一致性要求极高的场景(如金融交易)才建议引入。
Go 实现 Redsync(Redlock 的 Go 库):
import (
"github.com/go-redsync/redsync/v4"
"github.com/go-redsync/redsync/v4/redis/goredis/v9"
)
func NewRedlock() *redsync.Redsync {
// 创建多个独立 Redis 连接池
pools := []redsync.Pool{
goredis.NewPool(rdb1),
goredis.NewPool(rdb2),
goredis.NewPool(rdb3),
goredis.NewPool(rdb4),
goredis.NewPool(rdb5),
}
return redsync.New(pools...)
}
func LockWithRedlock(ctx context.Context, rs *redsync.Redsync) error {
mutex := rs.NewMutex("lock:order:1001",
redsync.WithExpiry(8*time.Second),
redsync.WithTries(3), // 重试次数
redsync.WithRetryDelay(100*time.Millisecond),
)
if err := mutex.LockContext(ctx); err != nil {
return err
}
defer mutex.UnlockContext(ctx)
// 业务逻辑
return nil
}3.5 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 释放锁时不校验 value | 可能删除别人持有的锁 |
| 🚨 锁超时时间设置过短 | 业务没执行完锁就过期,其他线程获取锁 |
| 🚨 未启用看门狗 | 不知道业务执行多久时,锁过期是关键风险 |
| 🚨 忘记设超时时间 | 客户端崩溃,锁永不释放 → 死锁 |
| 🚨 使用 Redlock 但实例共享了同一个物理机 | 宿主机宕机,所有实例同时不可用 |
4. 优惠券秒杀实战
4.1 全局 ID 生成器
秒杀场景需要全局唯一、趋势递增的订单 ID,不能依赖数据库自增(单点瓶颈)。
特性要求:
- 唯一性:全局唯一
- 高可用:随时可用,无单点故障
- 高性能:高并发下低延迟
- 递增性:趋势递增(利于数据库索引)
- 安全性:无法被猜测规律
64 位 ID 结构:
| 1 bit | 31 bit | 32 bit |
|--------|------------------|--------------------|
| 符号位 | 时间戳(秒) | 序列号(自增) |为什么是 31 bit 时间戳?
- 31 bit 可表示约 68 年(2^31 秒),从 2024 年开始可用到 2092 年
- 32 bit 序列号:单机每秒可生成 2^32 = 42 亿 ID
package idgen
import (
"context"
"fmt"
"time"
"github.com/redis/go-redis/v9"
)
const BeginTimeStamp uint64 = 1710000000 // 自定义起始时间戳
var rdb *redis.Client
// GenId 生成全局唯一 ID
func GenId(ctx context.Context, keyPrefix string) (int64, error) {
// 1. 生成时间戳部分(当前时间 - 基准时间)
now := time.Now().Unix()
timePart := uint64(now) - BeginTimeStamp
// 2. 使用 Redis INCR 生成序列号(原子操作、保证唯一)
dateKey := fmt.Sprintf("icr:%s:%s", keyPrefix, time.Now().Format("2006:01:02"))
seq, err := rdb.Incr(ctx, dateKey).Result()
if err != nil {
return 0, fmt.Errorf("redis incr failed: %w", err)
}
// 序列号 key 一天后自动过期
rdb.Expire(ctx, dateKey, 24*time.Hour)
// 3. 拼接:时间戳左移 32 位 | 序列号
id := int64((timePart << 32) | uint64(seq))
return id, nil
}
// ParseId 从 ID 中解析出时间戳和序列号(调试用)
func ParseId(id int64) (timestamp uint64, seq uint64) {
u := uint64(id)
return (u >> 32) + BeginTimeStamp, u & 0xFFFFFFFF
}⚡ 性能提示:Redis INCR 是内存操作,单机 QPS 可达 10w+。如果需要更高性能,可以预取一批 ID 到本地,用完再取下一批,减少 Redis 网络开销。
4.2 超卖问题
问题描述:并发减库存时,多个线程同时读到 stock > 0 后都执行扣减,导致库存变为负数。
stock = 1
线程1: read stock=1 → 判断 >0 → 扣减
线程2: read stock=1 → 判断 >0 → 扣减 (并发)
结果: stock = -1 ← 超卖!解决方案
方案一:乐观锁 - CAS 法
-- 扣减时判断 stock > 0
UPDATE seckill_voucher
SET stock = stock - 1
WHERE voucher_id = ? AND stock > 0;func DeductStock(ctx context.Context, voucherID int64) error {
result, err := db.Exec(ctx,
"UPDATE seckill_voucher SET stock=stock-1 WHERE voucher_id=? AND stock>0",
voucherID,
)
if err != nil {
return err
}
rowsAffected, _ := result.RowsAffected()
if rowsAffected == 0 {
return errors.New("stock exhausted")
}
return nil
}🔬 深入原理:MySQL InnoDB 在 UPDATE ... WHERE stock > 0 时会对匹配的行加行锁。并发线程中只有一个能拿到锁并执行更新。affected_rows = 0 表示库存为 0 或已被其他线程抢完。
方案二:乐观锁 - 版本号法
-- 扣减时检查版本号
UPDATE seckill_voucher
SET stock = stock - 1, version = version + 1
WHERE voucher_id = ? AND version = ?;版本号法比 CAS 法更通用,适用于不止简单扣减的场景(如修改商品信息)。
方案三:悲观锁(不推荐用于秒杀)
-- 使用 SELECT ... FOR UPDATE
SELECT stock FROM seckill_voucher WHERE voucher_id = ? FOR UPDATE;
-- 判断 stock > 0 后执行 UPDATE悲观锁性能差,不适合高并发秒杀场景。
超卖方案对比:
| 方案 | 原理 | 性能 | 实现难度 |
|---|---|---|---|
| CAS 乐观锁 | SET stock=stock-1 WHERE stock>0 |
高 | 低 |
| 版本号乐观锁 | WHERE version=? |
高 | 中 |
| 悲观锁 | SELECT FOR UPDATE |
低 | 低 |
| Redis 分布式锁 | SETNX | 中 | 中 |
| Redis 原子操作 | DECR stock + 判断 >=0 |
极高 | 低 |
💡 最佳实践:Redis DECR 原子扣减 + 乐观 CAS 兜底。先用 Redis DECR 做快速扣减,成功后异步写 DB。Redis 保证最终一致性,DB 保证数据持久。
4.3 一人一单
需求:同一个用户对同一优惠券只能下一单。
方案一:数据库唯一索引
ALTER TABLE seckill_order ADD UNIQUE INDEX idx_user_voucher (user_id, voucher_id);插入时依赖唯一索引约束,冲突即返回重复下单。问题是 DB 压力大(每次都要插 DB)。
方案二:Redis Set 去重
func CheckAndOrder(ctx context.Context, userID, voucherID int64) error {
member := fmt.Sprintf("%d", userID)
// 使用 SADD 的返回值判断:返回 1 表示首次添加(未下单),0 表示已存在(已下单)
added, err := rdb.SAdd(ctx, fmt.Sprintf("order:set:%d", voucherID), member).Result()
if err != nil {
return err
}
if added == 0 {
return errors.New("duplicate order")
}
// 这里还需要原子性地检查库存...
return nil
}🚨 上述代码分开执行库存检查和一人一单判断,不是原子操作!
方案三:Lua 脚本原子执行(推荐)
-- seckill.lua
-- KEYS[1]: 库存 key (stock:voucher:id)
-- KEYS[2]: 已下单用户集合 key (order:set:voucher:id)
-- ARGV[1]: 用户 ID
-- ARGV[2]: 订单 ID
local stock = tonumber(redis.call('GET', KEYS[1]))
if stock == nil or stock <= 0 then
return -1 -- 库存不足
end
-- 检查用户是否已下单
if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 1 then
return -2 -- 重复下单
end
-- 扣减库存
redis.call('DECR', KEYS[1])
-- 记录用户已下单
redis.call('SADD', KEYS[2], ARGV[1])
-- 将订单信息放入消息队列,异步写入 DB
redis.call('LPUSH', 'order:queue', ARGV[2])
return 1 -- 秒杀成功func Seckill(ctx context.Context, voucherID, userID int64, orderID string) (int, error) {
script := `
local stock = tonumber(redis.call('GET', KEYS[1]))
if stock == nil or stock <= 0 then
return -1
end
if redis.call('SISMEMBER', KEYS[2], ARGV[1]) == 1 then
return -2
end
redis.call('DECR', KEYS[1])
redis.call('SADD', KEYS[2], ARGV[1])
redis.call('LPUSH', KEYS[3], ARGV[2])
return 1
`
stockKey := fmt.Sprintf("stock:voucher:%d", voucherID)
userKey := fmt.Sprintf("order:set:%d", voucherID)
queueKey := "order:queue"
result, err := rdb.Eval(ctx, script, []string{stockKey, userKey, queueKey},
userID, orderID).Int()
if err != nil {
return 0, err
}
return result, nil
}4.4 秒杀优化思路
💡 最佳实践:前端分离 + Redis 预减库存 + 消息队列异步下单。
用户请求 → 前端:验证码 + 限流 + 按钮防抖
↓
网关:IP/用户级限流
↓
Redis:Lua 脚本原子判断库存 + 一人一单
↓
消息队列(List/Stream)
↓
独立 Worker 异步写入 MySQL优势:
| 收益 | 说明 |
|---|---|
| 削峰 | 消息队列缓冲瞬间高并发,保护数据库 |
| 解耦 | 扣库存与写订单分离,互不影响 |
| 提高吞吐 | Redis 判断在内存中完成,毫秒级响应 |
异步 Worker 消费:
func StartOrderWorker(ctx context.Context) {
for {
// BRPOP 阻塞弹出,超时 5 秒
result, err := rdb.BRPop(ctx, 5*time.Second, "order:queue").Result()
if err != nil {
continue
}
orderID := result[1]
// 写入数据库
if err := db.CreateOrder(ctx, orderID); err != nil {
log.Printf("create order failed: %v", err)
// 写入死信队列,人工处理
rdb.LPush(ctx, "order:dlq", orderID)
}
}
}4.5 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 库存检查 + 扣减不是原子操作 | 多线程读到同一库存值 → 超卖 |
| 🚨 Redis DECR 后不判断是否 >= 0 | 可能导致库存变为负数后再发货 |
| 🚨 一人一单与库存扣减分离 | 可能在库存扣减成功后才发现重复下单 |
| 🚨 异步下单不处理失败 | 订单丢失,用户钱已扣但没生成订单 |
5. Feed 流实现
5.1 模式选择
| 模式 | 说明 | 典型产品 |
|---|---|---|
| TimeLine(时间线) | 按时间排序,信息全面,关注内容全部呈现 | 微信朋友圈、微博关注流 |
| 智能排序 | 算法推荐,个性化排序 | 抖音、小红书推荐流 |
本节聚焦 TimeLine 模式的实现。
5.2 实现方案对比
| 方案 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| 拉模式(读扩散) | 用户查看 Feed 时实时拉取所有关注者的最新内容 | 节省存储空间 | 读取慢(N 次查询) |
| 推模式(写扩散) | 发布内容时推送到所有粉丝的收件箱 | 读取极快(1 次查询) | 存储膨胀、写放大(大 V 给百万粉丝写) |
| 推拉结合 | 大 V 用拉模式,普通用户用推模式 | 平衡读写 | 实现复杂、需判断用户类型 |
💡 最佳实践:推拉结合。粉丝数 < 阈值的用户发布内容时推模式写入粉丝收件箱;大 V 不推送,粉丝查阅时拉模式实时拉取。
const bigVFollowersThreshold = 10000
func PublishPost(ctx context.Context, userID int64, postID string, content string) error {
score := float64(time.Now().UnixMilli())
followerCount := getUserFollowerCount(ctx, userID)
if followerCount < bigVFollowersThreshold {
// 普通用户:推模式,写入所有粉丝的收件箱
followers := getFollowers(ctx, userID)
pipe := rdb.Pipeline()
for _, followerID := range followers {
feedKey := fmt.Sprintf("feed:%d", followerID)
pipe.ZAdd(ctx, feedKey, redis.Z{Score: score, Member: postID})
}
_, err := pipe.Exec(ctx)
return err
}
// 大 V:仅写入自己的发布列表,粉丝读取时拉取
outboxKey := fmt.Sprintf("outbox:%d", userID)
return rdb.ZAdd(ctx, outboxKey, redis.Z{Score: score, Member: postID}).Err()
}
func GetFeed(ctx context.Context, userID int64, lastScore float64, limit int) ([]string, error) {
feedKey := fmt.Sprintf("feed:%d", userID)
// 推模式:直接读取自己的收件箱
posts, err := rdb.ZRevRangeByScore(ctx, feedKey, &redis.ZRangeBy{
Min: "0",
Max: fmt.Sprintf("%f", lastScore),
Offset: 0,
Count: int64(limit),
}).Result()
if err != nil {
return nil, err
}
// 拉模式:拉取关注的大 V 的最新内容
bigVs := getBigVFollowing(ctx, userID)
for _, bvID := range bigVs {
bvPosts, _ := rdb.ZRevRangeByScore(ctx, fmt.Sprintf("outbox:%d", bvID), &redis.ZRangeBy{
Min: "0",
Max: fmt.Sprintf("%f", lastScore),
Offset: 0,
Count: int64(limit),
}).Result()
posts = append(posts, bvPosts...)
}
return posts, nil
}5.3 ZSet 实现滚动分页
🔬 深入原理:传统 LIMIT offset, limit 分页在 Feed 流中有数据重复和遗漏问题。因为在两次分页请求之间可能有新内容插入。
第1页: OFFSET 0 LIMIT 3 → [item5, item4, item3]
(此时 item6 插入,排在首位)
第2页: OFFSET 3 LIMIT 3 → [item2, item1, item0]
↑
item3 丢失!游标分页使用 score 作为游标:每次查询返回最后一条记录的 score,下次查询用 score < lastScore 作为条件。
// FeedPage 滚动分页查询
type FeedPage struct {
Items []string `json:"items"`
LastScore float64 `json:"last_score"` // 下一页游标
HasMore bool `json:"has_more"` // 是否还有更多
}
func GetFeedByCursor(ctx context.Context, userID int64, lastScore float64, limit int) (*FeedPage, error) {
feedKey := fmt.Sprintf("feed:%d", userID)
limitPlusOne := int64(limit + 1) // 多取一条,判断 has_more
maxScore := fmt.Sprintf("%f", lastScore)
if lastScore == 0 {
maxScore = "+inf" // 首次请求,从最新开始
}
// ZREVRANGEBYSCORE:按 score 倒序,从 maxScore 开始(不含)
results, err := rdb.ZRevRangeByScoreWithScores(ctx, feedKey, &redis.ZRangeBy{
Min: "-inf",
Max: maxScore,
Offset: 0,
Count: limitPlusOne,
}).Result()
if err != nil {
return nil, err
}
page := &FeedPage{
Items: make([]string, 0, limit),
HasMore: len(results) > limit,
}
for i, z := range results {
if i >= limit {
break
}
page.Items = append(page.Items, z.Member.(string))
}
if len(results) > 0 {
lastIdx := min(limit, len(results)) - 1
page.LastScore = results[lastIdx].Score
}
return page, nil
}🚨 注意相同 score 的问题:如果两条记录的 score(时间戳)相同,第一次查询可能取到第 1 条而第二次查询跳过第 2 条。解决方案:查询时使用上一次的最小 score + 上次查询到的最后一个 member ID 双重游标。
5.4 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 推模式下大 V 发布导致写放大 | 千万粉丝的大 V 写一次要写千万条,延迟极高 |
| 🚨 使用 OFFSET 分页 | 新数据插入导致数据重复/遗漏 |
| 🚨 相同 score 导致分页遗漏 | 需要双重游标(score + member ID) |
| 🚨 收件箱无限增长 | 需要定期清理或设置最大长度(ZREMRANGEBYRANK) |
6. 附近商铺(GEO 实战)
Redis GEO 基于 ZSet 实现,使用 GeoHash 算法将经纬度编码为 score,支持附近位置查询。
# 添加商铺地理坐标
GEOADD shops:location 116.404 39.915 "shop:1" # 北京
GEOADD shops:location 121.473 31.230 "shop:2" # 上海
GEOADD shops:location 113.264 23.129 "shop:3" # 广州
GEOADD shops:location 114.057 22.543 "shop:4" # 深圳
# 搜索指定坐标 5km 范围内的商铺
GEOSEARCH shops:location FROMLONLAT 116.397 39.908 BYRADIUS 5 km WITHDIST
# 按距离排序
GEOSEARCH shops:location FROMLONLAT 116.397 39.908 BYRADIUS 20 km WITHDIST ASC
# 计算两点距离
GEODIST shops:location "shop:1" "shop:2" kmGo 实现:
type Shop struct {
ID string `json:"id"`
Name string `json:"name"`
Distance float64 `json:"distance_km"`
}
// AddShopLocation 添加商铺地理坐标
func AddShopLocation(ctx context.Context, shopID string, lng, lat float64) error {
return rdb.GeoAdd(ctx, "shops:location",
&redis.GeoLocation{
Name: shopID,
Longitude: lng,
Latitude: lat,
},
).Err()
}
// SearchNearbyShops 搜索附近商铺(带滚动分页)
func SearchNearbyShops(ctx context.Context, lng, lat, radius float64, unit string, count int) ([]Shop, error) {
locations, err := rdb.GeoSearchLocation(ctx, "shops:location",
&redis.GeoSearchLocationQuery{
GeoSearchQuery: redis.GeoSearchQuery{
Longitude: lng,
Latitude: lat,
Radius: radius,
RadiusUnit: unit,
Sort: "ASC",
Count: count,
},
WithDist: true,
},
).Result()
if err != nil {
return nil, err
}
shops := make([]Shop, 0, len(locations))
for _, loc := range locations {
shops = append(shops, Shop{
ID: loc.Name,
Distance: loc.Dist,
})
}
return shops, nil
}⚡ 性能提示:GeoHash 的精度(即 ZSet score 的整数部分)受 key 名前缀影响。如果数据量大,可以按城市分 key(如
shops:location:beijing),减少单 key 数据量。
6.1 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 全国数据全放一个 key | 数据量过大导致查询变慢、内存占用高 |
| 🚨 GEO 分页依赖 ZSet 分页 | 同样存在相同 score 的遗漏问题 |
| 🚨 GeoHash 编码精度有限 | 极近距离(< 1m)精度不够 |
7. 用户签到(Bitmap 实战)
Redis Bitmap 用每个 bit 表示一个状态,非常适合记录用户每日签到。
# user_id=1001 的用户在第 1 天签到
SETBIT sign:2026:1001 0 1
# user_id=1001 的用户在第 7 天签到
SETBIT sign:2026:1001 6 1
# 统计第 1 到第 7 天的签到天数
BITCOUNT sign:2026:1001 0 6
# 获取第 7 天的签到状态
GETBIT sign:2026:1001 6
# BITFIELD:查看最近 7 天的签到情况(无符号,偏移从 0 开始)
BITFIELD sign:2026:1001 GET u7 0Go 实现:
// SignIn 用户签到
func SignIn(ctx context.Context, userID int64, year int, dayOfYear int) error {
key := fmt.Sprintf("sign:%d:%d", year, userID)
// dayOfYear 从 1 开始,offset 从 0 开始
return rdb.SetBit(ctx, key, int64(dayOfYear-1), 1).Err()
}
// GetSignInStatus 获取指定日的签到状态
func GetSignInStatus(ctx context.Context, userID int64, year int, dayOfYear int) (bool, error) {
key := fmt.Sprintf("sign:%d:%d", year, userID)
bit, err := rdb.GetBit(ctx, key, int64(dayOfYear-1)).Result()
return bit == 1, err
}
// GetTotalSignDays 统计用户某年累计签到天数
func GetTotalSignDays(ctx context.Context, userID int64, year int) (int64, error) {
key := fmt.Sprintf("sign:%d:%d", year, userID)
return rdb.BitCount(ctx, key, &redis.BitCount{
Start: 0,
End: 365, // 覆盖全年
}).Result()
}
// GetConsecutiveSignDays 统计到今天的连续签到天数
func GetConsecutiveSignDays(ctx context.Context, userID int64, year int, todayDayOfYear int) (int64, error) {
key := fmt.Sprintf("sign:%d:%d", year, userID)
// 1. 取从 0 到今天的所有位
bits, err := rdb.BitField(ctx, key,
"GET", fmt.Sprintf("u%d", todayDayOfYear), "0",
).Result()
if err != nil || len(bits) == 0 {
return 0, err
}
// 2. 从右向左逐位检查,遇 0 停止
val := bits[0]
var consecutive int64
for i := 0; i < todayDayOfYear; i++ {
if val&1 == 1 {
consecutive++
val >>= 1
} else {
break
}
}
return consecutive, nil
}
// GetSignMapOfMonth 获取某月的签到日历(用于前端展示)
func GetSignMapOfMonth(ctx context.Context, userID int64, year int, month int, daysInMonth int) ([]bool, error) {
key := fmt.Sprintf("sign:%d:%d", year, userID)
startDay := dayOfYear(year, month, 1)
pipe := rdb.Pipeline()
cmds := make([]*redis.IntCmd, daysInMonth)
for i := 0; i < daysInMonth; i++ {
cmds[i] = pipe.GetBit(ctx, key, int64(startDay+i-1))
}
pipe.Exec(ctx)
result := make([]bool, daysInMonth)
for i, cmd := range cmds {
result[i] = cmd.Val() == 1
}
return result, nil
}⚡ 性能提示:一年 365 天的签到数据仅占 46 字节(365 bits)。一亿用户的签到仅约 4.3 GB,远小于传统的关系型存储。
7.1 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 offset 从 0 开始 | BITFIELD GET 的偏移量也是从 0 开始 |
| 🚨 BITFIELD 返回的位顺序 | 返回值按 LSB 方向排列,从右向左读 |
| 🚨 跨天判断逻辑未考虑凌晨 | 用户凌晨签到,当天还没结束也算连续 |
8. UV 统计(HyperLogLog 实战)
HyperLogLog (HLL) 是一种基数估计算法,用于统计集合中不重复元素的数量。标准误差 0.81%。
# 记录用户访问(UV)
PFADD uv:page:homepage user:1001 user:1002 user:1003
# 统计 UV
PFCOUNT uv:page:homepage # → 3
# 重复添加不会增加计数
PFADD uv:page:homepage user:1001
PFCOUNT uv:page:homepage # → 还是 3
# 合并多天数据
PFADD uv:2026-07-09 user:1001 user:1002
PFADD uv:2026-07-10 user:1002 user:1003
PFMERGE uv:2026-07-09:10 uv:2026-07-09 uv:2026-07-10
PFCOUNT uv:2026-07-09:10 # → 3(1001, 1002, 1003 去重后)Go 实现:
// RecordUV 记录页面访问 UV
func RecordUV(ctx context.Context, pageKey string, userID int64) error {
key := fmt.Sprintf("uv:%s", pageKey)
return rdb.PFAdd(ctx, key, userID).Err()
}
// GetUV 统计指定页面 UV
func GetUV(ctx context.Context, pageKey string) (int64, error) {
key := fmt.Sprintf("uv:%s", pageKey)
return rdb.PFCount(ctx, key).Result()
}
// MergeDailyUV 合并多天 UV(如周报、月报统计)
func MergeDailyUV(ctx context.Context, pageKey string, dates []string) (int64, error) {
// 构建每天的 key
keys := make([]string, len(dates))
for i, date := range dates {
keys[i] = fmt.Sprintf("uv:%s:%s", pageKey, date)
}
mergedKey := fmt.Sprintf("uv:%s:merged:tmp", pageKey)
// 合并
if err := rdb.PFMerge(ctx, mergedKey, keys...).Err(); err != nil {
return 0, err
}
count, err := rdb.PFCount(ctx, mergedKey).Result()
rdb.Del(ctx, mergedKey) // 清理临时 key
return count, err
}🔬 深入原理:HyperLogLog 使用概率算法,每个 key 仅占用 12 KB 内存,无论统计的元素数量有多大。对于百万级 UV 统计,传统 Set 需要几十 MB 内存,而 HLL 仅需 12 KB。代价是约 0.81% 的标准误差。
| 方法 | 内存占用 | 精确度 |
|---|---|---|
| Redis Set | 取决于元素数量(百万 UV 约 50 MB) | 100% 精确 |
| HyperLogLog | 固定 12 KB | 约 0.81% 标准误差 |
8.1 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 低基数时误差大 | HLL 在基数 < 几千时误差百分比可能较大 |
| 🚨 把 HLL 当 Set 用 | 无法获取具体元素列表(不能 SMEMBERS) |
| 🚨 PFMERGE 后立刻 DELETE 源 key | 如果合并是周期性的,考虑保留源 key 用于去重 |
9. 短信登录
9.1 实现流程
用户输入手机号
↓
后端生成6位验证码 → 存入 Redis(key=phone, TTL=5min)→ 发送短信
↓
用户输入验证码
↓
后端从 Redis 取出验证码校验 → 校验通过 → 查DB是否有该用户
↓ ↓
已注册:生成 token 未注册:自动注册
↓ ↓
token 存入 Redis 生成 token 存入 Redis
↓ ↓
返回 token 给客户端 ←←←←←←←←←←←←←←←←←←←←←←←←←←9.2 完整实现
type LoginService struct {
rdb *redis.Client
db *sql.DB
}
// SendCode 发送短信验证码
func (s *LoginService) SendCode(ctx context.Context, phone string) error {
// 1. 校验手机号格式
if !isValidPhone(phone) {
return errors.New("invalid phone number")
}
// 2. 检查是否 60 秒内已发送(防止短信轰炸)
rateKey := fmt.Sprintf("sms:rate:%s", phone)
exists, _ := s.rdb.Exists(ctx, rateKey).Result()
if exists > 0 {
return errors.New("please wait 60 seconds before requesting another code")
}
// 3. 生成 6 位验证码
code := fmt.Sprintf("%06d", rand.Intn(1000000))
// 4. 存入 Redis(5 分钟有效)
codeKey := fmt.Sprintf("sms:code:%s", phone)
err := s.rdb.Set(ctx, codeKey, code, 5*time.Minute).Err()
if err != nil {
return fmt.Errorf("store code failed: %w", err)
}
// 5. 设置 60 秒发送间隔
s.rdb.Set(ctx, rateKey, "1", 60*time.Second)
// 6. 调用短信服务商发送验证码
go sendSMS(phone, code)
return nil
}
// Login 验证码登录
func (s *LoginService) Login(ctx context.Context, phone, code string) (string, error) {
// 1. 校验验证码
codeKey := fmt.Sprintf("sms:code:%s", phone)
storedCode, err := s.rdb.Get(ctx, codeKey).Result()
if errors.Is(err, redis.Nil) {
return "", errors.New("code expired or not sent")
}
if err != nil {
return "", err
}
if storedCode != code {
// 验证失败不删除验证码(保留重试机会),但记录失败次数
failKey := fmt.Sprintf("sms:fail:%s", phone)
fails, _ := s.rdb.Incr(ctx, failKey).Result()
s.rdb.Expire(ctx, failKey, 5*time.Minute)
if fails >= 5 {
s.rdb.Del(ctx, codeKey) // 失败 5 次,删除验证码
return "", errors.New("too many attempts, code invalidated")
}
return "", errors.New("invalid code")
}
// 2. 验证通过,删除验证码
s.rdb.Del(ctx, codeKey)
s.rdb.Del(ctx, fmt.Sprintf("sms:fail:%s", phone))
// 3. 查询或创建用户
user, err := s.db.FindUserByPhone(ctx, phone)
if errors.Is(err, sql.ErrNoRows) {
user = &User{Phone: phone}
if err := s.db.CreateUser(ctx, user); err != nil {
return "", err
}
} else if err != nil {
return "", err
}
// 4. 生成 token 并存入 Redis
token := generateToken()
tokenKey := fmt.Sprintf("token:%s", token)
sessionData := map[string]interface{}{
"user_id": user.ID,
"phone": phone,
}
sessionBytes, _ := json.Marshal(sessionData)
s.rdb.Set(ctx, tokenKey, sessionBytes, 7*24*time.Hour)
return token, nil
}
// GetCurrentUser 根据 token 获取当前登录用户
func (s *LoginService) GetCurrentUser(ctx context.Context, token string) (*User, error) {
tokenKey := fmt.Sprintf("token:%s", token)
data, err := s.rdb.Get(ctx, tokenKey).Result()
if errors.Is(err, redis.Nil) {
return nil, errors.New("token expired or invalid")
}
if err != nil {
return nil, err
}
var session map[string]interface{}
json.Unmarshal([]byte(data), &session)
// 刷新 token 有效期(每次访问续期 7 天)
s.rdb.Expire(ctx, tokenKey, 7*24*time.Hour)
userID := int64(session["user_id"].(float64))
return s.db.FindUserByID(ctx, userID)
}9.3 安全优化
| 措施 | 说明 |
|---|---|
| 发送频率限制 | 同一手机号 60 秒内只能请求一次验证码 |
| 验证次数限制 | 同一验证码最多验证 5 次,超过即失效 |
| 验证码 TTL | 5 分钟过期,防止暴力枚举 |
| IP 限制 | 同一 IP 每小时最多请求 N 次验证码 |
| Token 续期 | 每次访问自动续期,活跃用户不会掉线 |
9.4 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 验证码不设验证次数上限 | 6 位数字码可被暴力枚举 |
| 🚨 token 存入 Redis 不设置 TTL | 僵尸用户占用内存 |
| 🚨 验证码存入 Redis 但不关联 session | 验证码可被跨 session 复用 |
| 🚨 登录后没有做穿透保护 | 恶意请求不存在用户导致缓存击穿 |
10. 多级缓存
10.1 架构概览
客户端请求
↓
CDN(静态资源)
↓
Nginx / OpenResty(本地 L1 缓存,LRU,内存级别)
↓ miss
Redis 集群(L2 缓存)
↓ miss
MySQL(持久层)OpenResty 简介:OpenResty 基于 Nginx + LuaJIT,能在 Nginx 层面执行 Lua 脚本,实现高性能 Web 服务和本地缓存。适合在网关层做一层本地内存级缓存。
-- OpenResty Lua 示例:在 nginx.conf 中嵌入本地缓存
lua_shared_dict local_cache 100m;
location /api/item {
content_by_lua_block {
local cache = ngx.shared.local_cache
local id = ngx.var.arg_id
local key = "item:" .. id
-- L1: 本地缓存
local val = cache:get(key)
if val then
ngx.say(val)
return
end
-- L2: 请求后端 Redis 服务
local res = ngx.location.capture("/redis/" .. id)
if res.status == ngx.HTTP_OK then
cache:set(key, res.body, 300) -- 本地缓存 300s
ngx.say(res.body)
return
end
ngx.say('{"error": "not found"}')
}
}10.2 数据同步策略
| 策略 | 原理 | 时效性 | 耦合度 | 适用场景 |
|---|---|---|---|---|
| 设置有效期 | 缓存 TTL 到期后自动失效,下次读取拉取最新数据 | 差 | 无 | 低时效要求 |
| 同步双写 | 修改 DB 的同时更新/删除缓存 | 强 | 高 | 强一致性要求 |
| 异步通知 | 修改 DB 后发送事件(MQ/CDC),下游消费者更新缓存 | 一般 | 低 | 大多数业务(推荐) |
10.3 Canal(阿里巴巴 MySQL binlog 同步方案)
🔬 深入原理:Canal 是阿里巴巴开源的 MySQL binlog 增量订阅 & 消费组件。原理是伪装为 MySQL 从节点,监听并解析 binlog 变更事件。
MySQL 主从复制原理:
MySQL Master 写入数据 → 产生 binlog
↓
Canal Server(伪装成 slave)dump binlog
↓
Canal Server 解析 binlog → 转换为事件
↓
Canal Client 消费事件 → 更新 Redis / ES / KafkaCanal 架构:
MySQL Master
↓ binlog replication
Canal Server (单机或集群)
↓ parse & push
Canal Client (Java / Go)
↓
Redis / Elasticsearch / Kafka / Custom Sink配置要点:
- 在 MySQL 中为 Canal 创建专用账号并赋予 replication 权限:
-- MySQL 8.0+
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal_password';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;🚨 MySQL 8.0+ 的
GRANT语句不能带IDENTIFIED BY,需要先用CREATE USER创建账户。
- 在 MySQL 配置文件中启用 binlog(ROW 模式):
[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server-id=1- 配置 Canal Server 的
instance.properties:
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal_password
canal.instance.filter.regex=your_db\\..*Go 客户端示例(使用 canal-go):
import "github.com/withlin/canal-go/client"
func StartCanalClient() {
connector := client.NewSimpleCanalConnector(
"127.0.0.1", 11111,
"", "", "example", 60000, 60*60*1000,
)
connector.Connect()
connector.Subscribe("your_db\\.your_table")
for {
message, _ := connector.Get(100, nil, nil)
for _, entry := range message.Entries {
if entry.GetEntryType() == canal.EntryType_ROWDATA {
// 解析 binlog 事件
rowChange := &canal.RowChange{}
proto.Unmarshal(entry.GetStoreValue(), rowChange)
if rowChange.GetEventType() == canal.EventType_UPDATE {
for _, row := range rowChange.GetRowDatas() {
// row.GetBeforeColumns() → 旧值
// row.GetAfterColumns() → 新值
// 更新 Redis 缓存...
updateRedisCache(row.GetAfterColumns())
}
}
}
}
connector.Ack(message.Id)
}
}💡 最佳实践:使用 Canal 监听 binlog 是实现缓存与 DB 最终一致性的最佳方式之一。数据变更后 Canal 自动推送更新事件,无需在业务代码中维护双写逻辑,解耦度高且保证最终一致性。
10.4 常见陷阱
| 陷阱 | 说明 |
|---|---|
| 🚨 L1 本地缓存与 L2 Redis 数据不同步 | 需要合理的 TTL 组合 + 主动失效机制 |
| 🚨 Canal 单点故障 | 部署 Canal 集群(多 Server + ZooKeeper) |
| 🚨 binlog 格式不是 ROW | STATEMENT/MIXED 格式无法获取完整的行变更数据 |
| 🚨 MySQL 8.0 授权语法不兼容 | GRANT 不能带 IDENTIFIED BY |
11. 常见陷阱总汇
| 陷阱 | 分类 | 说明 |
|---|---|---|
| 🚨 缓存与 DB 双写不一致 | 缓存更新 | 先删缓存再更新 DB 导致并发写回旧值 |
| 🚨 缓存预热不足 | 运维 | 服务重启后大量请求同时查 DB,压力骤增 |
| 🚨 大 value 导致网络带宽打满 | 性能 | 缓存过大对象(> 1MB)拖慢 Redis 整体响应 |
| 🚨 热 key 发现与处理不及时 | 运维 | 单个 key 承载极高 QPS,导致集群热点不均衡 |
| 🚨 缓存雪崩的连锁反应 | 容量 | 缓存过期 → DB 过载 → 服务崩溃 → 重启后缓存全空 → 循环 |
| 🚨 分布式锁忘记设置超时 | 分布式锁 | 客户端崩溃后锁永不释放 → 死锁 |
| 🚨 释放别人的锁 | 分布式锁 | 不用 Lua 脚本原子校验 value 就 DEL |
| 🚨 ZSet 分页滚动时 score 相同导致遗漏 | Feed 流 | 需要双重游标(score + member ID) |
| 🚨 HyperLogLog 在低基数时误差较大 | UV 统计 | 基数 < 1000 时误差百分比可能超过 5% |
| 🚨 采用大 V 推模式后未处理写放大 | Feed 流 | 大 V 发布一条内容要写千万粉丝,延迟爆炸 |
| 🚨 验证码接口无防刷 | 安全 | 短信费用被恶意消耗 |
| 🚨 Redis 作为唯一数据源不做持久化 | 可靠性 | Redis 宕机丢失全部数据 |
| 🚨 布隆过滤器初始化后无法删除 | 缓存穿透 | 误判率随数据变更而上升 |
| 🚨 使用 KEYS 命令查询 | 性能 | KEYS 是 O(N) 阻塞命令,线上用 SCAN |
| 🚨 不设置 maxmemory + 淘汰策略 | 容量 | 内存撑爆导致 OOM 或直接拒绝写入 |
本文覆盖了 Redis 缓存实战的核心知识点,从缓存策略到三大经典问题,从分布式锁到秒杀系统,从 Feed 流到多级缓存架构。核心思想是:缓存是空间换时间,要在一致性、性能和维护成本之间找到平衡点。实际项目中,优先使用成熟的中间件(如 Redisson/RedisStack)和已验证的模式(如 Cache Aside),避免过度设计。