文章目录
- 基于redis实现分布式锁
- 基本实现
- 防死锁
- 防误删
- 高并发场景下无法保证原子性
- 使用lua保证删除原子性
- 把redis锁封装成方法
基于redis实现分布式锁
基本实现
借助于redis中的命令setnx(key, value)，key不存在就新增，存在就什么都不做。同时有多个客户端发送setnx命令，只有一个客户端可以成功，返回1（true）；其他的客户端返回0（false）。
- 多个客户端同时获取锁（setnx）
- 获取成功，执行业务逻辑，执行完成释放锁（del）
- 其他客户端等待重试
改造StockService方法：
@Service
public class StockService {
@Autowired
private StockMapper stockMapper;
@Autowired
private StringRedisTemplate redisTemplate;
public void deduct() {
// 加锁setnx
Boolean lock = this.redisTemplate.opsForValue().setIfAbsent("lock", "111");
// 重试：递归调用
if (!lock){
try {
Thread.sleep(50);
this.deduct();
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
try {
// 1. 查询库存信息
String stock = redisTemplate.opsForValue().get("stock").toString();
// 2. 判断库存是否充足
if (stock != null && stock.length() != 0) {
Integer st = Integer.valueOf(stock);
if (st > 0) {
// 3.扣减库存
redisTemplate.opsForValue().set("stock", String.valueOf(--st));
}
}
} finally {
// 解锁
this.redisTemplate.delete("lock");
}
}
}
}
其中，加锁也可以使用循环：
// 加锁，获取锁失败重试
while (!this.redisTemplate.opsForValue().setIfAbsent("lock", "111")){
try {
Thread.sleep(40);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
解锁：
// 释放锁
this.redisTemplate.delete("lock");
使用Jmeter压力测试如下：
防死锁
问题：setnx刚刚获取到锁，当前服务器宕机，导致del释放锁无法执行，进而导致锁无法锁无法释放（死锁）
解决：给锁设置过期时间，自动释放锁。
设置过期时间两种方式：
- 通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
- 使用set指令设置过期时间：set key value ex 3 nx（既达到setnx的效果，又设置了过期时间）
防误删
问题：可能会释放其他服务器的锁。
场景：如果业务逻辑的执行时间是7s。执行流程如下
-
index1业务逻辑没执行完，3秒后锁被自动释放。
-
index2获取到锁，执行业务逻辑，3秒后锁被自动释放。
-
index3获取到锁，执行业务逻辑
-
index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。
最终等于没锁的情况。
解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁
实现如下：
问题：删除操作缺乏原子性。
场景：
- index1执行删除时，查询到的lock值确实和uuid相等
- index1执行删除前，lock刚好过期时间已到，被redis自动释放
- index2获取了lock
- index1执行删除，此时会把index2的lock删除
解决方案：没有一个命令可以同时做到判断 + 删除，所有只能通过其他方式实现（LUA脚本）
高并发场景下无法保证原子性
redis采用单线程架构，可以保证单个命令的原子性，但是无法保证一组命令在高并发场景下的原子性。例如：
在串行场景下：A和B的值肯定都是3
在并发场景下：A和B的值可能在0-6之间。
极限情况下1：
则A的结果是0，B的结果是3
极限情况下2：
则A和B的结果都是6
如果redis客户端通过lua脚本把3个命令一次性发送给redis服务器，那么这三个指令就不会被其他客户端指令打断。Redis 也保证脚本会以原子性(atomic)的方式执行： 当某个脚本正在运行的时候，不会有其他脚本或 Redis 命令被执行。 这和使用 MULTI/ EXEC 包围的事务很类似。
但是MULTI/ EXEC方法来使用事务功能，将一组命令打包执行，无法进行业务逻辑的操作。这期间有某一条命令执行报错（例如给字符串自增），其他的命令还是会执行，并不会回滚。
使用lua保证删除原子性
删除LUA脚本：
if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end
代码实现：
public void deduct() {
String uuid = UUID.randomUUID().toString();
// 加锁setnx
while (!this.redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS)) {
// 重试：循环
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
try {
// this.redisTemplate.expire("lock", 3, TimeUnit.SECONDS);
// 1. 查询库存信息
String stock = redisTemplate.opsForValue().get("stock").toString();
// 2. 判断库存是否充足
if (stock != null && stock.length() != 0) {
Integer st = Integer.valueOf(stock);
if (st > 0) {
// 3.扣减库存
redisTemplate.opsForValue().set("stock", String.valueOf(--st));
}
}
} finally {
// 先判断是否自己的锁，再解锁
String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
"then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end";
this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList("lock"), uuid);
}
}
把redis锁封装成方法
/**
* @Author: hrd
* @CreateTime: 2023/10/20 14:57
* @Description:
*/
public interface Lock {
/**
* 获取所 默认30后自动释放锁
* @param key 业务key 根据自己业务取名
* @param code 唯一标识
*/
default void get(String key, String code) {
get(key,code,30);
}
/**
* 获取所
* @param key 业务key 根据自己业务取名
* @param code 唯一标识
* @param timeout 过期时间 单位：秒
*/
void get(String key, String code, long timeout);
/** 释放锁
* @param key 业务key 根据自己业务取名
* @param code 唯一标识
*/
void release(String key, String code);
}
import com.common.star.base.abs.lock.Lock;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @Author: hrd
* @CreateTime: 2023/10/20 14:58
* @Description:
*/
@Component
@Slf4j
@RequiredArgsConstructor
public class RedisLock implements Lock {
private final RedisTemplate<String, Object> redisTemplate;
@Override
public void get(String key,String code,long timeout) {
if (key == null) {
throw new RuntimeException("redis key 不能为空");
}
while (!Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(key, code, timeout, TimeUnit.SECONDS))) {
log.info("尝试获取锁----------------------{}", key);
// 重试：循环
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.info("Interrupted!", e);
}
}
log.info("成功获取锁-----------------{}", key);
}
@Override
public void release(String key,String code) {
if (key == null) {
return;
}
log.info("释放锁-----------------{}", key);
// 先判断是否自己的锁，再解锁
String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
"then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end";
this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), List.of(key), code);
}
}
