一、分布式锁的必要性
在单体应用时代,我们使用ReentrantLock
或synchronized
就能解决线程安全问题。但当系统拆分为分布式架构后(目前大多数公司应该不会只是单体应用了),跨进程的共享资源竞争就成了必须要解决的问题。
分布式锁由此应运而生,但是必须解决三大核心问题:
- 竞态条件:多人操作共享资源,顺序不可控
- 锁失效:锁自动过期但业务未执行完,其他客户端抢占资源 / 加锁成功但未设置过期时间,服务宕机导致死锁
- 锁误删:客户端A释放了客户端B持有的锁。
二、核心实现解析(附源码)
2.1 原子性加锁
local lockKey = KEYS[1] -- 锁的键名,如"order_lock_123"
local lockSecret = ARGV[1] -- 锁的唯一标识(建议UUID)
local expireTime = tonumber(ARGV[2]) -- 过期时间(单位:秒)
-- 参数有效性校验
if not expireTime or expireTime <= 0 then
return "0" -- 参数非法直接返回失败
end
-- 原子操作:SET lockKey lockSecret NX EX expireTime
local result = redis.call("set", lockKey, lockSecret, "NX", "EX", expireTime)
return result and "1" or "0" -- 成功返回"1",失败返回"0"
设计思路:
- value使用客户端唯一标识(推荐SnowflakeID)
- 参数校验:防止传入非法过期时间
- 原子性:单命令完成"判断+设置+过期"操作
2.2 看门狗续期机制
local lockKey = KEYS[1] -- 锁的键名
local lockSecret = ARGV[1] -- 锁标识
local expireTime = tonumber(ARGV[2]) -- 新的过期时间
-- 参数校验
if not expireTime or expireTime <= 0 then
return "0"
end
-- 获取当前锁的值
local storedSecret = redis.call("get", lockKey)
-- 续期逻辑
if storedSecret == lockSecret then
-- 值匹配则延长过期时间
local result = redis.call("expire", lockKey, expireTime)
return result == 1 and "1" or "0" -- 续期成功返回"1"
else
-- 锁不存在或值不匹配
return "0"
end
// 定时续约线程
watchdogExecutor.scheduleAtFixedRate(() -> {
locks.entrySet().removeIf(entry -> entry.getValue().isCancelled());
for (Entry<String, Lock> entry : locks.entrySet()) {
if (!entry.getValue().isCancelled()) {
String result = redisTemplate.execute(RENEWAL_SCRIPT,
Collections.singletonList(key),
lock.value, "30");
if ("0".equals(result)) lock.cancel();
}
}
}, 0, 10, TimeUnit.SECONDS);
设计思路:
- 续期间隔=过期时间/3(如30s过期则10s续期)
- 异步线程池需单独配置
- 双重校验锁状态(内存标记+Redis实际值)
2.3 安全释放锁
local lockKey = KEYS[1] -- 锁的键名
local lockSecret = ARGV[1] -- 要释放的锁标识
-- 获取当前锁的值
local storedSecret = redis.call("get", lockKey)
-- 校验锁归属
if storedSecret == lockSecret then
-- 值匹配则删除Key
return redis.call("del", lockKey) == 1 and "1" or "0"
else
-- 值不匹配
return "0"
end
设计思路:
- 校验value避免误删其他线程的锁
三、源码
package org.example.tao.util;
import com.alibaba.fastjson2.JSON;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import javax.annotation.PreDestroy;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class RedisUtils {
static class Lock {
private final String value;
private volatile boolean isCancelled = false;
public Lock(String value) {
this.value = value;
}
public boolean isCancelled() {
return isCancelled;
}
public void cancel() {
isCancelled = true;
}
}
private static final String LOCK_LUA = "local lockKey = KEYS[1]\n" + "local lockSecret = ARGV[1]\n" + "local expireTime = tonumber(ARGV[2]) -- 动态过期时间\n" + "if not expireTime or expireTime <= 0 then\n" + " return \"0\"\n" + "end\n" + "local result = redis.call(\"set\", lockKey, lockSecret, \"NX\", \"EX\", expireTime)\n" + "return result and \"1\" or \"0\"";
private static final String RELEASE_LOCK_LUA = "local lockKey = KEYS[1]\n" + "local lockSecret = ARGV[1]\n" + "local storedSecret = redis.call(\"get\", lockKey)\n" + "if storedSecret == lockSecret then\n" + " return redis.call(\"del\", lockKey) == 1 and \"1\" or \"0\"\n" + "else\n" + " return \"0\"\n" + "end";
private static final String RENEWAL_LUA = "local lockKey = KEYS[1]\n" + "local lockSecret = ARGV[1]\n" + "local expireTime = tonumber(ARGV[2])\n" + "if not expireTime or expireTime <= 0 then\n" + " return \"0\"\n" + "end\n" + "local storedSecret = redis.call(\"get\", lockKey)\n" + "if storedSecret == lockSecret then\n" + " local result = redis.call(\"expire\", lockKey, expireTime)\n" + " return result == 1 and \"1\" or \"0\"\n" + "else\n" + " return \"0\"\n" + "end";
private final String defaultExpireTime = "30";
private final RedisTemplate<String, String> redisTemplate;
private final Map<String, Lock> locks = new ConcurrentHashMap<>();
private final ScheduledExecutorService watchdogExecutor = Executors.newScheduledThreadPool(1);
public RedisUtils(RedisTemplate<String, String> redisTemplate) {
this.redisTemplate = redisTemplate;
watchdogExecutor.scheduleAtFixedRate(() -> {
try {
System.out.println("watchdogExecutor 执行中... locks => " + JSON.toJSONString(locks));
locks.entrySet().removeIf(entry -> entry.getValue().isCancelled());
for (Map.Entry<String, Lock> entry : locks.entrySet()) {
String key = entry.getKey();
Lock lock = entry.getValue();
if (!lock.isCancelled()) {
RedisScript<String> redisScript = RedisScript.of(RENEWAL_LUA, String.class);
String result = redisTemplate.execute(redisScript, Collections.singletonList(key), lock.value, defaultExpireTime);
if (Objects.equals(result, "0")) {
lock.cancel(); // 移除已经释放的锁
}
}
}
} catch (Exception e) {
System.err.println("看门狗任务执行失败: " + e.getMessage());
}
}, 0, 10, TimeUnit.SECONDS);
}
public boolean acquireLock(String key, String value) {
RedisScript<String> redisScript = RedisScript.of(LOCK_LUA, String.class);
String result = redisTemplate.execute(redisScript, Collections.singletonList(key), value, defaultExpireTime);
if (Objects.equals(result, "1")) {
locks.put(key, new Lock(value));
return true;
}
return false;
}
public boolean acquireLockWithRetry(String key, String value, int maxRetries, long retryIntervalMillis) {
int retryCount = 0;
while (retryCount < maxRetries) {
boolean result = this.acquireLock(key, value);
if (result) {
locks.put(key, new Lock(value));
return true;
}
retryCount++;
try {
Thread.sleep(retryIntervalMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
return false;
}
public boolean releaseLock(String key, String value) {
RedisScript<String> redisScript = RedisScript.of(RELEASE_LOCK_LUA, String.class);
String result = redisTemplate.execute(redisScript, Collections.singletonList(key), value);
if (Objects.equals(result, "1")) {
Lock lock = locks.get(key);
if (lock != null) {
lock.cancel();
}
return true;
}
return false;
}
@PreDestroy
public void shutdown() {
watchdogExecutor.shutdown();
try {
if (!watchdogExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
watchdogExecutor.shutdownNow();
}
} catch (InterruptedException e) {
watchdogExecutor.shutdownNow();
}
}
}
四、如何使用
4.1 配置类
@Configuration
public class AppConfig {
@Resource
private RedisTemplate<String, String> redisTemplate;
@Bean
public RedisUtils init() {
return new RedisUtils(redisTemplate);
}
}
4.2 使用
@RestController
@RequestMapping("/users")
public class UserController {
@Resource
private RedisTemplate<String, String> redisTemplate;
@PostMapping("/test2")
public Boolean test2(@RequestBody Map<String, String> map) {
boolean res;
if (Objects.equals(map.get("lockFlag"), "true")) {
res = redisUtils.acquireLock(map.get("key"), map.get("value"));
} else {
res = redisUtils.releaseLock(map.get("key"), map.get("value"));
}
return res;
}
}
后记
还是免责声明,仅供学习参考