一、为什么需要分布式锁第一步没有分布式锁时本地锁有什么问题假设你有一个扣减库存的方法需要防止并发超卖。在单台服务器上你会这样写Service public class StockService { // Java 提供的本地可重入锁 private final ReentrantLock localLock new ReentrantLock(); public void deductStock(Long productId, int quantity) { localLock.lock(); // 加锁 try { // 查库存 int stock stockMapper.getStock(productId); if (stock quantity) { // 扣减 stockMapper.deduct(productId, quantity); } } finally { localLock.unlock(); // 释放锁 } } }这种写法在单台服务器上没问题但在企业级项目中有什么弊端企业级项目通常采用微服务架构同一个服务会部署多个实例比如 3 台服务器。每个实例都是一个独立的 JVM拥有自己独立的localLock对象。当两个用户同时下单时用户 A 的请求打到实例 1实例 1 的localLock加锁成功开始扣库存用户 B 的请求打到实例 2实例 2 的localLock也加锁成功也开始扣库存两个实例的锁互不相干它们同时进入了临界区同时查到了相同的库存数量同时做了扣减。这就是本地锁在分布式环境下的失效。synchronized、ReentrantLock、ConcurrentHashMap这些 Java 并发工具锁的范围仅限于当前 JVM 进程只能保证同一个 JVM 里面的多个线程互斥。无法跨越网络影响到其他服务器上的进程。因为本地锁在分布式场景下失效所以我们需要一个所有实例都能看到的锁这就是分布式锁。所以分布式锁的目的就是分布式锁就是让多个不同机器、不同 JVM 的服务实例也能够竞争同一把锁。第二步分布式锁的本质是什么分布式锁的核心思路是把锁的状态存储在一个独立于所有应用实例之外的第三方存储中所有实例都通过访问这个第三方存储来判断锁是否被占用。实例 1 ──┐ 实例 2 ──┼──→ Redis存储锁的状态 ←── 所有实例共享同一把锁 实例 3 ──┘一个合格的分布式锁必须满足三个条件条件为什么必须满足互斥性同一时刻只能有一个客户端持有锁防死锁客户端崩溃后锁能自动释放否则其他客户端永远拿不到锁可重入性同一个客户端可以多次获取同一把锁避免自己把自己锁死第三步最原始的 Redis 分布式锁实现及其弊端既然 Redis 是所有实例都能访问的最直观的思路就是用 Redis 的一个 Key 来表示锁。3.1 第一版SETNX EXPIREService public class StockService { Autowired private StringRedisTemplate redisTemplate; public void deductStock(Long productId, int quantity) { String lockKey lock:stock: productId; String clientId UUID.randomUUID().toString(); // 1. 尝试加锁SETNX SET if Not eXists Boolean locked redisTemplate.opsForValue() .setIfAbsent(lockKey, clientId); // 2. 设置过期时间防止死锁 redisTemplate.expire(lockKey, 30, TimeUnit.SECONDS); if (!Boolean.TRUE.equals(locked)) { throw new RuntimeException(获取锁失败请重试); } try { // 3. 执行业务查库存、扣减 int stock stockMapper.getStock(productId); if (stock quantity) { stockMapper.deduct(productId, quantity); } } finally { // 4. 释放锁 redisTemplate.delete(lockKey); } } }这个代码本质上是以下命令SET lock:order:123 uuid NX EX 30这里几个参数非常重要。NX表示只有这个 Key 不存在的时候才能设置成功。所以线程A SET lock:order:123 uuidA NX EX 30 ↓ 成功线程BSET lock:order:123 uuidB NX EX 30 ↓ 失败EX假设线程A ↓ 获取 Redis 锁 ↓ 开始处理业务 ↓ 突然服务器宕机如果没有过期时间lock:order:123会永远存在。那么以后线程B → 永远获取不到锁。所以必须设置EX 30也就是锁最多存在 30 秒。即使持有锁的服务挂掉30 秒之后锁也会自动释放。这版代码还有什么问题问题一SETNX 和 EXPIRE 不是原子操作如果setIfAbsent执行成功后应用突然挂了还没执行到expire这个 Key 就永远存在于 Redis 中其他实例永远拿不到锁系统死锁。问题二业务执行时间超过锁的过期时间你设置了 30 秒过期但数据库查询很慢业务执行了 35 秒。第 30 秒时 Redis 自动删除了 Key锁提前释放了。此时另一个实例拿到了锁也开始执行业务。等你的业务执行完去delete时你把别人持有的锁给删掉了。问题三释放锁时没有判断是不是自己加的锁finally里直接delete如果锁已经被 Redis 自动过期删除了另一个实例又创建了新锁你的delete会误删别人的锁。3.2 第二版原子加锁 Lua 释放针对问题一Redis 提供了原子命令SET lock:stock:1001 uuid NX PX 30000参数含义示例EX秒secondsEX 30 30秒PX毫秒millisecondsPX 30000 30000毫秒 30秒NX只有 Key 不存在时才设置相当于 SETNXPX 30000同时设置 30 秒过期时间针对问题三释放锁时用Lua 脚本保证判断删除的原子性if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 endjavaService public class StockService { Autowired private StringRedisTemplate redisTemplate; private static final String RELEASE_SCRIPT if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end; public void deductStock(Long productId, int quantity) { String lockKey lock:stock: productId; String clientId UUID.randomUUID().toString(); // 原子加锁SET key value NX PX 30000 Boolean locked redisTemplate.opsForValue() .setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); if (!Boolean.TRUE.equals(locked)) { throw new RuntimeException(获取锁失败); } try { int stock stockMapper.getStock(productId); if (stock quantity) { stockMapper.deduct(productId, quantity); } } finally { // 原子释放先判断 value 是否匹配匹配才删除 redisTemplate.execute( new DefaultRedisScript(RELEASE_SCRIPT, Long.class), Collections.singletonList(lockKey), clientId ); } } }这版还有什么问题问题二仍然没有解决业务执行时间超过过期时间如果业务执行了 35 秒第 30 秒锁自动过期另一个实例拿到锁两个实例同时执行业务。分布式锁失效了。你可能会想那把过期时间设长一点比如 5 分钟但如果应用真的挂了锁要 5 分钟后才释放系统在这 5 分钟内完全不可用。因为手动处理锁续期极其复杂且容易出错所以诞生了 Redisson。第四步Redisson 的设计——它到底解决了什么Redisson 是一个基于 Redis 的 Java 驻内存数据网格它把分布式锁的复杂性全部封装了起来。Redisson 解决了三个核心问题问题Redisson 的解决方案锁续期看门狗自动启动后台线程每隔 10 秒检查一次如果业务还没执行完自动给锁续期可重入性同一个线程可以多次获取同一把锁内部用计数器记录重入次数锁释放的原子性内部已经用 Lua 脚本封装好了不需要你手写第五步Spring Boot 集成 Redisson 的完整流程5.1 引入依赖dependencies !-- Spring Boot Web -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- Redis 依赖Spring Boot 自带 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency !-- Redisson Spring Boot Starter -- dependency groupIdorg.redisson/groupId artifactIdredisson-spring-boot-starter/artifactId version3.27.0/version /dependency /dependencies为什么用redisson-spring-boot-starter而不是普通redisson因为 starter 会自动读取application.yml中的 Redis 配置自动创建RedissonClientBean 并注入 Spring 容器。没有它你需要手动写配置类创建RedissonClient。5.2 配置文件spring: redis: host: localhost port: 6379 # 如果有密码 # password: yourpassword # Redisson 额外配置 redisson: config: | singleServerConfig: # 连接超时时间 connectTimeout: 10000 # 看门狗超时时间默认 30 秒 lockWatchdogTimeout: 300005.3 业务代码中使用分布式锁Service public class StockService { Autowired private RedissonClient redissonClient; Autowired private StockMapper stockMapper; public void deductStock(Long productId, int quantity) { // 1. 获取锁对象每个商品一把独立的锁细粒度 String lockKey lock:stock: productId; RLock lock redissonClient.getLock(lockKey); // 2. 尝试加锁 boolean isLocked false; try { // tryLock(等待时间, 锁自动释放时间, 时间单位) // 如果 leaseTime 设为 -1启用看门狗自动续期 isLocked lock.tryLock(5, -1, TimeUnit.SECONDS); if (!isLocked) { throw new RuntimeException(系统繁忙请稍后重试); } // 3. 执行业务逻辑临界区 int stock stockMapper.getStock(productId); if (stock quantity) { throw new RuntimeException(库存不足); } stockMapper.deduct(productId, quantity); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(获取锁被中断); } finally { // 4. 释放锁只有当前线程持有锁时才释放 if (isLocked lock.isHeldByCurrentThread()) { lock.unlock(); } } } }代码逐行解释代码为什么这样写没有它会怎样redissonClient.getLock(lockKey)通过锁名获取锁对象相同名称的锁全局唯一无法定位到具体的锁资源tryLock(5, -1, TimeUnit.SECONDS)最多等待 5 秒-1表示不设置固定过期时间启用看门狗自动续期如果设了固定时间业务执行超时会提前释放如果不等待获取不到锁立即失败lock.isHeldByCurrentThread()判断当前线程是否持有锁防止误删如果直接 unlock可能释放的是其他线程的锁finally { unlock() }保证无论业务是否异常锁最终都会被释放业务抛异常后锁不释放导致死锁看门狗Watch Dog的工作机制当你使用tryLock(waitTime, -1, unit)时 leaseTime 为 -1 线程获取锁成功 ↓ Redisson 启动一个后台线程看门狗 ↓ 每隔 10 秒检查业务线程还活着吗 ↓ 是 自动把锁的过期时间重置为 30 秒 ↓ 业务执行完调用 unlock() ↓ 看门狗线程停止锁被删除如果没有看门狗你必须预估业务执行时间设置leaseTime。预估长了应用崩溃后死锁时间长预估短了业务还没执行完锁就过期了。第六步Spring Cloud 场景下的考虑在 Spring Cloud 微服务中通常有多个服务实例。分布式锁的命名需要特别注意6.1 锁的粒度// ❌ 错误所有商品共用一把锁并发度极低 RLock lock redissonClient.getLock(lock:stock); // ✅ 正确每个商品一把锁互不干扰细粒度 RLock lock redissonClient.getLock(lock:stock: productId);为什么必须细粒度如果所有商品共用一把锁用户 A 买商品 1 时用户 B 买商品 2 也会被阻塞系统吞吐量急剧下降。6.2 锁前缀规范建议统一命名格式避免不同业务模块的锁冲突lock:{业务模块}:{资源类型}:{资源ID} 例如 lock:stock:product:1001 -- 商品库存锁 lock:order:deduct:20240820 -- 订单日结锁 lock:user:coupon:5678 -- 用户领券锁6.3 配合本地缓存的双层锁高并发优化在超高并发场景下每个请求都访问 Redis 加锁会有网络开销。可以结合本地锁做双层锁Service public class StockService { // 本地锁减少 Redis 访问次数 private final ConcurrentHashMapLong, ReentrantLock localLocks new ConcurrentHashMap(); Autowired private RedissonClient redissonClient; public void deductStock(Long productId, int quantity) { // 第一层本地锁JVM 级别减少 Redis 竞争 ReentrantLock localLock localLocks.computeIfAbsent(productId, k - new ReentrantLock()); localLock.lock(); try { // 第二层分布式锁跨 JVM 级别 RLock redisLock redissonClient.getLock(lock:stock: productId); boolean locked redisLock.tryLock(3, -1, TimeUnit.SECONDS); if (!locked) { throw new RuntimeException(系统繁忙); } try { // 执行业务 stockMapper.deduct(productId, quantity); } finally { redisLock.unlock(); } } finally { localLock.unlock(); } } }这个 ConcurrentHashMap 存的是商品 ID → 这个商品对应的本地锁例如现在有三个商品productId 1001productId 1002productId 1003那么这个 Map 里面可能就是localLocks ┌───────────┬─────────────────┐ │ 商品ID │ 本地锁 │ ├───────────┼─────────────────┤ │ 1001 │ ReentrantLock A │ │ 1002 │ ReentrantLock B │ │ 1003 │ ReentrantLock C │ └───────────┴─────────────────┘意思就是如果这个商品还没有对应的锁就创建一个如果已经有了就直接拿之前创建好的锁。第七步Redisson 的其他锁类型Redisson 不仅提供了普通的可重入锁还提供了更复杂的锁锁类型适用场景代码公平锁防止线程饥饿按请求顺序获取锁redissonClient.getFairLock(lock:order)读写锁读多写少场景读锁共享、写锁互斥redissonClient.getReadWriteLock(lock:config)红锁RedLock对可靠性要求极高防止单点 Redis 故障new RedissonRedLock(lock1, lock2, lock3)红锁是什么当单台 Redis 挂了锁信息丢失怎么办红锁要求在多个独立的 Redis 实例上同时加锁超过半数成功才算加锁成功。这样即使一台 Redis 挂了其他实例上的锁仍然有效。完整逻辑回顾步骤解决了什么问题如果没有它会怎样本地锁synchronized单 JVM 内线程互斥多实例部署时完全失效RedisSET NX PX跨 JVM 的互斥非原子操作导致死锁或误删Lua 脚本释放锁保证判断删除原子性误删其他实例持有的锁Redisson 看门狗自动续期解决业务超时锁提前释放并发安全问题Redisson 封装隐藏 Redis 命令细节每个业务都要手写几十行锁逻辑细粒度锁命名减少锁竞争提升并发全局一把锁系统吞吐量极低二、企业级场景中redis的部署情况用于分布式锁的 Redis实际部署时通常是一个独立的 Redis 服务/集群不是跟某一个微服务部署在一起。比如一个真实的企业项目可能是这样的用户 ↓ Nginx ↓ ┌──────────┼──────────┐ ↓ ↓ ↓ 微服务A 微服务A 微服务A 服务器1 服务器2 服务器3 │ │ │ └──────────┼──────────┘ ↓ Redis集群 ┌───────┼───────┐ ↓ ↓ ↓ Redis1 Redis2 Redis3这里最关键的是Redis 和微服务是两个独立的基础设施。1、为什么所有微服务都能访问 Redis其实和你访问 MySQL 是一样的。比如你的 Spring Boot 项目配置spring: datasource: url: jdbc:mysql://192.168.1.100:3306/order你的微服务服务器可能是192.168.1.101MySQL 是192.168.1.100虽然 MySQL 和 Spring Boot 根本不是一台机器但是 Spring Boot 可以通过192.168.1.100:3306访问 MySQL。Redis 完全一样。例如spring: data: redis: host: 192.168.1.200 port: 6379那么微服务1 ──────┐ │ 微服务2 ──────┼──→ 192.168.1.200:6379 │ 微服务3 ──────┘ Redis只要网络是通的并且 Redis 允许这些服务器访问那么它们都可以访问这个 Redis。2、那实际生产环境 Redis 会和服务器放在一起吗通常不会简单地和某个微服务绑在一起。更典型的是生产环境 ┌──────────────────────────────────────┐ │ │ │ 微服务服务器 │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │服务A │ │服务B │ │服务C │ │ │ └──┬───┘ └──┬───┘ └──┬───┘ │ │ │ │ │ │ │ └─────────┼─────────┘ │ │ ↓ │ │ ┌──────────────┐ │ │ │ Redis 集群 │ │ │ │ │ │ │ │ Master │ │ │ │ Slave │ │ │ │ Slave │ │ │ └──────────────┘ │ │ │ └──────────────────────────────────────┘甚至 Redis 根本不一定是公司自己部署的。比如使用云厂商提供的 RedisSpring Boot ↓ 云 Redis你的程序只需要知道Redis地址 Redis端口 密码就可以连接。3、这里还有一个非常重要的点你可能会产生一个疑问“既然 Redis 是一个独立服务器那 Redis 挂了怎么办”因为所有微服务 ↓ Redis ↓ 分布式锁Redis 如果只有一台Redis挂了 ↓ 所有服务都无法正常获取分布式锁所以生产环境一般不会简单搞一个单机 Redis而会根据可靠性要求使用Redis主从 Redis Sentinel Redis Cluster 云Redis高可用也就是说分布式锁本身也是依赖 Redis 集群的高可用能力的。
