做后端开发这几年Redis 相关的“三兄弟”问题——穿透、击穿、雪崩几乎是每个团队都要正面撞上一次的。尤其是缓存击穿平时看着温温吞吞一到活动高峰热点 key 刚过期那一瞬间几万请求同时打到数据库上CPU 直接拉满慢查询一条接一条整个服务像被人掐住了脖子。今天想聊的这套方案就是专门针对缓存击穿的“双重判定锁”。很多文章叫它双重检测锁也有叫双检锁的本质上是同一套思路用 Redis 的互斥锁配合“加锁后再次检查缓存”的二次判定把并发打到数据库的请求压到只剩一个。这套方案的适用范围很明确缓存里是热点数据更新不频繁但读取量极大业务上可以接受短时间内的锁等待。比如商品详情、活动配置、用户画像这类场景都非常适合。如果你正在用 SpringBoot 整合 Redis 做缓存治理或者面试前想搞懂 Redis 分布式锁到底怎么落地这篇内容应该能给你一个可以直接抄作业的答案。1. 缓存击穿问题剖析1.1 穿透、击穿、雪崩到底怎么区分很多新手把这三个概念混在一起实际排查问题时才发现方向完全错了。我习惯用一句话区分穿透是查了一个缓存和数据库里都不存在的数据击穿是缓存里本来有但恰好在某个时间点过期了雪崩是大量 key 在同一时间段集体过期。穿透的核心问题是“不存在”所以布隆过滤器、缓存空值是主流解法击穿的核心问题是“热点 key 过期瞬间的高并发”所以互斥锁、逻辑过期是主流解法雪崩的核心问题是“过期时间设置不合理或宕机”所以过期时间加随机值、多级缓存、高可用是主流解法。三者的表现也很不一样。穿透的攻击特征很明显请求的 key 千奇百怪Redis 命中率骤降数据库出现大量空查询击穿则集中在某一个或某几个 key 上数据库的 QPS 曲线会在 key 过期后出现一个尖锐的峰值雪崩是整个 Redis 缓存的命中率断崖式下跌数据库所有表都感受到压力。我之前接手过一个线上事故现象是单条 SQL 的慢查询飙升排查了半天发现就是商品详情页的某个热点 key 设置了固定的 30 分钟过期一到整点大家同时来刷就把数据库打穿了。这种事故有个特点Redis 本身没问题数据库连接池却被打满了很多不懂的人会先去优化 SQL方向完全错了。1.2 为什么击穿最难搞穿透和雪崩都有比较成熟的“无脑”解法击穿麻烦在于“热点 key”三个字。首先你很难提前预判哪个 key 会突然变成热点等线上告警响了再处理流量早就把数据库压垮了其次热点 key 的并发量级往往远超预期一个 key 承载整个集群的读流量一旦失效压力是瞬间叠加的。网上常见几种方案各有各的坑。缓存空值能解决穿透但对击穿几乎无效因为击穿场景下 key 本来就有值只是过期了空值缓存根本不会生效。布隆过滤器也一样它是拦截“不存在”的 key对“存在但过期”的 key 毫无办法。所以击穿场景几乎只能靠锁或者“提前续期”的思路来兜底。还有一个容易被忽略的点击穿和穿透经常会叠加出现。比如某个 key 查询数据库时确实没有数据你又没有做空值缓存那么这个 key 就会在缓存层和数据库层反复穿透而这种 key 一旦被打上“热点”标签就会演变成击穿问题。所以成熟的方案里双重判定锁往往会搭配空值缓存一起使用一层防击穿一层防穿透两个问题一起解决。2. 双重判定锁的核心原理与设计取舍2.1 “双重判定”到底指哪两次判定双重判定锁这个名字我第一次听的时候也觉得有点绕其实看代码就一目了然。它借鉴了 Java 单例模式里双重检测锁DCL的思路只不过把对象实例换成了缓存中的 key。第一次判定发生在加锁之前请求进来后先查一次 Redis如果缓存命中了直接返回锁都不用碰如果没命中说明缓存可能过期了这时候才去尝试获取互斥锁。第二次判定发生在成功拿到锁之后再次查一次 Redis看看缓存是不是已经被其他线程重建好了。为什么要多查这一次因为在高并发场景下可能出现“多个线程同时发现缓存未命中同时尝试加锁但只有一个线程成功拿到锁”的情况。如果拿到锁的线程去数据库查询并回填了缓存那么其他正在等待锁的线程在拿到锁之后其实已经没有必要再去查询数据库了。如果少了第二次判定所有线程拿锁后都会重复查库锁就形同虚设。这个细节我见过太多人写错。很多人以为加了锁就万事大吉结果压测一打数据库 QPS 还是冲上了天。原因就是没有做二次判定每个线程拿到锁后都老老实实去查了一次库虽然并发被锁串行化了但重复查询的问题依然存在。2.2 两种主流变体缓存空值与逻辑过期双重判定锁在落地的时候会根据业务对一致性和性能的要求演变成两种风格。第一种是“缓存空值 互斥锁”。请求过来先查 Redis如果命中且值不为空直接返回如果值为空加锁后再查一次 Redis若仍然为空则查数据库。数据库查到了就回填缓存查不到就回填一个特殊空值并设置短过期时间。这种方案实现最简单数据一致性最好但有一个代价热点 key 过期瞬间拿不到锁的请求需要短暂等待也就是“牺牲一点响应时间换数据库安全”。第二种是“逻辑过期 互斥锁”。这种方案下Redis 缓存的值是一个包装对象里面除了业务数据还带一个逻辑上的过期时间。请求读缓存时如果发现逻辑时间还没到直接返回数据如果发现逻辑时间已经到了会尝试加锁加锁成功后再查一次缓存如果还是过期的状态就另起一个线程去重建缓存当前线程则先返回旧值。这种方案的好处是响应速度快请求基本不会阻塞缺点是数据一致性偏弱极端情况下用户可能看到很短一段时间的旧数据。两种方案没有绝对的好坏。我个人的选择标准很简单如果业务对一致性敏感比如库存、金额相关用互斥锁方案如果是商品标题、描述这种允许短暂不一致的内容用逻辑过期方案用户体验更好。但无论哪种双重判定的核心逻辑都跑不掉加锁后必须二次检查否则并发控制就是空谈。2.3 和普通 Redis 分布式锁相比差在哪有朋友会问这不就是一个分布式锁吗为什么单独起个名字叫双重判定锁区别在于一般讨论 Redis 分布式锁时重点是“互斥”和“可靠性”比如防止死锁、防止误删、保证原子性而双重判定锁的重点是“在互斥的基础上减少不必要的数据库访问”。你可以这么理解普通分布式锁解决的是“并发下同一资源只能被一个线程操作”的问题双重判定锁解决的是“缓存失效瞬间大量请求穿透到数据库”的问题。后者是在前者之上叠加了“缓存二次检查”这个缓存治理专属的步骤。所以你单独用 Redisson 的 RLock 也能做互斥但如果没有二次判定锁的性能收益会大打折扣。另外双重判定锁的锁粒度没必要做得很重。普通分布式锁可能要考虑可重入、看门狗续期、公平锁这些特性双重判定锁通常只需要最基础的 SETNX 就能满足需求。因为锁的持有时间极短就是一次数据库查询加缓存回填的时间只要确保不会死锁业务就能跑得很稳。3. SpringBoot 完整落地实现3.1 环境准备与 RedisTemplate 配置先说环境。我用的是 SpringBoot 2.7 Redis 7.0连接工具用的 Redis Desktop Manager平时排查 key 很方便。如果你是在 Windows 本机开发Redis 官方没有 Windows 版本可以下载微软维护的发行版或者用 Docker 起一个 redis 镜像我个人更推荐 Docker 方式和 Linux 环境行为一致不容易踩坑。引入依赖时SpringBoot 的spring-boot-starter-data-redis就够了。这里要注意一个序列化问题。很多新手用 RedisTemplate 直接存字符串结果 key 变成\xac\xed\x00\x05t\x00...这种乱码就是因为默认的 JdkSerializationRedisSerializer。在缓存场景下我建议直接用 StringRedisTemplate或者手动把 RedisTemplate 的 key 和 value 序列化器都改成 StringRedisSerializer。实操时还有一个坑RedisTemplate 的setIfAbsent方法在不同版本里签名不一样低版本叫setIfAbsent(K key, V value)高版本重载了setIfAbsent(K key, V value, long timeout, TimeUnit unit)。如果你想要加锁的同时设置过期时间一定要用带过期参数的版本否则还得额外调一次expire方法两次操作不具备原子性锁就存在永久不释放的风险。3.2 用 SETNX 实现双重判定锁完整代码下面这份代码是我比较常用的互斥锁版本核心流程完整可以直接跑。为了清晰我把加锁、解锁、缓存读取都单独拆开。首先定义一个简单的数据对象模拟商品public class Product { private Long id; private String name; private BigDecimal price; // getter / setter 省略 }然后写核心的服务方法Service public class ProductService { private static final String CACHE_KEY_PREFIX product:; private static final String LOCK_KEY_PREFIX lock:product:; private static final long CACHE_TTL_SECONDS 600; private static final long LOCK_EXPIRE_SECONDS 5; private static final long EMPTY_VALUE_TTL_SECONDS 60; Autowired private StringRedisTemplate stringRedisTemplate; Autowired private ObjectMapper objectMapper; public Product getProductById(Long id) { String cacheKey CACHE_KEY_PREFIX id; String lockKey LOCK_KEY_PREFIX id; // 第一次判定直接查缓存 String cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { // 处理缓存空值的情况 if (NULL_VALUE.equals(cached)) { return null; } return deserialize(cached); } // 尝试加锁只在获取到锁的线程里执行重建逻辑 String requestId UUID.randomUUID().toString(); Boolean locked stringRedisTemplate.opsForValue() .setIfAbsent(lockKey, requestId, LOCK_EXPIRE_SECONDS, TimeUnit.SECONDS); if (Boolean.TRUE.equals(locked)) { try { // 第二次判定拿到锁后再查一次缓存 cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { if (NULL_VALUE.equals(cached)) { return null; } return deserialize(cached); } // 缓存仍然不存在才去数据库查询 Product product queryFromDatabase(id); if (product null) { // 缓存空值防止缓存穿透 stringRedisTemplate.opsForValue() .set(cacheKey, NULL_VALUE, EMPTY_VALUE_TTL_SECONDS, TimeUnit.SECONDS); return null; } stringRedisTemplate.opsForValue() .set(cacheKey, serialize(product), CACHE_TTL_SECONDS, TimeUnit.SECONDS); return product; } finally { // 释放锁时必须校验 value且保证原子性 releaseLock(lockKey, requestId); } } // 没拿到锁的线程自旋重试 for (int i 0; i 3; i) { try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { if (NULL_VALUE.equals(cached)) { return null; } return deserialize(cached); } } // 重试后仍未拿到缓存回源数据库兜底策略避免请求失败 Product product queryFromDatabase(id); return product; } private void releaseLock(String lockKey, String requestId) { // 使用 Lua 脚本保证“校验 value 删除 key”的原子性 String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end; DefaultRedisScriptLong redisScript new DefaultRedisScript(script, Long.class); stringRedisTemplate.execute(redisScript, Collections.singletonList(lockKey), requestId); } private Product queryFromDatabase(Long id) { // 模拟慢查询 try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } // 实际场景从 Mapper 查库 return new Product(id, 测试商品- id, new BigDecimal(99.00)); } private String serialize(Product product) { try { return objectMapper.writeValueAsString(product); } catch (JsonProcessingException e) { throw new RuntimeException(序列化失败, e); } } private Product deserialize(String json) { try { return objectMapper.readValue(json, Product.class); } catch (JsonProcessingException e) { throw new RuntimeException(反序列化失败, e); } } }这段代码有几个细节值得展开讲。requestId是每个线程加锁时生成的唯一标识释放锁时先判断 Redis 里的 value 是否等于当前线程的 requestId再执行删除。这个判断必须放在 Lua 脚本里做因为“判断 删除”不是原子操作的话会出现一个线程把另一个线程刚获取的锁误删的情况。我见过不少人用两步操作写释放锁压测时全靠运气没出事但迟早会出问题。自旋重试的Thread.sleep(50)需要根据场景调整。50 毫秒是比较保守的值如果你对响应时间不敏感可以放到 100 甚至 200 毫秒如果业务要求高并发下的低延迟50 毫秒内的等待很多人是可以接受的。注意不要用while(true)无限自旋万一数据库挂了所有请求都会堆积在锁等待上直接把连接池拖垮。最后那个“兜底策略”是个人习惯。当线程重试几次后还是没读到缓存说明可能出现了异常情况比如 Redis 服务不稳定或者持有锁的线程还没回填。这时候直接查一次数据库兜底至少保证请求不失败。代价是极端情况下数据库会多承受部分查询压力但比接口直接报错要好得多。3.3 逻辑过期方案的核心实现如果你的业务对响应时间敏感不想让任何请求阻塞等待逻辑过期方案会更合适。先定义一个包装类public class RedisDataT { private T data; private LocalDateTime expireTime; // getter / setter 省略 }然后实现读取逻辑public Product getProductWithLogicalExpire(Long id) { String cacheKey CACHE_KEY_PREFIX id; String cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached null) { // 缓存不存在不阻塞直接查库重建也可用互斥锁兜底 return queryFromDatabase(id); } RedisDataProduct redisData deserializeRedisData(cached); // 逻辑时间未到直接返回旧值 if (redisData.getExpireTime().isAfter(LocalDateTime.now())) { return redisData.getData(); } // 逻辑时间到期尝试获取锁 String lockKey LOCK_KEY_PREFIX id; String requestId UUID.randomUUID().toString(); Boolean locked stringRedisTemplate.opsForValue() .setIfAbsent(lockKey, requestId, LOCK_EXPIRE_SECONDS, TimeUnit.SECONDS); if (Boolean.TRUE.equals(locked)) { try { // 双重判定拿到锁后再查一次缓存 cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { RedisDataProduct latestData deserializeRedisData(cached); if (latestData.getExpireTime().isAfter(LocalDateTime.now())) { return latestData.getData(); } } // 另起线程重建缓存当前线程先返回旧数据 Product product queryFromDatabase(id); RedisDataProduct newData new RedisData(); newData.setData(product); newData.setExpireTime(LocalDateTime.now().plusSeconds(600)); stringRedisTemplate.opsForValue().set(cacheKey, serializeRedisData(newData), 12, TimeUnit.HOURS); return product; } finally { releaseLock(lockKey, requestId); } } // 获取不到锁直接返回当前缓存里的旧数据 return redisData.getData(); }这种方案最大的特点是拿不到锁的线程不需要等待直接返回旧数据所以接口响应时间非常稳定。但需要注意两个问题一是缓存里的物理过期时间要设得比逻辑过期时间长很多防止逻辑过期还没到物理 key 先被 Redis 淘汰了二是如果业务数据更新频繁逻辑过期方案会导致用户读到旧值的时间窗口变长需要业务方能够接受。3.4 锁的过期时间怎么定锁的过期时间是最容易拍脑袋又最容易出事的地方。设短了数据库查询慢一点锁就自动失效其他线程趁虚而入锁形同虚设设长了万一持有锁的线程发生异常其他请求会长时间阻塞等待重试的线程会越积越多。常规做法是预估“数据库查询 缓存回填”的最长耗时然后在这个基础上乘一个 3 到 5 倍的保险系数。比如数据库查询加上网络耗时平均 50 毫秒极端情况 1 秒内能完成那把锁的过期时间设为 5 秒已经足够。千万不要在锁里面执行 RPC 调用或者大批量数据操作否则锁注定要超时。这里可以套一个简单的公式锁过期时间 业务预估最大耗时 × 3然后向上取整到一个常见的秒数。如果业务耗时极不稳定建议升级方案使用 Redisson 这类带看门狗自动续期的分布式锁而不是死磕 SETNX。4. 实际操作中的问题与排查技巧实录4.1 锁误删问题没有原子性校验的代价有一次压测我故意把释放锁的逻辑写成了“先 get 判断 value再 del”结果在高并发下真的复现了误删线程 A 拿到锁后处理时间较长锁过期了线程 B 拿到锁并设置了自己的 value线程 A 处理完准备释放锁它读到 Redis 里的 value 已经不是自己的 requestId但因为在“判断”和“删除”之间发生了线程切换最终它把线程 B 的锁给删了。线程 C 趁虚而入再次拿到锁并发控制彻底失效。之后我所有释放锁的操作都强制要求用 Lua 脚本或者直接用 Redisson 的unlock()方法。判断和删除必须是一个原子操作没有任何商量余地。4.2 获取锁失败后重试策略不当自旋重试如果不加次数限制在热点 key 失效的瞬间会形成死循环风暴。比如 1 万个请求同时进来1 个请求拿到锁去查库剩下 9999 个请求全部在while(true)里循环读缓存对 Redis 的读压力本身就变成了一种二次冲击。更糟的是如果数据库查询时间较长这些自旋请求会把 Redis 连接池耗尽导致整个应用不可用。我的办法是限制自旋次数并且在每次自旋后 sleep 一个很小的随机时间避免所有请求在同一个时间点同时重试。你可以把 sleep 时间设置成 30 到 80 毫秒之间的随机值效果比固定 50 毫秒要好很多因为固定值容易造成“惊群”。4.3 空值缓存引发的数据短暂不一致缓存空值能防止穿透但也会带来一个副作用如果数据库里新插入了一条数据空值缓存还没过期用户会一直读到空结果。常规做法是把空值缓存的过期时间设得很短比如 30 到 60 秒同时配合主动删除业务侧插入数据时顺便把对应的空值缓存删掉。如果你使用 Redis 的发布订阅或者 Canal 同步数据变更也可以做到更实时的缓存更新。4.4 问题排查速查表现象可能原因排查方向缓存击穿仍然发生加锁后没有二次判定检查拿到锁之后是否重新查询缓存锁一直不释放请求大量堆积锁过期时间设置过长或释放锁代码没有在 finally 中执行检查 finally 块确认异常情况下锁也会释放锁被误删释放锁时未校验 value或校验与删除非原子操作改用 Lua 脚本释放锁自旋请求打爆 Redis无限循环重试增加重试次数上限sleep 加随机值数据库短暂出现压垮性查询空值缓存时间过长缓存删除后旧值未失效缩短空值 TTL同步删除缓存主从切换后锁失效锁写入了旧主节点考虑 Redisson RedLock或对锁追加 Watchdog 续期5. 延伸从单体锁到 Redis 生产环境可靠性5.1 单机锁到分布式锁的演进双重判定锁用到 Redis本质上是因为多个应用节点之间需要共享同一个互斥状态。如果你的服务只有单节点用 JVM 内部的synchronized或者ReentrantLock就能完成互斥根本不需要引入 Redis。但现在的业务系统基本都是多实例部署请求经过负载均衡分发到不同节点JVM 锁只能锁住当前进程跨节点就失效了。这也是为什么 Redis 分布式锁在缓存治理中如此重要。SETNX 的语义很简单只有当 key 不存在时才能设置成功谁能成功设置谁就获得了锁。这个操作天然就是跨节点共享的任何一个应用节点执行setIfAbsent(lockKey, requestId)都遵循同一套规则所以可以实现全局互斥。不过要记住一个长期被争论的点普通 Redis 主从架构下如果 master 节点宕机锁数据还没来得及同步到 slave新的 master 不会有这把锁的记录其他节点就能重复加锁这就是经典的锁丢失问题。官方给的建议是使用 RedLock但 RedLock 本身也有争议生产环境是否采用取决于你对锁的敏感度。对于大多数缓存击穿场景锁丢失导致的后果只是多几个请求穿透到数据库影响可控所以我在大部分项目里都没上 RedLock够用就行。5.2 部署形态对锁可靠性的影响开发环境我用 Docker 直接跑 redis 镜像一条命令就能起来反正不用考虑数据持久化。生产环境就需要重视部署形态了。如果是主从哨兵模式Redis Sentinel 负责故障自动切换但切换过程存在短暂的不可用窗口双重判定锁在这个窗口内可能不可用。如果是 Redis Cluster 集群模式锁 key 会被哈希到某个 slot加锁操作只会落在对应的主节点上整体可用性更高但仍然要注意主从切换导致的锁丢失。另一个容易忽略的点是 Redis 持久化。锁 key 虽然生命周期很短但如果在没开启持久化的情况下 Redis 重启所有锁都会丢失。曾有人问过我Redis 重启后锁全丢了是不是双重判定锁就废了我的回答是如果 Redis 已经重启缓存大概率也没了这时候系统要处理的是缓存雪崩而不是单一 key 的击穿双重判定锁本来就该配合缓存预热、空值缓存、多级缓存一起使用单靠一把锁扛不住这种极端场景。5.3 热点场景下的落地建议以商品详情页为例我见过的比较稳的搭配是一级本地缓存Caffeine挡掉大部分读流量二级 Redis 缓存存商品数据三级数据库兜底。热点 key 的过期时间在基础值上加上随机偏移量避免所有商品同时过期形成雪崩。本地缓存失效后请求走到 RedisRedis 失效后再走双重判定锁回源数据库。秒杀场景又是另一套玩法。秒杀商品的库存数据对一致性要求极高通常不会让请求直接穿透到数据库而是把库存预加载到 Redis 里用 Redis 的原子命令比如decr或 Lua 脚本扣减库存扣减成功后再异步落库。这时候双重判定锁用不上的原因很简单库存不是“过期后回填”的缓存数据而是实时变更的数据源锁的作用是保护数据库不被并发写击穿方向完全不同。所以我在实操中的建议是不要试图用一个方案解决所有缓存问题。双重判定锁是缓存击穿场景下的“银弹”但它只解决“热点 key 过期瞬间的并发读穿透”这一个问题。搭配缓存空值解决穿透、搭配随机过期时间解决雪崩、搭配本地缓存提升性能组合起来才是完整的缓存治理方案。6. 我踩过的坑和给你的一些建议做这个方案从前到后踩了不少坑最值得说的有三个。第一个是二次判定这个是最容易犯的错误代码写完一眼扫过去觉得没问题压测时才发现数据库压力没有预期下降排查半天才意识到拿到锁之后没有再查一次缓存。第二个是释放锁的原子性不用 Lua 脚本之前我总觉得自己不会遇到误删直到亲手复现了一次才彻底老实了。第三个是锁过期时间的设定一开始我给了一个看起来很长的 30 秒想着肯定够用结果有一次数据库慢查询真的把锁给撑过期了后面所有请求全部穿到了数据库那一次真的把我吓出一身冷汗。如果你准备在项目里落地双重判定锁我建议先做一个最小验证写一个模拟热点 key 的接口用 JMeter 开 1000 个并发线程同时请求观察 Redis 的命中率和数据库的 QPS 曲线。这个实验能直观地看到双重判定的效果也能帮你调整锁的过期时间和重试策略。等到参数调稳了再推广到核心业务风险会小很多。最后再分享一个小技巧不要在锁里面做“先查数据库再调外部接口再更新缓存”这种长链路操作。双重判定锁的正确姿势是锁内只做最必要的操作链路越短锁持有的时间越短系统的并发能力就越高。如果确实需要做长链路建议把锁拆成多个短锁或者直接用异步更新的方式让请求先拿旧数据返回后台再慢慢重建缓存。这也是逻辑过期方案在互联网大厂里如此流行的原因取舍之间全是业务方的真实诉求。
Redis缓存击穿解决方案：双重判定锁原理与SpringBoot实践
做后端开发这几年Redis 相关的“三兄弟”问题——穿透、击穿、雪崩几乎是每个团队都要正面撞上一次的。尤其是缓存击穿平时看着温温吞吞一到活动高峰热点 key 刚过期那一瞬间几万请求同时打到数据库上CPU 直接拉满慢查询一条接一条整个服务像被人掐住了脖子。今天想聊的这套方案就是专门针对缓存击穿的“双重判定锁”。很多文章叫它双重检测锁也有叫双检锁的本质上是同一套思路用 Redis 的互斥锁配合“加锁后再次检查缓存”的二次判定把并发打到数据库的请求压到只剩一个。这套方案的适用范围很明确缓存里是热点数据更新不频繁但读取量极大业务上可以接受短时间内的锁等待。比如商品详情、活动配置、用户画像这类场景都非常适合。如果你正在用 SpringBoot 整合 Redis 做缓存治理或者面试前想搞懂 Redis 分布式锁到底怎么落地这篇内容应该能给你一个可以直接抄作业的答案。1. 缓存击穿问题剖析1.1 穿透、击穿、雪崩到底怎么区分很多新手把这三个概念混在一起实际排查问题时才发现方向完全错了。我习惯用一句话区分穿透是查了一个缓存和数据库里都不存在的数据击穿是缓存里本来有但恰好在某个时间点过期了雪崩是大量 key 在同一时间段集体过期。穿透的核心问题是“不存在”所以布隆过滤器、缓存空值是主流解法击穿的核心问题是“热点 key 过期瞬间的高并发”所以互斥锁、逻辑过期是主流解法雪崩的核心问题是“过期时间设置不合理或宕机”所以过期时间加随机值、多级缓存、高可用是主流解法。三者的表现也很不一样。穿透的攻击特征很明显请求的 key 千奇百怪Redis 命中率骤降数据库出现大量空查询击穿则集中在某一个或某几个 key 上数据库的 QPS 曲线会在 key 过期后出现一个尖锐的峰值雪崩是整个 Redis 缓存的命中率断崖式下跌数据库所有表都感受到压力。我之前接手过一个线上事故现象是单条 SQL 的慢查询飙升排查了半天发现就是商品详情页的某个热点 key 设置了固定的 30 分钟过期一到整点大家同时来刷就把数据库打穿了。这种事故有个特点Redis 本身没问题数据库连接池却被打满了很多不懂的人会先去优化 SQL方向完全错了。1.2 为什么击穿最难搞穿透和雪崩都有比较成熟的“无脑”解法击穿麻烦在于“热点 key”三个字。首先你很难提前预判哪个 key 会突然变成热点等线上告警响了再处理流量早就把数据库压垮了其次热点 key 的并发量级往往远超预期一个 key 承载整个集群的读流量一旦失效压力是瞬间叠加的。网上常见几种方案各有各的坑。缓存空值能解决穿透但对击穿几乎无效因为击穿场景下 key 本来就有值只是过期了空值缓存根本不会生效。布隆过滤器也一样它是拦截“不存在”的 key对“存在但过期”的 key 毫无办法。所以击穿场景几乎只能靠锁或者“提前续期”的思路来兜底。还有一个容易被忽略的点击穿和穿透经常会叠加出现。比如某个 key 查询数据库时确实没有数据你又没有做空值缓存那么这个 key 就会在缓存层和数据库层反复穿透而这种 key 一旦被打上“热点”标签就会演变成击穿问题。所以成熟的方案里双重判定锁往往会搭配空值缓存一起使用一层防击穿一层防穿透两个问题一起解决。2. 双重判定锁的核心原理与设计取舍2.1 “双重判定”到底指哪两次判定双重判定锁这个名字我第一次听的时候也觉得有点绕其实看代码就一目了然。它借鉴了 Java 单例模式里双重检测锁DCL的思路只不过把对象实例换成了缓存中的 key。第一次判定发生在加锁之前请求进来后先查一次 Redis如果缓存命中了直接返回锁都不用碰如果没命中说明缓存可能过期了这时候才去尝试获取互斥锁。第二次判定发生在成功拿到锁之后再次查一次 Redis看看缓存是不是已经被其他线程重建好了。为什么要多查这一次因为在高并发场景下可能出现“多个线程同时发现缓存未命中同时尝试加锁但只有一个线程成功拿到锁”的情况。如果拿到锁的线程去数据库查询并回填了缓存那么其他正在等待锁的线程在拿到锁之后其实已经没有必要再去查询数据库了。如果少了第二次判定所有线程拿锁后都会重复查库锁就形同虚设。这个细节我见过太多人写错。很多人以为加了锁就万事大吉结果压测一打数据库 QPS 还是冲上了天。原因就是没有做二次判定每个线程拿到锁后都老老实实去查了一次库虽然并发被锁串行化了但重复查询的问题依然存在。2.2 两种主流变体缓存空值与逻辑过期双重判定锁在落地的时候会根据业务对一致性和性能的要求演变成两种风格。第一种是“缓存空值 互斥锁”。请求过来先查 Redis如果命中且值不为空直接返回如果值为空加锁后再查一次 Redis若仍然为空则查数据库。数据库查到了就回填缓存查不到就回填一个特殊空值并设置短过期时间。这种方案实现最简单数据一致性最好但有一个代价热点 key 过期瞬间拿不到锁的请求需要短暂等待也就是“牺牲一点响应时间换数据库安全”。第二种是“逻辑过期 互斥锁”。这种方案下Redis 缓存的值是一个包装对象里面除了业务数据还带一个逻辑上的过期时间。请求读缓存时如果发现逻辑时间还没到直接返回数据如果发现逻辑时间已经到了会尝试加锁加锁成功后再查一次缓存如果还是过期的状态就另起一个线程去重建缓存当前线程则先返回旧值。这种方案的好处是响应速度快请求基本不会阻塞缺点是数据一致性偏弱极端情况下用户可能看到很短一段时间的旧数据。两种方案没有绝对的好坏。我个人的选择标准很简单如果业务对一致性敏感比如库存、金额相关用互斥锁方案如果是商品标题、描述这种允许短暂不一致的内容用逻辑过期方案用户体验更好。但无论哪种双重判定的核心逻辑都跑不掉加锁后必须二次检查否则并发控制就是空谈。2.3 和普通 Redis 分布式锁相比差在哪有朋友会问这不就是一个分布式锁吗为什么单独起个名字叫双重判定锁区别在于一般讨论 Redis 分布式锁时重点是“互斥”和“可靠性”比如防止死锁、防止误删、保证原子性而双重判定锁的重点是“在互斥的基础上减少不必要的数据库访问”。你可以这么理解普通分布式锁解决的是“并发下同一资源只能被一个线程操作”的问题双重判定锁解决的是“缓存失效瞬间大量请求穿透到数据库”的问题。后者是在前者之上叠加了“缓存二次检查”这个缓存治理专属的步骤。所以你单独用 Redisson 的 RLock 也能做互斥但如果没有二次判定锁的性能收益会大打折扣。另外双重判定锁的锁粒度没必要做得很重。普通分布式锁可能要考虑可重入、看门狗续期、公平锁这些特性双重判定锁通常只需要最基础的 SETNX 就能满足需求。因为锁的持有时间极短就是一次数据库查询加缓存回填的时间只要确保不会死锁业务就能跑得很稳。3. SpringBoot 完整落地实现3.1 环境准备与 RedisTemplate 配置先说环境。我用的是 SpringBoot 2.7 Redis 7.0连接工具用的 Redis Desktop Manager平时排查 key 很方便。如果你是在 Windows 本机开发Redis 官方没有 Windows 版本可以下载微软维护的发行版或者用 Docker 起一个 redis 镜像我个人更推荐 Docker 方式和 Linux 环境行为一致不容易踩坑。引入依赖时SpringBoot 的spring-boot-starter-data-redis就够了。这里要注意一个序列化问题。很多新手用 RedisTemplate 直接存字符串结果 key 变成\xac\xed\x00\x05t\x00...这种乱码就是因为默认的 JdkSerializationRedisSerializer。在缓存场景下我建议直接用 StringRedisTemplate或者手动把 RedisTemplate 的 key 和 value 序列化器都改成 StringRedisSerializer。实操时还有一个坑RedisTemplate 的setIfAbsent方法在不同版本里签名不一样低版本叫setIfAbsent(K key, V value)高版本重载了setIfAbsent(K key, V value, long timeout, TimeUnit unit)。如果你想要加锁的同时设置过期时间一定要用带过期参数的版本否则还得额外调一次expire方法两次操作不具备原子性锁就存在永久不释放的风险。3.2 用 SETNX 实现双重判定锁完整代码下面这份代码是我比较常用的互斥锁版本核心流程完整可以直接跑。为了清晰我把加锁、解锁、缓存读取都单独拆开。首先定义一个简单的数据对象模拟商品public class Product { private Long id; private String name; private BigDecimal price; // getter / setter 省略 }然后写核心的服务方法Service public class ProductService { private static final String CACHE_KEY_PREFIX product:; private static final String LOCK_KEY_PREFIX lock:product:; private static final long CACHE_TTL_SECONDS 600; private static final long LOCK_EXPIRE_SECONDS 5; private static final long EMPTY_VALUE_TTL_SECONDS 60; Autowired private StringRedisTemplate stringRedisTemplate; Autowired private ObjectMapper objectMapper; public Product getProductById(Long id) { String cacheKey CACHE_KEY_PREFIX id; String lockKey LOCK_KEY_PREFIX id; // 第一次判定直接查缓存 String cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { // 处理缓存空值的情况 if (NULL_VALUE.equals(cached)) { return null; } return deserialize(cached); } // 尝试加锁只在获取到锁的线程里执行重建逻辑 String requestId UUID.randomUUID().toString(); Boolean locked stringRedisTemplate.opsForValue() .setIfAbsent(lockKey, requestId, LOCK_EXPIRE_SECONDS, TimeUnit.SECONDS); if (Boolean.TRUE.equals(locked)) { try { // 第二次判定拿到锁后再查一次缓存 cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { if (NULL_VALUE.equals(cached)) { return null; } return deserialize(cached); } // 缓存仍然不存在才去数据库查询 Product product queryFromDatabase(id); if (product null) { // 缓存空值防止缓存穿透 stringRedisTemplate.opsForValue() .set(cacheKey, NULL_VALUE, EMPTY_VALUE_TTL_SECONDS, TimeUnit.SECONDS); return null; } stringRedisTemplate.opsForValue() .set(cacheKey, serialize(product), CACHE_TTL_SECONDS, TimeUnit.SECONDS); return product; } finally { // 释放锁时必须校验 value且保证原子性 releaseLock(lockKey, requestId); } } // 没拿到锁的线程自旋重试 for (int i 0; i 3; i) { try { Thread.sleep(50); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { if (NULL_VALUE.equals(cached)) { return null; } return deserialize(cached); } } // 重试后仍未拿到缓存回源数据库兜底策略避免请求失败 Product product queryFromDatabase(id); return product; } private void releaseLock(String lockKey, String requestId) { // 使用 Lua 脚本保证“校验 value 删除 key”的原子性 String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end; DefaultRedisScriptLong redisScript new DefaultRedisScript(script, Long.class); stringRedisTemplate.execute(redisScript, Collections.singletonList(lockKey), requestId); } private Product queryFromDatabase(Long id) { // 模拟慢查询 try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } // 实际场景从 Mapper 查库 return new Product(id, 测试商品- id, new BigDecimal(99.00)); } private String serialize(Product product) { try { return objectMapper.writeValueAsString(product); } catch (JsonProcessingException e) { throw new RuntimeException(序列化失败, e); } } private Product deserialize(String json) { try { return objectMapper.readValue(json, Product.class); } catch (JsonProcessingException e) { throw new RuntimeException(反序列化失败, e); } } }这段代码有几个细节值得展开讲。requestId是每个线程加锁时生成的唯一标识释放锁时先判断 Redis 里的 value 是否等于当前线程的 requestId再执行删除。这个判断必须放在 Lua 脚本里做因为“判断 删除”不是原子操作的话会出现一个线程把另一个线程刚获取的锁误删的情况。我见过不少人用两步操作写释放锁压测时全靠运气没出事但迟早会出问题。自旋重试的Thread.sleep(50)需要根据场景调整。50 毫秒是比较保守的值如果你对响应时间不敏感可以放到 100 甚至 200 毫秒如果业务要求高并发下的低延迟50 毫秒内的等待很多人是可以接受的。注意不要用while(true)无限自旋万一数据库挂了所有请求都会堆积在锁等待上直接把连接池拖垮。最后那个“兜底策略”是个人习惯。当线程重试几次后还是没读到缓存说明可能出现了异常情况比如 Redis 服务不稳定或者持有锁的线程还没回填。这时候直接查一次数据库兜底至少保证请求不失败。代价是极端情况下数据库会多承受部分查询压力但比接口直接报错要好得多。3.3 逻辑过期方案的核心实现如果你的业务对响应时间敏感不想让任何请求阻塞等待逻辑过期方案会更合适。先定义一个包装类public class RedisDataT { private T data; private LocalDateTime expireTime; // getter / setter 省略 }然后实现读取逻辑public Product getProductWithLogicalExpire(Long id) { String cacheKey CACHE_KEY_PREFIX id; String cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached null) { // 缓存不存在不阻塞直接查库重建也可用互斥锁兜底 return queryFromDatabase(id); } RedisDataProduct redisData deserializeRedisData(cached); // 逻辑时间未到直接返回旧值 if (redisData.getExpireTime().isAfter(LocalDateTime.now())) { return redisData.getData(); } // 逻辑时间到期尝试获取锁 String lockKey LOCK_KEY_PREFIX id; String requestId UUID.randomUUID().toString(); Boolean locked stringRedisTemplate.opsForValue() .setIfAbsent(lockKey, requestId, LOCK_EXPIRE_SECONDS, TimeUnit.SECONDS); if (Boolean.TRUE.equals(locked)) { try { // 双重判定拿到锁后再查一次缓存 cached stringRedisTemplate.opsForValue().get(cacheKey); if (cached ! null) { RedisDataProduct latestData deserializeRedisData(cached); if (latestData.getExpireTime().isAfter(LocalDateTime.now())) { return latestData.getData(); } } // 另起线程重建缓存当前线程先返回旧数据 Product product queryFromDatabase(id); RedisDataProduct newData new RedisData(); newData.setData(product); newData.setExpireTime(LocalDateTime.now().plusSeconds(600)); stringRedisTemplate.opsForValue().set(cacheKey, serializeRedisData(newData), 12, TimeUnit.HOURS); return product; } finally { releaseLock(lockKey, requestId); } } // 获取不到锁直接返回当前缓存里的旧数据 return redisData.getData(); }这种方案最大的特点是拿不到锁的线程不需要等待直接返回旧数据所以接口响应时间非常稳定。但需要注意两个问题一是缓存里的物理过期时间要设得比逻辑过期时间长很多防止逻辑过期还没到物理 key 先被 Redis 淘汰了二是如果业务数据更新频繁逻辑过期方案会导致用户读到旧值的时间窗口变长需要业务方能够接受。3.4 锁的过期时间怎么定锁的过期时间是最容易拍脑袋又最容易出事的地方。设短了数据库查询慢一点锁就自动失效其他线程趁虚而入锁形同虚设设长了万一持有锁的线程发生异常其他请求会长时间阻塞等待重试的线程会越积越多。常规做法是预估“数据库查询 缓存回填”的最长耗时然后在这个基础上乘一个 3 到 5 倍的保险系数。比如数据库查询加上网络耗时平均 50 毫秒极端情况 1 秒内能完成那把锁的过期时间设为 5 秒已经足够。千万不要在锁里面执行 RPC 调用或者大批量数据操作否则锁注定要超时。这里可以套一个简单的公式锁过期时间 业务预估最大耗时 × 3然后向上取整到一个常见的秒数。如果业务耗时极不稳定建议升级方案使用 Redisson 这类带看门狗自动续期的分布式锁而不是死磕 SETNX。4. 实际操作中的问题与排查技巧实录4.1 锁误删问题没有原子性校验的代价有一次压测我故意把释放锁的逻辑写成了“先 get 判断 value再 del”结果在高并发下真的复现了误删线程 A 拿到锁后处理时间较长锁过期了线程 B 拿到锁并设置了自己的 value线程 A 处理完准备释放锁它读到 Redis 里的 value 已经不是自己的 requestId但因为在“判断”和“删除”之间发生了线程切换最终它把线程 B 的锁给删了。线程 C 趁虚而入再次拿到锁并发控制彻底失效。之后我所有释放锁的操作都强制要求用 Lua 脚本或者直接用 Redisson 的unlock()方法。判断和删除必须是一个原子操作没有任何商量余地。4.2 获取锁失败后重试策略不当自旋重试如果不加次数限制在热点 key 失效的瞬间会形成死循环风暴。比如 1 万个请求同时进来1 个请求拿到锁去查库剩下 9999 个请求全部在while(true)里循环读缓存对 Redis 的读压力本身就变成了一种二次冲击。更糟的是如果数据库查询时间较长这些自旋请求会把 Redis 连接池耗尽导致整个应用不可用。我的办法是限制自旋次数并且在每次自旋后 sleep 一个很小的随机时间避免所有请求在同一个时间点同时重试。你可以把 sleep 时间设置成 30 到 80 毫秒之间的随机值效果比固定 50 毫秒要好很多因为固定值容易造成“惊群”。4.3 空值缓存引发的数据短暂不一致缓存空值能防止穿透但也会带来一个副作用如果数据库里新插入了一条数据空值缓存还没过期用户会一直读到空结果。常规做法是把空值缓存的过期时间设得很短比如 30 到 60 秒同时配合主动删除业务侧插入数据时顺便把对应的空值缓存删掉。如果你使用 Redis 的发布订阅或者 Canal 同步数据变更也可以做到更实时的缓存更新。4.4 问题排查速查表现象可能原因排查方向缓存击穿仍然发生加锁后没有二次判定检查拿到锁之后是否重新查询缓存锁一直不释放请求大量堆积锁过期时间设置过长或释放锁代码没有在 finally 中执行检查 finally 块确认异常情况下锁也会释放锁被误删释放锁时未校验 value或校验与删除非原子操作改用 Lua 脚本释放锁自旋请求打爆 Redis无限循环重试增加重试次数上限sleep 加随机值数据库短暂出现压垮性查询空值缓存时间过长缓存删除后旧值未失效缩短空值 TTL同步删除缓存主从切换后锁失效锁写入了旧主节点考虑 Redisson RedLock或对锁追加 Watchdog 续期5. 延伸从单体锁到 Redis 生产环境可靠性5.1 单机锁到分布式锁的演进双重判定锁用到 Redis本质上是因为多个应用节点之间需要共享同一个互斥状态。如果你的服务只有单节点用 JVM 内部的synchronized或者ReentrantLock就能完成互斥根本不需要引入 Redis。但现在的业务系统基本都是多实例部署请求经过负载均衡分发到不同节点JVM 锁只能锁住当前进程跨节点就失效了。这也是为什么 Redis 分布式锁在缓存治理中如此重要。SETNX 的语义很简单只有当 key 不存在时才能设置成功谁能成功设置谁就获得了锁。这个操作天然就是跨节点共享的任何一个应用节点执行setIfAbsent(lockKey, requestId)都遵循同一套规则所以可以实现全局互斥。不过要记住一个长期被争论的点普通 Redis 主从架构下如果 master 节点宕机锁数据还没来得及同步到 slave新的 master 不会有这把锁的记录其他节点就能重复加锁这就是经典的锁丢失问题。官方给的建议是使用 RedLock但 RedLock 本身也有争议生产环境是否采用取决于你对锁的敏感度。对于大多数缓存击穿场景锁丢失导致的后果只是多几个请求穿透到数据库影响可控所以我在大部分项目里都没上 RedLock够用就行。5.2 部署形态对锁可靠性的影响开发环境我用 Docker 直接跑 redis 镜像一条命令就能起来反正不用考虑数据持久化。生产环境就需要重视部署形态了。如果是主从哨兵模式Redis Sentinel 负责故障自动切换但切换过程存在短暂的不可用窗口双重判定锁在这个窗口内可能不可用。如果是 Redis Cluster 集群模式锁 key 会被哈希到某个 slot加锁操作只会落在对应的主节点上整体可用性更高但仍然要注意主从切换导致的锁丢失。另一个容易忽略的点是 Redis 持久化。锁 key 虽然生命周期很短但如果在没开启持久化的情况下 Redis 重启所有锁都会丢失。曾有人问过我Redis 重启后锁全丢了是不是双重判定锁就废了我的回答是如果 Redis 已经重启缓存大概率也没了这时候系统要处理的是缓存雪崩而不是单一 key 的击穿双重判定锁本来就该配合缓存预热、空值缓存、多级缓存一起使用单靠一把锁扛不住这种极端场景。5.3 热点场景下的落地建议以商品详情页为例我见过的比较稳的搭配是一级本地缓存Caffeine挡掉大部分读流量二级 Redis 缓存存商品数据三级数据库兜底。热点 key 的过期时间在基础值上加上随机偏移量避免所有商品同时过期形成雪崩。本地缓存失效后请求走到 RedisRedis 失效后再走双重判定锁回源数据库。秒杀场景又是另一套玩法。秒杀商品的库存数据对一致性要求极高通常不会让请求直接穿透到数据库而是把库存预加载到 Redis 里用 Redis 的原子命令比如decr或 Lua 脚本扣减库存扣减成功后再异步落库。这时候双重判定锁用不上的原因很简单库存不是“过期后回填”的缓存数据而是实时变更的数据源锁的作用是保护数据库不被并发写击穿方向完全不同。所以我在实操中的建议是不要试图用一个方案解决所有缓存问题。双重判定锁是缓存击穿场景下的“银弹”但它只解决“热点 key 过期瞬间的并发读穿透”这一个问题。搭配缓存空值解决穿透、搭配随机过期时间解决雪崩、搭配本地缓存提升性能组合起来才是完整的缓存治理方案。6. 我踩过的坑和给你的一些建议做这个方案从前到后踩了不少坑最值得说的有三个。第一个是二次判定这个是最容易犯的错误代码写完一眼扫过去觉得没问题压测时才发现数据库压力没有预期下降排查半天才意识到拿到锁之后没有再查一次缓存。第二个是释放锁的原子性不用 Lua 脚本之前我总觉得自己不会遇到误删直到亲手复现了一次才彻底老实了。第三个是锁过期时间的设定一开始我给了一个看起来很长的 30 秒想着肯定够用结果有一次数据库慢查询真的把锁给撑过期了后面所有请求全部穿到了数据库那一次真的把我吓出一身冷汗。如果你准备在项目里落地双重判定锁我建议先做一个最小验证写一个模拟热点 key 的接口用 JMeter 开 1000 个并发线程同时请求观察 Redis 的命中率和数据库的 QPS 曲线。这个实验能直观地看到双重判定的效果也能帮你调整锁的过期时间和重试策略。等到参数调稳了再推广到核心业务风险会小很多。最后再分享一个小技巧不要在锁里面做“先查数据库再调外部接口再更新缓存”这种长链路操作。双重判定锁的正确姿势是锁内只做最必要的操作链路越短锁持有的时间越短系统的并发能力就越高。如果确实需要做长链路建议把锁拆成多个短锁或者直接用异步更新的方式让请求先拿旧数据返回后台再慢慢重建缓存。这也是逻辑过期方案在互联网大厂里如此流行的原因取舍之间全是业务方的真实诉求。