缓存与数据一致性

缓存策略、分布式一致性、并发控制

缓存策略

【Standard】缓存采用「Cache Aside（旁路缓存）」模式；读多写少场景使用缓存；缓存 Key 必须带业务前缀和版本号；必须设置过期时间。

1.1 缓存模式

模式	说明	适用场景
Cache Aside	应用维护缓存，读时回填	通用场景（推荐）
Read/Write Through	缓存层代理读写	对一致性要求高
Write Behind	异步写库	写多读少、可容忍丢失

1.2 Cache Aside 模式（推荐）

读流程：

1. 先查缓存，命中则返回
2. 缓存未命中，查数据库
3. 将结果写入缓存（设过期时间）
4. 返回结果

写流程：

1. 先更新数据库
2. 再删除缓存（不是更新缓存）

// 读
public User getUser(Long id) {
    String key = "user:" + id;
    User user = redisTemplate.get(key);
    if (user != null) return user;
    user = userMapper.selectById(id);
    if (user != null) {
        redisTemplate.set(key, user, 1, TimeUnit.HOURS);  // 必须设过期时间
    }
    return user;
}
 
// 写
public void updateUser(User user) {
    userMapper.updateById(user);           // 1. 先更新数据库
    redisTemplate.delete("user:" + user.getId());  // 2. 再删除缓存
}

1.3 Key 设计

【Requirement】Key 格式：<业务域>:<实体>:<ID>，全部小写，冒号分隔，必须设 TTL。

user:profile:123
order:detail:ORD20240101
product:stock:456

Key 长度 ≤ 128 字节
Value 大小 ≤ 10KB（大对象拆分或压缩）
必须设过期时间，禁止永久缓存（PERSIST）

1.4 过期时间设置

数据类型	TTL	说明
用户信息	1 小时	允许短暂不一致
商品库存	不缓存 / 短 TTL	强一致，谨慎缓存
配置数据	24 小时	变更频率低
热点数据	5 分钟	防雪崩加随机值
会话信息	30 分钟	滑动过期

【Caution】为防止缓存雪崩，TTL 加随机偏移：

// TTL = 基础时间 + 随机偏移（0-300秒）
long ttl = 3600 + new Random().nextInt(300);
redisTemplate.set(key, value, ttl, TimeUnit.SECONDS);

一致性方案

【Standard】缓存与数据库一致性采用「先更新数据库，再删除缓存」+「延迟双删」+「订阅 binlog 补偿」策略。

2.1 经典问题：先删缓存还是先更库

顺序	问题	结论
先删缓存，再更库	并发读会回填旧值，导致脏数据	不推荐
先更库，再删缓存	并发读可能在删缓存前读到旧值并回填	推荐 + 补偿

2.2 延迟双删

更新数据库后，延迟一段时间再次删除缓存，消除并发回填的旧值：

public void updateUser(User user) {
    userMapper.updateById(user);              // 1. 更新数据库
    redisTemplate.delete("user:" + user.getId());  // 2. 删缓存
    // 3. 延迟 500ms 再删一次
    scheduledExecutor.schedule(
        () -> redisTemplate.delete("user:" + user.getId()),
        500, TimeUnit.MILLISECONDS
    );
}

2.3 binlog 订阅补偿（终极方案）

通过 Canal 订阅 MySQL binlog，异步删除缓存，保证最终一致：

1. 应用更新数据库
2. Canal 订阅到 binlog 变更
3. 解析变更数据，删除对应缓存 Key
4. 失败重试，保证最终删除

2.4 强一致场景

【Caution】资金、库存等强一致场景禁止使用缓存，直接读数据库。

如必须缓存，采用以下策略：

写操作加分布式锁，串行化
使用 Redisson 可重入锁
锁释放时主动刷新缓存

public void deductStock(Long productId, int count) {
    RLock lock = redisson.getLock("stock:" + productId);
    lock.lock(10, TimeUnit.SECONDS);
    try {
        stockMapper.deduct(productId, count);
        redisTemplate.delete("stock:" + productId);
    } finally {
        lock.unlock();
    }
}

缓存穿透/击穿/雪崩

【Requirement】必须防范缓存穿透、击穿、雪崩三大问题，建立布隆过滤器、互斥锁、随机 TTL 等防护机制。

3.1 缓存穿透

问题：查询不存在的数据，缓存和数据库都没有，每次都打到数据库。

解决方案：

缓存空值：查询结果为 null 也缓存，设短 TTL（60 秒）
布隆过滤器：提前加载所有合法 Key，不存在直接拦截

// 方案1：缓存空值
public User getUser(Long id) {
    String key = "user:" + id;
    User user = redisTemplate.get(key);
    if (user != null) {
        return user == NULL_USER ? null : user;  // 区分空值
    }
    user = userMapper.selectById(id);
    if (user == null) {
        redisTemplate.set(key, NULL_USER, 60, TimeUnit.SECONDS);  // 缓存空值
    } else {
        redisTemplate.set(key, user, 1, TimeUnit.HOURS);
    }
    return user;
}
 
// 方案2：布隆过滤器
private BloomFilter<Long> bloomFilter = BloomFilter.create(Funnels.longFunnel(), 1000000);
 
public User getUser(Long id) {
    if (!bloomFilter.mightContain(id)) {
        return null;  // 不存在直接返回
    }
    // 正常查缓存和数据库
}

3.2 缓存击穿

问题：热点 Key 过期瞬间，大量并发请求打到数据库。

解决方案：互斥锁（只有一个请求查库，其他等待）

public User getHotUser(Long id) {
    String key = "user:" + id;
    User user = redisTemplate.get(key);
    if (user != null) return user;
 
    String lockKey = "lock:user:" + id;
    RLock lock = redisson.getLock(lockKey);
    try {
        if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
            // 双重检查
            user = redisTemplate.get(key);
            if (user != null) return user;
            user = userMapper.selectById(id);
            redisTemplate.set(key, user, 1, TimeUnit.HOURS);
        } else {
            // 未获取锁，短暂休眠后重试
            Thread.sleep(50);
            return getHotUser(id);
        }
    } finally {
        if (lock.isHeldByCurrentThread()) lock.unlock();
    }
    return user;
}

3.3 缓存雪崩

问题：大量 Key 同时过期，或 Redis 宕机，请求全部打到数据库。

解决方案：

TTL 加随机值：避免同时过期
集群部署：Redis 哨兵 / Cluster，高可用
限流降级：数据库压力过大时限流
多级缓存：本地缓存（Caffeine）+ Redis

// TTL 随机化
long baseTtl = 3600;
long randomTtl = new Random().nextInt(600);  // 0-600 秒随机
redisTemplate.set(key, value, baseTtl + randomTtl, TimeUnit.SECONDS);
 
// 多级缓存
public User getUser(Long id) {
    String key = "user:" + id;
    User user = localCache.get(key);  // 1. 本地缓存
    if (user != null) return user;
    user = redisTemplate.get(key);    // 2. Redis
    if (user != null) {
        localCache.put(key, user);
        return user;
    }
    user = userMapper.selectById(id); // 3. 数据库
    if (user != null) {
        redisTemplate.set(key, user, 1, TimeUnit.HOURS);
        localCache.put(key, user);
    }
    return user;
}

分布式锁

【Standard】分布式锁使用 Redisson 实现，必须设置超时时间和自动续期；锁粒度要小，持有时间要短。

4.1 锁的实现要求

要求	说明
互斥性	同一时间只有一个客户端持有
超时释放	防止死锁，必须设过期时间
可重入	同一线程可重复获取
自动续期	业务未执行完自动续期
正确释放	只能释放自己持有的锁

4.2 Redisson 分布式锁

// 可重入锁 + 自动续期
RLock lock = redisson.getLock("order:create:" + userId);
try {
    // 等待3秒，持有后自动续期（默认30秒，每10秒续期）
    if (lock.tryLock(3, -1, TimeUnit.SECONDS)) {
        // 业务逻辑
        createOrder(userId);
    }
} finally {
    if (lock.isHeldByCurrentThread()) {
        lock.unlock();
    }
}

4.3 禁止使用的锁实现

【Caution】禁止使用 SETNX + 手动删除的简单实现，存在以下问题：

业务未执行完锁过期，其他线程获取锁
释放锁时可能删除其他线程的锁

// 错误：简单 SETNX 实现
Boolean locked = redisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
if (locked) {
    try { ... } finally {
        redisTemplate.delete(key);  // 可能误删其他线程的锁
    }
}

4.4 锁粒度

锁的 Key 尽量精确，避免大锁
锁内只包含必要操作，减少持有时间
禁止在锁内执行远程调用

// 正确：细粒度锁，按用户加锁
RLock lock = redisson.getLock("order:create:" + userId);
 
// 错误：全局锁，所有用户串行
RLock lock = redisson.getLock("order:create");

监控

【Standard】缓存监控必须覆盖：命中率、Key 数量、内存使用、慢查询、连接数；命中率 < 80% 需告警。

5.1 核心监控指标

指标	说明	告警阈值
缓存命中率	命中次数 / 总请求数	< 80% 告警
Key 数量	当前缓存 Key 总数	突增 > 50%
内存使用率	used_memory / maxmemory	> 80%
慢查询数	执行时间 > 1ms 的命令数	> 10 次/分钟
连接数	当前客户端连接数	> 80% maxclients
QPS	每秒命令数	环比突增 > 100%

5.2 监控命令

# Redis 状态
redis-cli info stats        # 统计信息
redis-cli info memory       # 内存使用
redis-cli info clients      # 客户端连接
redis-cli info keyspace     # 数据库 Key 信息
redis-cli slowlog get 10    # 慢查询

5.3 慢查询处理

# 查看慢查询
redis-cli SLOWLOG GET 10
 
# 配置慢查询阈值（微秒）
redis-cli CONFIG SET slowlog-log-slower-than 1000
redis-cli CONFIG SET slowlog-max-len 128

5.4 大 Key 治理

【Caution】单个 Key 的 Value > 10KB 为大 Key，必须拆分或优化。

# 扫描大 Key
redis-cli --bigkeys
 
# 或使用 redis-rdb-tools
rdb --command memory --bytes 10240 dump.rdb > big_keys.txt

大 Key 处理：

String 大 Value：拆分为多个 Key 或压缩存储
Hash/List/Set 元素过多：按业务维度拆分
禁止使用 KEYS *，使用 SCAN 代替

// 正确：使用 SCAN 遍历
ScanOptions options = ScanOptions.scanOptions().match("user:*").count(100).build();
try (Cursor<String> cursor = redisTemplate.scan(options)) {
    while (cursor.hasNext()) {
        String key = cursor.next();
        // 处理
    }
}
 
// 错误：KEYS * 会阻塞 Redis
Set<String> keys = redisTemplate.keys("user:*");