分布式锁实战：从 JVM 锁到 Redis、Redisson 与 ZooKeeper
📝 笔记摘要：分布式锁从 JVM 锁到 Redis、Redisson 与 ZooKeeper
超卖、缓存击穿、定时任务重复执行，这些问题的解法最后都指向同一件事：让互斥发生在所有参与者都看得见的地方。这篇从 synchronized 走到 Redis，再走到 Redisson 和 ZooKeeper，把每一步解决了什么、又留下了什么坑串起来看。
这篇的结构
1. 分布式锁与普通锁
2. JVM 锁小试
3. 使用 MySQL 加 JVM 锁
4. MySQL 解决超卖问题
5. 锁总结
6. Redis 锁
7. 分布式锁
8. Redis 分布式锁
9. 红锁算法
10. Redisson
11. ZooKeeper 分布式锁
相关阅读
MySQL 事务、锁与索引 行锁、间隙锁、MVCC 的底层机制
Redis 缓存实战 缓存击穿怎么用互斥锁挡
Redis 核心 单线程模型和过期删除
Spring Boot 整合 Redis 连接配置和工具类
1. 分布式锁
分布式锁的实现方法有很多，mysql、redis、zk、etcd，都可以实现
1.1 与普通锁的区别
普通锁： 解决同一进程内多个线程操作同一资源的问题。
分布式锁： 解决多个进程同时操作同一资源的问题。（难点）
边界在哪
进程内 多个线程共享同一块内存，synchronized、ReentrantLock就能保证互斥
跨进程 线程不共享内存，JVM 锁完全看不见对方，必须有一个所有进程都能看见的地方当仲裁者
仲裁者的三种形态 数据库的行、Redis 的 key、ZooKeeper 的临时节点
难点不在加锁，在解锁 加锁只要「原子地判断 + 占用」，解锁要保证「只有持锁者能删」并且「进程崩了锁也能消失」
1.2 分布式锁的缺陷
引入了第三方的业务，行能肯定有所下降。
代价清单
多一次网络往返 加锁解锁各一次 RTT，锁粒度越细，延迟被放大的倍数越大
多一个可用性依赖 Redis 挂了，所有依赖这把锁的写操作全部排队或直接失败
持锁时长 = 业务耗时 业务慢，持锁者就是一道别人进不来的墙
不是所有地方都值得上 单机能串行的就别跨机器，跨机器的也别把锁粒度做得比业务还细
1.3 分布式锁和普通锁一起使用
这样可以提高性能，如下图所示，每个服务只会角逐出一个胜者， 这样到达第三方分布式锁的时候会减少压力提高性能。
两级锁是通用套路
先本地后远程 进程内先用synchronized抢一轮，只有这一轮的胜者才去抢分布式锁
收益是数量级的 N 个实例并发进来，打到 Redis 的只有 N 个，不是 N×线程数
Redisson 把它做成了配置MultiLock就是把本地锁和远程锁串起来
两层都不能省 本地锁只挡自己进程内的线程，跨实例的那部分还得靠分布式锁
2. JVM 锁小试
代码如下，使用jmeter模拟5000此并发请求，可以得到下面的结果，库存剩余数量还有剩余，说明存在争抢资源。
我们在调用的方法上面加上synchronized
可以看到解决了线程争抢数据的问题
控制台输出的顺序本身就是证据
没加锁 357、356、355 是乱的，说明读-改-写三步被别的线程插了进来
加锁后 变成严格递减，但仍然能减到负数——锁只保证串行，不保证业务正确
锁的是对象不是方法synchronized方法等价于锁this，粒度是整个对象
粒度要细 锁住整个 Service 方法，等于这个 Service 的所有操作都被串起来了
3. 使用 MySQL 加 JVM 锁
此时jvm本地锁会因为下面的情况失效。
当我们的实现类使用了多例模式而不是单例模式的时候会出现错误。如下图，就相当于又变成了并行实现，jvm的synchronized锁失效了.
当我们用了事务的时候synchronized锁机制会失效。
当我们的服务进行集群部署的时候也会失效，所以要用分布式锁。
多例场景 每个线程拿到的是不同的 Service 实例，
synchronized锁的是不同对象，等于完全不互斥。ReentrantLock一样，它锁的是实例字段
事务场景 加了@Transactional之后，方法上的锁在方法返回时就释放了，而事务是在方法返回后才提交——锁已经放开，数据还没写进去
集群场景 两个实例各锁各的 JVM，同一把「业务锁」变成了两把
结论 JVM 锁只适合单进程单实例，一旦多实例，就得把仲裁权交给外部
4. MySQL 解决超卖问题
4.1 使用一整个 sql语句
首先可以使用一个语句来解决，例如将售卖的逻辑改成一个sql语句，因为sql中的update等语句是自带锁的。
缺点：局限性太大，不实用。
用的是表级锁，会影响程序的行能
当同一个商品又多个仓库记录时不好调度
无法记录表前后的数据变更
那条 update 走的是行锁不是表锁
InnoDB 在 RR 级别下update ... where id = ?命中索引时只锁住命中的那几行
但条件写不严谨就退化成锁全表where version = ?如果 version 没建索引，会扫全表并把扫过的行都锁上
「无法记录变更前后」这条不成立 变更前后的值可以靠 binlog / CDC 拿到，或者自己开一张流水表记
真正的问题是局限性 库存同时存多个仓库、多规格、多批次时，一条 SQL 表达不了
4.2 悲观锁
悲观锁是行锁，具体实现的方法就是先用select。。。。for update来查询要变更的数据，注意查的时候where条件后面的数据必须有索引，否则就是表锁，然后再更新。
缺点：会产生死锁问题。
for update的锁范围由 where 决定
等值查询命中唯一索引 锁一行
命中普通索引 锁命中行 + 间隙
没走索引 锁全表lock in share mode是共享锁 多个事务可以同时持有，只能读不能写
间隙锁看隔离级别 RR 级别下for update才锁间隙，RC 级别只锁行，「幻读」依然可能
死锁怎么查show engine innodb status看最近一次死锁，或者看information_schema.innodb_trx
降低死锁的办法 固定加锁顺序、缩短事务、给热点行加索引
4.3 乐观锁
自己新加version等字段来解决。
具体的查询思路如下。
先查找要修改的数据，在修改的时候更改其版本号让其他线程修改不到你正在操作的数据。
代码实现如下
缺点：
吞吐量会越来越低，因为会越来越多的线程递归调用。浪费cpu资源
会出现栈溢出，可以加上线程sleep一段时间。
如果这个方法有事务，会出现连接超时。
mysql读写分离的情况下，不可靠。
版本号必须跟着 where 一起走
写法update ... set version = version + 1 where id = ? and version = ?，靠影响行数判断有没有抢到
重试不是免费的 库存 1 件、100 个请求抢，就是 99 次空转。热点商品上乐观锁的 CPU 开销比悲观锁还难看
递归改循环 栈溢出只是表象，根因是重试没有上限也没有退避
读写分离下为什么不可靠 读走从库，主库的版本号变化还没同步过去，读到的还是旧版本，更新会直接覆盖掉别人刚写的数据——主从延迟期间的乐观锁等于没有锁
事务里超时 乐观锁失败后立刻重试，会把数据库连接一个个占住，最后连接池被打满
5. 锁总结
这个排序有前提
排序描述的是「并发争抢激烈时的相对代价」 不是任何场景下的绝对优劣
争得少的时候一条 SQL 就够了 根本轮不到锁出场
JVM 本地锁不是不该用，是不能只用 单实例用它最划算，多实例它就兜不住了
选型顺序 先想办法不用锁 → 不行就用一条 SQL → 再不行用数据库的锁 → 还不行再上 Redis 或 ZooKeeper
6. Redis 锁
在service中编写下面的方法进行测试
public void redis() {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer=integer-1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock",String.valueOf(integer));
System.out.println("integer");
}
}else {
System.out.println("没数据");
}
}会出现超卖的情况
我们可以使用以下方法解决
使用jvm的锁
使用redis的乐观锁机制，用watch来进行。参考redis
使用分布式锁机制
超卖是怎么发生的
窗口在 get 和 set 之间 两个线程都读到 4946，都写 4945，卖了两件，库存只减了 1
单条命令原子不等于多条命令原子get+set是两条命令，中间就是窗口
压测报告看三个数 样本数、平均值、最大值。最大值 1422ms 说明有请求在排队stock的值和样本数对不上是正常的 一压就超卖，扣减次数和请求次数本来就不是一回事
6.1 乐观锁
public void redisleguansuo() {
// 先使用watch开启事务,要使用redisTemplate里面的如下写法
redisTemplate.execute(new SessionCallback<Object>() {
// 将业务放进这个方法里面
@Override
public Object execute(RedisOperations redisOperations) throws DataAccessException {
// 入参的redisOperations就相当于redisTemplate
// 开启watch,监听key
redisOperations.watch("stock");
// 1.查询库存
String stock= (String) redisOperations.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
// multi
redisOperations.multi();
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
// exec
List exec = redisOperations.exec();
// 根据exec的结果判断是否修改成功，如果返回的是null说明被锁了没操作成功需要进行递归
if(null==exec||exec.size()==0){
// 进行重试,递归调用，睡一下防止栈溢出
Thread.sleep(20);
redisleguansuo();
}
}
}
return null;
}
});
}缺点：会降低行能
WATCH的语义
它不是锁 是一个「有冲突就把整条事务作废」的标记exec返回 null 或空列表就是被改过 这时候必须重新 WATCH 再读，不能拿旧的值重试
重试要退避 固定 sleep 20ms 在高并发下会让所有线程同时醒来再同时冲突
循环次数要有上限 这版是递归，改成 while 加计数上限更稳
有一处笔误 回调里写的是redisTemplate.opsForValue().set(...)，应该用回调传进来的redisOperations。两者不是同一个连接，WATCH 根本不起作用
7. 分布式锁
使用场景：解决超卖现象、防止缓存击穿（热点数据过期，一瞬间大量的请求访问到数据库，这时我们加一层锁，同一时间只能一个请求访问数据库，就解决了问题）
实现方式：
redis进行实现
zookeeper进行实现
mysql实现分布式锁
还能用在哪
定时任务去重 多实例部署时同一个定时任务只让一个实例跑
幂等去重 同一笔请求在锁内只被处理一次
Leader 选举 拿到锁的实例负责一件事，其他实例待命
分布式限流 见后面 Redisson 的信号量
选型判据 能不能接受「偶尔两个实例同时执行」——能，Redis 够用；不能，走 ZooKeeper 或 etcd 加 fencing token
MySQL 实现最省事 靠唯一索引或者一张锁表的事务，占用连接但没有额外组件，延迟在毫秒级
8. Redis 分布式锁
8.1 使用方法
使用setncx命令来加一个锁，这个命令有排他性，只能第一个人能set返回的是1，后面的人set不了，返回的值是0.只有del删除所以后才能重新加锁。流程如下图所示。
标准写法已经不是 SETNX 了
一条命令搞定两件事SET key value NX PX ttl同时做「不存在才设置」和「设置过期时间」
为什么必须一条命令SETNX再EXPIRE是两条，中间宕机就留下一个永不过期的锁
value 必须是唯一标识 后面判断「这把锁是不是我的」全靠它
锁 key 要带业务前缀lock:order:1001，不要用一个裸的lock，否则不同业务的锁会互相踩
8.2 流程解读
首先加锁
其次解锁
最后对于枷锁失败的线程进行递归调用
递归改循环
栈会爆 竞争激烈时递归层数等于重试次数，每层一个栈帧
延迟会叠加 每次递归都新建栈帧，栈越深，调度和 GC 压力越大
改成 while 加退避 sleep 再配一个最大重试次数，超过就快速失败
8.3 代码实现
下面的区别是如果Ifprnsent是存在值了我才设置，IfAbsent是不存在了我就设置
public void redisfenbushisuo() {
// 使用setnx进行加锁
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111");
// 判断加锁是否成功并进行递归
if(!lock){
try {
Thread.sleep(20);
redisfenbushisuo();
} catch (InterruptedException e) {
e.printStackTrace();
}
return;
}
try {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
System.out.println("integer");
}
} else {
System.out.println("没数据");
}
}finally {
// 进行解锁
redisTemplate.delete("lock");
}
}两个方法的映射
setIfAbsent就是SETNXkey 不存在才写setIfPresent就是SET XXkey 存在才写。名字容易看反
这版代码有三个问题 没有过期时间（宕机就死锁）、delete("lock")不校验持有者（会误删别人的锁）、递归重试没有上限
8.4 修改循环不用递归
public void redisfenbushisuo() {
// 使用setnx进行加锁
Boolean lock = false;
// 判断加锁是否成功并进行循环
while(!lock){
lock =redisTemplate.opsForValue().setIfAbsent("lock", "111");
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
System.out.println("integer");
}
} else {
System.out.println("没数据");
}
}finally {
// 进行解锁
redisTemplate.delete("lock");
}
}改成 while 只解决了栈溢出
死锁还在 这版依然没有过期时间，服务一挂锁就永久留着
sleep 的位置 这版是抢锁失败后睡 50ms，锁竞争激烈时延迟线性叠加，而「抢锁成功后也睡」的做法更糟——拿到的锁还没开始用就在白白过期
8.5 解决死锁
当一个线程正在运行，刚加完锁服务器宕机了，没有解锁，这时其他线程也不能使用了了，因为锁没有释放。我们可以给redis的锁加一个过期时间来解决死锁问题。
public void redisfenbushisuo() {
// 使用setnx进行加锁
Boolean lock = false;
// 判断加锁是否成功并进行循环
while(!lock){
// 在这里加上过期时间
lock =redisTemplate.opsForValue().setIfAbsent("lock", "111",20, TimeUnit.MICROSECONDS);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
System.out.println("integer");
}
} else {
System.out.println("没数据");
}
}finally {
// 进行解锁
redisTemplate.delete("lock");
}
}单位写错是这类代码最常见的坑
这里写的是20, TimeUnit.MICROSECONDS20 微秒。业务还没跑完锁就没了，等于没加锁
正确的量级 给到业务 P99 的两到三倍，Redis 官方示例是 30 秒
加了过期也救不了误删 锁过期后被 B 拿到，A 醒来去删，删掉的是 B 的锁
8.6 解决死锁——误删
如果给的过期时间不合适会导致没执行完业务就把锁给放掉，导致锁失去意义。
解决办案：进行续期，有程序来完成，如果业务逻辑没有执行完可以给锁续期。和设置过期时间不冲突。
给锁设置uuid，当删除的时候判断一下是不是自己的uuid再进行删除。
public void redisfenbushisuo() {
//设置一个uuid
String uuid= UUID.randomUUID().toString();
// 使用setnx进行加锁
Boolean lock = false;
// 判断加锁是否成功并进行循环
while(!lock){
lock =redisTemplate.opsForValue().setIfAbsent("lock", uuid,20, TimeUnit.MICROSECONDS);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
System.out.println("integer");
}
} else {
System.out.println("没数据");
}
}finally {
//先判断是不是自己的锁
if(redisTemplate.opsForValue().get("lock").equals(uuid)){
// 进行解锁
redisTemplate.delete("lock");
}
}
}思路对了，但还差一步
比对和删除不是原子的 比对完到执行delete之间，另一个线程可能刚好锁过期并重新加锁，还是会误删
解法是 Lua 把 get 和 del 塞进一个脚本，Redis 单线程执行，天然原子
还有一个空指针 锁已经过期时get返回 null，.equals(uuid)直接 NPE
续期和过期不冲突 续期的思路是给锁再加一段 TTL，业务跑完再主动释放
8.7 redis的原子性
要保持原子性要使用lua脚本redis，教程地址。
将判断锁的归属，防误删的代码进行lua脚本化修改。
public void redisfenbushisuo() {
//设置一个uuid
String uuid= UUID.randomUUID().toString();
// 使用setnx进行加锁
Boolean lock = false;
// 判断加锁是否成功并进行循环
while(!lock){
lock =redisTemplate.opsForValue().setIfAbsent("lock", uuid,20, TimeUnit.SECONDS);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
System.out.println("integer");
}
} else {
System.out.println("没数据");
}
}finally {
//使用lua脚本
String scrip="if redis.call('get',KEYS[1])==ARGV[1]\n" +
"then\n" +
"return redis.call('del',KEYS[1])\n" +
"else\n" +
"return 0\n" +
"end";
// new DefaultRedisScript<>(scrip,boolean.class)里面一定要指定返回值的类型
redisTemplate.execute(new DefaultRedisScript<>(scrip,boolean.class), Arrays.asList("lock"),uuid);
}
}Lua 在 Redis 里是原子执行的
整个脚本跑完之前不会有别的命令插进来 所以「判断 + 删除」可以放进同一个脚本
脚本里有个拼写错误 Lua 关键字必须是return，写错执行时会直接报错
返回类型要写对 脚本返回 1 / 0，Java 侧用boolean.class；返回行数就用Long.class
还能省流量 脚本会被存成 SHA1，走EVALSHA只传 SHA
8.8 可重入锁
如下图所示，如果锁里面的方法也要set锁，就会死锁，因为A里面还没有删除锁，所以我们需要可重入锁
可重入锁的实现：使用hash结构的数据加上lua脚本
设计思路：
加锁思路：
首先判断锁存不存在(exists)，如果不存在就新设置一个锁（exists），hset key field value
如果存在锁了则判断是不是同一个uuid家的锁(hexists)，如果是就让value加上一，hincrby key field increment
否则循环
if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then redis.call('hincrby',KEYS[1],ARGV[1],1) redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 endif redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1
then
redis.call('hincrby',KEYS[1],ARGV[1],1)
redis.call('expire',KEYS[1],ARGV[2])
retrun 1
else
retrun 0
endkey: lock(锁的名称)
arg： uuid（锁的uuid） 30（过期时间）
解锁的思路
先判断自己对应uuid的所是否存在（hexists），不存在就抛出异常返回null
如果自己的锁存在，则value减1（hincrby），如果减1后为0，就删除锁
不为0，则返回0
if redis.call('hexists',KEYS[1],ARGV[1])==0 then return nil elseif redis.call('hincrby',KEYS[1],ARGV[1],-1)==0 then return redis.call('del',KEYS[1]) else return 0 endif redis.call('hexists',KEYS[1],ARGV[1])==0
then
return nil
elseif redis.call('hincrby',KEYS[1],ARGV[1],-1)==0
then
return redis.call('del',KEYS[1])
else
return 0
endkey: lock(锁的名称)
arg： uuid（锁的uuid）
代码实现：
首先要实现redis加锁解锁的一个接口来作为工具类
public class redislock implements Lock {
private StringRedisTemplate stringRedisTemplate;
private String lockName;
private String uuid;
private Long tttime=30l;
public redislock(StringRedisTemplate stringRedisTemplate,String lockName,String s) {
this.stringRedisTemplate = stringRedisTemplate;
this.lockName = lockName;
this.uuid= s;
}
// 调用tryLock()方法
@Override
public void lock() {
tryLock();
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
// 调用tryLock()有参构造方法
@Override
public boolean tryLock() {
try {
return tryLock(-1l,TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 有异常失败
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
if(time!=-1){
this.tttime=unit.toSeconds(time);
}
String script="if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then redis.call('hincrby',KEYS[1],ARGV[1],1) redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";
Boolean a=false;
while (!a){
Thread.sleep(50);
a=stringRedisTemplate.execute(new DefaultRedisScript<>(script,boolean.class), Arrays.asList(lockName),uuid,String.valueOf(tttime));
}
return false;
}
@Override
public void unlock() {
String script="if redis.call('hexists',KEYS[1],ARGV[1])==0 then return nil elseif redis.call('hincrby',KEYS[1],ARGV[1],-1)==0 then return redis.call('del',KEYS[1]) else return 0 end";
Long execute = stringRedisTemplate.execute(new DefaultRedisScript<>(script, long.class), Arrays.asList(lockName), uuid);
if(null==execute){
throw new RuntimeException("抛出异常");
}
}
@Override
public Condition newCondition() {
return null;
}
}
其次编写一个代理工厂模式方便后续使用
@Component
public class dailigongchang {
@Autowired
private StringRedisTemplate stringRedisTemplate;
public redislock redislock(String lockName,String s){
return new redislock(stringRedisTemplate,lockName,s);
}
}
用的时候直接注入代理工厂类进行使用
可重入锁解决的是自己和自己抢
数据结构从 string 换成 hash key 是锁名，field 是 uuid，value 是重入次数
加锁脚本exists == 0说明没人拿，hincrby计数到 1；hexists == 1说明是自己拿的，计数 +1；否则失败
解锁脚本 计数减 1，减到 0 才真删 key，否则只减计数
脚本里的retrun同样是拼错的 两段 Lua 都要改成return
工具类里有个 bugtryLock循环结束后return false，永远拿不到锁也永远返回失败，这里应该是return truelockInterruptibly()是空实现 调它不会真的等，要用得自己补上等待逻辑
命名不合规 类名redislock首字母小写，正式代码里应该叫RedisLock
8.9 给锁续期
在业务中设置一个定时任务美国一定的时间给锁续期。
续期要有个心跳
定时任务每 TTL/3 检查一次 锁还在就再续一个 TTL
定时任务救不了「进程假死」 进程在、网络断了，心跳任务照样跑，锁会被无限续下去，别人永远进不来
更稳的做法 续期时带一个「这次业务最多跑多久」的硬上限，超时就主动释放
Redisson 的 watchdog 就是这个思路 不指定租约时间时自动续期，指定了就变成一次性租约
9. 红锁算法
为了解决分布式锁再redis集群中存在的问题。一般没人用
红锁是什么
N 个独立主节点各拿一把锁 过半成功才算真正拿到
「一般没人用」今天依然成立 Kleppmann 2016 年那篇批评至今没有共识——主从切换时的时钟漂移会让两个客户端同时拿到锁
批评的三个点 客户端超时与锁过期之间的竞态、时钟漂移依赖假设、GC 停顿导致的「客户端以为自己还持锁其实锁早过期了」
业界现在的答案不是红锁本身，而是 fencing token 锁服务发一个单调递增的令牌，下游（数据库）拒绝比上次更小的令牌
实践建议 关键路径（资金、库存最终扣减）走 ZooKeeper / etcd 加 fencing token；Redis 锁用在「重复执行无害」的场景
10. redisson
redisson的宗旨是让用户的关注从redis命令上面分离，能供集中注意力再业务上。
10.1 使用方法-准备阶段
第一步：引入相关的依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.18.0</version>
</dependency>第二步：编写一个配置类将相关配置通过Bean引入方便我们后续使用。
@Configuration
public class redissonConfig {
@Bean
public RedissonClient redissonClient(){
Config config = new Config();
config.useSingleServer().setAddress("${spring.data.redis.host}:${spring.data.redis.port}");
config.useSingleServer().setPassword("${spring.data.redis.password}");
config.useSingleServer().setDatabase(3);
RedissonClient redisson = Redisson.create(config);
return redisson;
}
}redisson连接的具体配置如文档所示。redisson配置类连接配置
依赖版本
3.18.0是 2025 年初的版本 现在最新已经到 4.8.0，新项目直接上 4.xredisson和redisson-spring-boot-starter是两回事 starter 会自动装配RedissonClient，只要在配置文件里写spring.redis.redisson.config就够了，手写配置类是老写法
地址和密码不要写死在代码里 从配置注入，和 Redis 自己的连接配置保持一份
单机模式之外还有集群和哨兵useClusterServers() 和 useSentinelServers()，参数和 Lettuce 的配置项一一对应
10.2 redisson分布式锁
@Autowired
private RedissonClient redissonClient;
public void redisson() {
//使用redissonClient进行加锁
// 1.通过redissonClient获得一个锁
RLock lock = redissonClient.getLock("lock");
// 2.给锁上锁
lock.lock();
try {
// 1.查询库存
String stock = redisTemplate.opsForValue().get("stock");
// 2.判断库存充足
if (null != stock && stock.length() != 0) {
Integer integer = Integer.valueOf(stock);
integer = integer - 1;
// 3.进行删减
if (integer > 0) {
redisTemplate.opsForValue().set("stock", String.valueOf(integer));
}
} else {
System.out.println("没数据");
}
}finally {
// 3.给锁解锁
lock.unlock();
}
}redisson已经帮我们完成了锁的续期和可重入。
lock() 不带参数等于租约时间无限
watchdog 会自动续期 每 10 秒续到 30 秒，直到 unlock() 被调用
unlock() 必须在 finally 里 否则业务抛异常时锁会一直续着，别人永远进不来
只有持锁线程能解锁 换线程调 unlock() 会抛 IllegalMonitorStateException
isHeldByCurrentThread() 判断锁是不是当前线程持有的，常用来问「有没有别人正在处理」
wait 语义 lock() 是抢不到就一直等，tryLock() 抢不到立刻返回
10.3 redisson公平锁
秉承先来后到的原则，可以按照顺序进行执行，前面的锁拼的是运气，这个锁拼的是手速顺序。
RLock fairLock = redisson.getFairLock("anyLock");
// 最常见的使用方法
fairLock.lock();注意如果出现重复的发送请求，看看是不是nginx的代理超时时间是不是没设置，导致请求重发。
公平锁靠队列实现
Redisson 用一个 zset 记录等待者 按分数排队，先来先得
代价是吞吐 公平锁要维护队列，性能明显低于普通锁
重复请求这个坑值得单独记 同一个请求被执行两次，第二次 unlock() 就会抛异常，日志里看到 unlock 相关报错要先排查代理超时
幂等比公平更重要 请求会重发是常态，业务侧还得有幂等兜底
10.4 rsemaphore信号量
相当于给请求设置了一个最大的量，最大量不会超过设置的量，
单机版信号量
juc中的单机版信号量实现如下图所示，分布式有多个服务就不能使用。
public class test {
public static void main(String[] args) {
// 假设最大车位有三个
Semaphore semaphore = new Semaphore(3);
// 模拟进来了10个车
for (int i = 0; i < 10; i++) {
new Thread(() ->{
// 开启信号量。
try {
semaphore.acquire();
System.out.println("第"+Thread.currentThread().getName()+"号车进来了");
// 假设每个车都停10秒以内的一个随机时间
TimeUnit.SECONDS.sleep(new Random().nextInt(10));
System.out.println("第"+Thread.currentThread().getName()+"号车走了");
// 关闭信号量。
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
},i+"号车").start();
}
}
}可以看到最大的车位只有3
redisson分布式信号量
只要将Semaphore换成redisson提供的信号量就行。
public void test2() {
// 通过redissonClient设置分布式信号量
RSemaphore semaphore = redissonClient.getSemaphore("semaphore");
// 假设最大车位有三个
semaphore.trySetPermits(3);
// 模拟进来了10个车
for (int i = 0; i < 10; i++) {
new Thread(() ->{
try {
// 开启信号量。
semaphore.acquire();
System.out.println("第"+Thread.currentThread().getName()+"号车进来了");
// 假设每个车都停10秒以内的一个随机时间
TimeUnit.SECONDS.sleep(new Random().nextInt(10));
System.out.println("第"+Thread.currentThread().getName()+"号车走了");
// 关闭信号量。
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
},i+"号车").start();
}
}redisson闭锁
相当于进行倒数，当所有的条件满足了才会执行。
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.trySetCount(1);
latch.await();
// 在其他线程或其他JVM里
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.countDown();
JUC 的 Semaphore 是进程内的
多实例部署时每个实例各有 3 个车位 总车位变成 3×N，限流完全失效
信号量和锁不一样 信号量不互斥，它限制的是并发数，多个线程可以同时持有
trySetPermits(3) 只在第一次生效 第二次调用返回 false，因为许可数已经设过了
许可数改小不会踢人 已经在跑的请求照常跑完，新的请求开始阻塞
释放一定要在 finally 里 和锁一样，异常路径漏掉释放就是永久的容量损失
闭锁和信号量方向相反 信号量是「到点放行」，闭锁是「全部到齐才放行」
await() 会一直阻塞到计数归零 要超时用 await(long, TimeUnit)
11. zookeeper分布式锁
11.1 安装
docker安装zk，docker部署zookeeper
zk可视化工具地址
单机跑一条就够
官方镜像 docker run -d -p 2181:2181 zookeeper，2181 是客户端端口
可视化 ZooKeeper Dashboard、zkCli，或者 IDEA 的 ZK 插件
连不上先看端口 客户端端口和服务端端口不一样，宿主机映射要写对
11.2 基本命令
1.ls命令
可以使用下面的命令来查看文件结构
ls 文件路径2.get命令
获取节点下面的数据
get 节点名称3.create创建节点
create 节点名称4.delete删除节点
delete 节点名称5.添加数据
再create的时候添加数据，或者set添加数据
create 节点名称 数据内容
set 节点名称 数据内容6.添加子节点
create 节点名称/子节点名称
怎么进去
zkCli.sh -server 127.0.0.1:2181 进去之后就是交互式命令行
create 的父节点必须存在 否则报 Node does not exist
delete 删不掉有子节点的节点 要加 -r 递归删，或者先删子节点
这六条就是日常够用的全部 锁的实现最终也落在这几个原语上
11.3 节点类型
永久节点：一直都在
临时节点：退出客户端节点就不存在了，在创建的时候加一个 -e就行了，图形化界面勾选Ephemeral就行了。
永久序列化节点：创建的时候使用 -s命令，图形化界面勾选Sequence，创建的时候可以重名，会在后面自动加编号
临时序列化节点：按照上面的来
四种节点
持久、临时、持久顺序、临时顺序 组合出两类维度的区别：会不会自动消失、名字是不是自动带序号
临时节点是分布式锁的关键 客户端断连（会话超时）ZK 会自动删掉它，进程崩了锁自动释放，这是 Redis 锁做不到的
临时节点不能有子节点 需要顺序的场景就用「临时顺序节点」
顺序节点的名字带 10 位补零的序号 靠它排出先后顺序
11.4 节点监听
节点的监听都是一次性的。
节点创建时监听
stat -w 路径节点删除时监听
stat -w 路径节点数据变化时监听
get -w 路径子节点监听
ls -w 路径
watch 是一次性的
触发一次就自动注销 要继续监听得在回调里重新注册
ZooKeeper 3.6+ 换了语法 旧的 stat path [watch]、get path [watch]、ls path [watch] 已经废弃，新写法是 stat -w path、get -w path、ls -w path
get -w 加 -s 能同时看数据和子节点 get -s -w path
锁场景真正用的是 ls -w 监听子节点的增删，这正是「排队拿锁」的信号
11.5 spring使用zookeeper
ZkClient Curator两种工具
引入相关的依赖
Curator 是最常用的客户端
它封装了什么 在原生 ZK 客户端之上加了重试、连接状态管理、事件监听和一批现成的「配方」
curator-framework 和 curator-recipes 的区别 前者只提供客户端，后者才带 InterProcessMutex 这些现成的锁
ZK 客户端的取舍 延迟比 Redis 高（每次加解锁要几次 zk 写和 watch），换来的是强一致和崩溃自动释放
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.7.0</version>
</dependency>CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString("127.0.0.1:2181")
.retryPolicy(new ExponentialBackoffRetry(1000, 3))
.build();
client.start();
InterProcessMutex lock = new InterProcessMutex(client, "/lock/order:1001");
try {
if (lock.acquire(10, TimeUnit.SECONDS)) {
// 临界区
}
} finally {
if (lock.isAcquiredInThisProcess()) {
lock.release();
}
}
InterProcessMutex 就是可重入的分布式锁
底层是「临时顺序节点」 每个客户端在 /lock/order:1001 下创建一个带序号的临时节点，谁的序号最小谁持锁
没抢到的监听自己的前一个节点 前一个释放时唤醒自己——公平排队 + 崩溃自动释放 两件事同时成立
acquire 带超时会返回布尔值 拿不到锁就走降级，不要无限等
release() 前先 isAcquiredInThisProcess() 没拿到锁就释放会抛异常
和 Redis 锁怎么选 Redis 锁适合「偶尔重复执行无害」；ZK 锁适合「绝对不能同时执行」
💡 补充：三把锁的边界不一样，别指望一把锁通吃。JVM 锁管线程，Redis 锁管跨实例的并发数，数据库锁管最终的写入一致性。真正决定系统扛不扛得住的往往不是锁的种类，而是锁的粒度——粒度粗的锁再正确也会拖垮吞吐。以及一把锁的代价往往大于它的收益：能用唯一索引、幂等键、一条原子 SQL 解决的，不要上锁。真正写锁的时候，把四件事记住就不会错：加锁和设过期必须一条命令、释放必须校验持有者、释放必须原子、进程崩了锁要能自动消失。
🔥 一句话总结：锁的本质是把互斥的仲裁权交给所有参与者都看得见的地方——单进程用 JVM，多实例用 Redis 或 ZooKeeper，而正确性不取决于你选了哪把锁，只取决于「原子加锁 + 唯一持有者 + 原子释放 + 崩溃可恢复」这四件事有没有做全。