作者：微信小助手
发布时间：2024-09-12T22:20:11
- 尼恩说在前面 - Redis分布式锁过期怎么办？ - 两大Redis 分布式 - 分布式锁一般有如下的特点： - 基于Jedis 的API实现分布式锁 - Redis几种架构 - 首先看两个命令： - 基于Jedis API的分布式锁的总体流程： - 简单加锁：使用set的命令时，同时设置过期时间 - 基于Jedis 的API实现简单解锁代码 - 基于Lua脚本实现分布式锁 - lua脚本的好处 - 基于纯Lua脚本的分布式锁的执行流程 - 加锁的Lua脚本：lock.lua - 解锁的Lua脚本：unlock.lua： - 两个文件，放在资源文件夹下备用： - 在Java中调用lua脚本，完成加锁操作 - 在Java中调用lua脚本，完成加锁操作 - 通过实现JUC的显示锁Lock接口，实现一个简单的分布式锁 - 编写RedisLockService用于管理JedisLock - 测试用例 - STW导致的锁过期问题 - 锁过期问题 的解决方案 - 方式一：模拟CAS乐观锁的方式，增加版本号 - 方式二：watch dog自动延期机制 - 为啥推荐使用Redission - Redisson简介 - Redisson与Jedis的对比 - Redission 的源码地址： - 特性 & 功能： - Redisson的使用 - 如何安装 Redisson - 获取RedissonClient对象 - SpringBoot整合Redisson - 使用 RLock 实现 Redis 分布式锁 - Redision锁 核心源码分析 - getLock()方法 - tryLock方法 - tryAcquire()方法 - tryLockInnerAsync - 原理：加锁机制 - Lua脚本的详解 - 关于 lua脚本的参数解释： - 锁互斥机制 - 可重入加锁机制 - 释放锁机制 - unlock 源码 - unlockInnerAsync方法 - 原理：Redision 解锁机制 - 通过redis Channel 解锁订阅 - watch dog自动延期机制 - 使用watchDog机制实现锁的续期 - redisson watchdog 使用和原理 - tryAcquireAsync原理 - renewExpiration执行延期动作 - renewExpirationAsync - watchLog总结 - 总结：Redis分布式锁过期怎么办？ - 说在最后：有问题找老架构取经 上岸奇迹：中厂大龄34岁，被裁8月收一大厂offer， 年薪65W，转架构后逆天改命! 案例1：42岁被裁2年，天快塌了，急救1个月，拿到开发经理offer，起死回生 案例2：35岁被裁6个月， 职业绝望，转架构急救上岸，DDD和3高项目太重要了 案例3：失业15个月，学习40天拿offer， 绝境翻盘，如何实现？ 实现职业转型，极速上岸本文目录
-一、导入Maven依赖
-二、核心配置文件
-三、添加配置类
-自定义starter
Redis分布式锁过期怎么办？
两大Redis 分布式
分布式锁一般有如下的特点：
基于Jedis 的API实现分布式锁
Redis几种架构
首先看两个命令：
OK
127.0.0.1:6379> setnx lock "unlock"
(integer) 0
127.0.0.1:6379> setnx lock "lock"
(integer) 0
127.0.0.1:6379>
(integer) 1
127.0.0.1:6379> ttl lock
8
基于Jedis API的分布式锁的总体流程：
setnx
是成功的，但是
expire
设置失败，一旦出现了释放锁失败，或者没有手工释放，那么这个锁永远被占用，其他线程永远也抢不到锁。
简单加锁：使用set的命令时，同时设置过期时间
(nil)
127.0.0.1:6379>
127.0.0.1:6379> set test "111" EX 100 NX
OK
PX milliseconds：设置失效时长，单位毫秒
NX：key不存在时设置value，成功返回OK，失败返回(nil)
XX：key存在时设置value，成功返回OK，失败返回(nil)
@Slf4j
@Data
@AllArgsConstructor
public class JedisCommandLock {
private RedisTemplate redisTemplate;
private static final String LOCK_SUCCESS = "OK";
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "PX";
/**
* 尝试获取分布式锁
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @param expireTime 超期时间
* @return 是否获取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
}
jedis.set(String key, String value, String nxxx, String expx, int time)
UUID.randomUUID().toString()
方法生成。
基于Jedis 的API实现简单解锁代码
@Slf4j
@Data
@AllArgsConstructor
public class JedisCommandLock {
private static final Long RELEASE_SUCCESS = 1L;
/**
* 释放分布式锁
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @return 是否释放成功
*/
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
}
}
jedis.eval()
方法里，并使参数KEYS[1]赋值为lockKey，ARGV[1]赋值为requestId。eval()方法是将Lua代码交给Redis服务端执行。
jedis.del(lockKey);
}
// 判断加锁与解锁是不是同一个客户端
if (requestId.equals(jedis.get(lockKey))) {
// 若在此时，这把锁突然不是这个客户端的，则会误解锁
jedis.del(lockKey);
}
}
基于Lua脚本实现分布式锁
lua脚本的好处
基于纯Lua脚本的分布式锁的执行流程
加锁的Lua脚本：lock.lua
--- 1 success
---
local key = KEYS[1]
local requestId = KEYS[2]
local ttl = tonumber(KEYS[3])
local result = redis.call('setnx', key, requestId)
if result == 1 then
--PEXPIRE:以毫秒的形式指定过期时间
redis.call('pexpire', key, ttl)
else
result = -1;
-- 如果value相同，则认为是同一个线程的请求，则认为重入锁
local value = redis.call('get', key)
if (value == requestId) then
result = 1;
redis.call('pexpire', key, ttl)
end
end
-- 如果获取锁成功，则返回 1
return result
解锁的Lua脚本：unlock.lua：
--- 1 success
-- unlock key
local key = KEYS[1]
local requestId = KEYS[2]
local value = redis.call('get', key)
if value == requestId then
redis.call('del', key);
return 1;
end
return -1
两个文件，放在资源文件夹下备用：
在Java中调用lua脚本，完成加锁操作
import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.common.util.IOUtil;
import com.crazymaker.springcloud.standard.context.SpringContextUtil;
import com.crazymaker.springcloud.standard.lua.ScriptHolder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import java.util.ArrayList;
import java.util.List;
@Slf4j
public class InnerLock {
private RedisTemplate redisTemplate;
public static final Long LOCKED = Long.valueOf(1);
public static final Long UNLOCKED = Long.valueOf(1);
public static final int EXPIRE = 2000;
String key;
String requestId; // lockValue 锁的value ,代表线程的uuid
/**
* 默认为2000ms
*/
long expire = 2000L;
private volatile boolean isLocked = false;
private RedisScript lockScript;
private RedisScript unLockScript;
public InnerLock(String lockKey, String requestId) {
this.key = lockKey;
this.requestId = requestId;
lockScript = ScriptHolder.getLockScript();
unLockScript = ScriptHolder.getUnlockScript();
}
/**
* 抢夺锁
*/
public void lock() {
if (null == key) {
return;
}
try {
List<String> redisKeys = new ArrayList<>();
redisKeys.add(key);
redisKeys.add(requestId);
redisKeys.add(String.valueOf(expire));
Long res = (Long) getRedisTemplate().execute(lockScript, redisKeys);
isLocked = false;
} catch (Exception e) {
e.printStackTrace();
throw BusinessException.builder().errMsg("抢锁失败").build();
}
}
/**
* 有返回值的抢夺锁
*
* @param millisToWait
*/
public boolean lock(Long millisToWait) {
if (null == key) {
return false;
}
try {
List<String> redisKeys = new ArrayList<>();
redisKeys.add(key);
redisKeys.add(requestId);
redisKeys.add(String.valueOf(millisToWait));
Long res = (Long) getRedisTemplate().execute(lockScript, redisKeys);
return res != null && res.equals(LOCKED);
} catch (Exception e) {
e.printStackTrace();
throw BusinessException.builder().errMsg("抢锁失败").build();
}
}
//释放锁
public void unlock() {
if (key == null || requestId == null) {
return;
}
try {
List<String> redisKeys = new ArrayList<>();
redisKeys.add(key);
redisKeys.add(requestId);
Long res = (Long) getRedisTemplate().execute(unLockScript, redisKeys);
// boolean unlocked = res != null && res.equals(UNLOCKED);
} catch (Exception e) {
e.printStackTrace();
throw BusinessException.builder().errMsg("释放锁失败").build();
}
}
private RedisTemplate getRedisTemplate() {
if(null==redisTemplate)
{
redisTemplate= (RedisTemplate) SpringContextUtil.getBean("stringRedisTemplate");
}
return redisTemplate;
}
}
在Java中调用lua脚本，完成加锁操作
import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.common.util.ThreadUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@Slf4j
@Data
@AllArgsConstructor
public class JedisLock implements Lock {
private RedisTemplate redisTemplate;
RedisScript<Long> lockScript = null;
RedisScript<Long> unLockScript = null;
public static final int DEFAULT_TIMEOUT = 2000;
public static final Long LOCKED = Long.valueOf(1);
public static final Long UNLOCKED = Long.valueOf(1);
public static final Long WAIT_GAT = Long.valueOf(200);
public static final int EXPIRE = 2000;
String key;
String lockValue; // lockValue 锁的value ,代表线程的uuid
/**
* 默认为2000ms
*/
long expire = 2000L;
public JedisLock(String lockKey, String lockValue) {
this.key = lockKey;
this.lockValue = lockValue;
}
private volatile boolean isLocked = false;
private Thread thread;
/**
* 获取一个分布式锁 , 超时则返回失败
*
* @return 获锁成功 - true | 获锁失败 - false
*/
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
//本地可重入
if (isLocked && thread == Thread.currentThread()) {
return true;
}
expire = unit != null ? unit.toMillis(time) : DEFAULT_TIMEOUT;
long startMillis = System.currentTimeMillis();
Long millisToWait = expire;
boolean localLocked = false;
int turn = 1;
while (!localLocked) {
localLocked = this.lockInner(expire);
if (!localLocked) {
millisToWait = millisToWait - (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if (millisToWait > 0L) {
/**
* 还没有超时
*/
ThreadUtil.sleepMilliSeconds(WAIT_GAT);
log.info("睡眠一下，重新开始，turn:{},剩余时间：{}", turn++, millisToWait);
} else {
log.info("抢锁超时");
return false;
}
} else {
isLocked = true;
localLocked = true;
}
}
return isLocked;
}
/**
* 有返回值的抢夺锁
*
* @param millisToWait
*/
public boolean lockInner(Long millisToWait) {
if (null == key) {
return false;
}
try {
List<String> redisKeys = new ArrayList<>();
redisKeys.add(key);
redisKeys.add(lockValue);
redisKeys.add(String.valueOf(millisToWait));
Long res = (Long) redisTemplate.execute(lockScript, redisKeys);
return res != null && res.equals(LOCKED);
} catch (Exception e) {
e.printStackTrace();
throw BusinessException.builder().errMsg("抢锁失败").build();
}
}
}
通过实现JUC的显示锁Lock接口，实现一个简单的分布式锁
import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.common.util.ThreadUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@Slf4j
@Data
@AllArgsConstructor
public class JedisLock implements Lock {
private RedisTemplate redisTemplate;
RedisScript<Long> lockScript = null;
RedisScript<Long> unLockScript = null;
//释放锁
@Override
public void unlock() {
if (key == null || requestId == null) {
return;
}
try {
List<String> redisKeys = new ArrayList<>();
redisKeys.add(key);
redisKeys.add(requestId);
Long res = (Long) redisTemplate.execute(unLockScript, redisKeys);
} catch (Exception e) {
e.printStackTrace();
throw BusinessException.builder().errMsg("释放锁失败").build();
}
}
}
编写RedisLockService用于管理JedisLock
import com.crazymaker.springcloud.common.util.IOUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
@Slf4j
@Data
public class RedisLockService
{
private RedisTemplate redisTemplate;
static String lockLua = "script/lock.lua";
static String unLockLua = "script/unlock.lua";
static RedisScript<Long> lockScript = null;
static RedisScript<Long> unLockScript = null;
{
String script = IOUtil.loadJarFile(RedisLockService.class.getClassLoader(),lockLua);
// String script = FileUtil.readString(lockLua, Charset.forName("UTF-8" ));
if(StringUtils.isEmpty(script))
{
log.error("lua load failed:"+lockLua);
}
lockScript = new DefaultRedisScript<>(script, Long.class);
// script = FileUtil.readString(unLockLua, Charset.forName("UTF-8" ));
script = IOUtil.loadJarFile(RedisLockService.class.getClassLoader(),unLockLua);
if(StringUtils.isEmpty(script))
{
log.error("lua load failed:"+unLockLua);
}
unLockScript = new DefaultRedisScript<>(script, Long.class);
}
public RedisLockService(RedisTemplate redisTemplate)
{
this.redisTemplate = redisTemplate;
}
public Lock getLock(String lockKey, String lockValue) {
JedisLock lock=new JedisLock(lockKey,lockValue);
lock.setRedisTemplate(redisTemplate);
lock.setLockScript(lockScript);
lock.setUnLockScript(unLockScript);
return lock;
}
}
测试用例
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {DemoCloudApplication.class})
// 指定启动类
public class RedisLockTest {
@Resource
RedisLockService redisLockService;
private ExecutorService pool = Executors.newFixedThreadPool(10);
@Test
public void testLock() {
int threads = 10;
final int[] count = {0};
CountDownLatch countDownLatch = new CountDownLatch(threads);
long start = System.currentTimeMillis();
for (int i = 0; i < threads; i++) {
pool.submit(() ->
{
String lockValue = UUID.randomUUID().toString();
try {
Lock lock = redisLockService.getLock("test:lock:1", lockValue);
boolean locked = lock.tryLock(10, TimeUnit.SECONDS);
if (locked) {
for (int j = 0; j < 1000; j++) {
count[0]++;
}
log.info("count = " + count[0]);
lock.unlock();
} else {
System.out.println("抢锁失败");
}
} catch (Exception e) {
e.printStackTrace();
}
countDownLatch.countDown();
});
}
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("10个线程每个累加1000为：= " + count[0]);
//输出统计结果
float time = System.currentTimeMillis() - start;
System.out.println("运行的时长为(ms)：" + time);
System.out.println("每一次执行的时长为(ms)：" + time / count[0]);
}
}
2021-05-04 23:02:11.901 INFO 22120 --- [pool-1-thread-1] c.c.springcloud.standard.lock.JedisLock LN:81 睡眠一下，重新开始，turn:3,剩余时间：9585
2021-05-04 23:02:11.902 INFO 22120 --- [pool-1-thread-1] c.c.springcloud.lock.RedisLockTest LN:50 count = 7000
2021-05-04 23:02:12.100 INFO 22120 --- [pool-1-thread-4] c.c.springcloud.standard.lock.JedisLock LN:81 睡眠一下，重新开始，turn:3,剩余时间：9586
2021-05-04 23:02:12.101 INFO 22120 --- [pool-1-thread-5] c.c.springcloud.standard.lock.JedisLock LN:81 睡眠一下，重新开始，turn:3,剩余时间：9585
2021-05-04 23:02:12.101 INFO 22120 --- [pool-1-thread-8] c.c.springcloud.standard.lock.JedisLock LN:81 睡眠一下，重新开始，turn:3,剩余时间：9585
2021-05-04 23:02:12.101 INFO 22120 --- [pool-1-thread-4] c.c.springcloud.lock.RedisLockTest LN:50 count = 8000
2021-05-04 23:02:12.102 INFO 22120 --- [pool-1-thread-8] c.c.springcloud.lock.RedisLockTest LN:50 count = 9000
2021-05-04 23:02:12.304 INFO 22120 --- [pool-1-thread-5] c.c.springcloud.standard.lock.JedisLock LN:81 睡眠一下，重新开始，turn:4,剩余时间：9383
2021-05-04 23:02:12.307 INFO 22120 --- [pool-1-thread-5] c.c.springcloud.lock.RedisLockTest LN:50 count = 10000
10个线程每个累加1000为：= 10000
运行的时长为(ms)：827.0
每一次执行的时长为(ms)：0.0827
STW导致的锁过期问题
function writeData(filename, data) {
boolean locked = lock.tryLock(10, TimeUnit.SECONDS);
if (!locked) {
throw 'Failed to acquire lock';
}
try {
//将数据写到文件
var file = storage.readFile(filename);
var updated = updateContents(file, data);
storage.writeFile(filename, updated);
} finally {
lock.unlock();
}
}
锁过期问题 的解决方案
方式一：模拟CAS乐观锁的方式，增加版本号
方式二：watch dog自动延期机制
为啥推荐使用Redission
Redisson简介
Redisson与Jedis的对比
Redission 的源码地址：
特性 & 功能：
Redisson的使用
如何安装 Redisson
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.11.4</version>
</dependency>
获取RedissonClient对象
config.useSingleServer().setAddress("redis://myredisserver:6379");
RedissonClient redisson = Redisson.create(config);
Config config = new Config();
config.useSingleServer().setAddress("redis://myredisserver:6379");
RedissonClient redisson = Redisson.create(config);// connects to 127.0.0.1:6379 by defaultRedisson
Client redisson = Redisson.create();
SpringBoot整合Redisson
一、导入Maven依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.11.4</version>
</dependency>
二、核心配置文件
redis:
host: 127.0.0.1
port: 6379
database: 0
timeout: 5000
三、添加配置类
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RedissonConfig {
@Autowired
private RedisProperties redisProperties;
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
String redisUrl = String.format("redis://%s:%s", redisProperties.getHost() + "", redisProperties.getPort() + "");
config.useSingleServer().setAddress(redisUrl).setPassword(redisProperties.getPassword());
config.useSingleServer().setDatabase(3);
return Redisson.create(config);
}
}
自定义starter
使用 RLock 实现 Redis 分布式锁
@Resource
RedissonManager redissonManager;
@Test
public void testLockExamples() {
// 默认连接上 127.0.0.1:6379
RedissonClient redisson = redissonManager.getRedisson();
// RLock 继承了 java.util.concurrent.locks.Lock 接口
RLock lock = redisson.getLock("redission:test:lock:1");
final int[] count = {0};
int threads = 10;
ExecutorService pool = Executors.newFixedThreadPool(10);
CountDownLatch countDownLatch = new CountDownLatch(threads);
long start = System.currentTimeMillis();
for (int i = 0; i < threads; i++) {
pool.submit(() ->
{
for (int j = 0; j < 1000; j++) {
lock.lock();
count[0]++;
lock.unlock();
}
countDownLatch.countDown();
});
}
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("10个线程每个累加1000为：= " + count[0]);
//输出统计结果
float time = System.currentTimeMillis() - start;
System.out.println("运行的时长为：" + time);
System.out.println("每一次执行的时长为：" + time/count[0]);
}
}
运行的时长为：14172.0
每一次执行的时长为：1.4172
Redision锁 核心源码分析
Config config = new Config();
config.useSingleServer().setAddress("redis://172.29.1.180:5379").setPassword("a123456").setDatabase(0);
// 构造RedissonClient
RedissonClient redissonClient = Redisson.create(config);
// 设置锁定资源名称
RLock disLock = redissonClient.getLock("DISLOCK");
//尝试获取分布式锁
boolean isLock= disLock.tryLock(500, 15000, TimeUnit.MILLISECONDS);
if (isLock) {
try {
//TODO if get lock success, do something;
Thread.sleep(15000);
} catch (Exception e) {
} finally {
// 无论如何, 最后都要解锁
disLock.unlock();
}
}
1) "01a6d806-d282-4715-9bec-f51b9aa98110:1"
2) "1"
getLock()方法
public RLock getLock(String name) {
//默认创建的同步执行器, (存在异步执行器, 因为锁的获取和释放是有强一致性要求, 默认同步)
return new RedissonLock(this.connectionManager.getCommandExecutor(), name);
}
super(commandExecutor, name);
this.commandExecutor = commandExecutor;
//唯一ID
this.id = commandExecutor.getConnectionManager().getId();
//等待获取锁时间
this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
//ID + 锁名称
this.entryName = this.id + ":" + name;
//发布订阅
this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}
tryLock方法
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
current = System.currentTimeMillis();
RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.onComplete((res, e) -> {
if (e == null) {
unsubscribe(subscribeFuture, threadId);
}
});
}
acquireFailed(threadId);
return false;
}
try {
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
while (true) {
long currentTime = System.currentTimeMillis();
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
// waiting for message
currentTime = System.currentTimeMillis();
if (ttl >= 0 && ttl < time) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
}
} finally {
unsubscribe(subscribeFuture, threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}
tryAcquire()方法
tryLockInnerAsync
if (leaseTime != -1L) {
//进行异步获取锁
return this.tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
//尝试异步获取锁, 获取锁成功返回空, 否则返回锁剩余过期时间
RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
//ttlRemainingFuture 执行完成后触发此操作
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e == null) {
//ttlRemaining == null 代表获取了锁
//获取到锁后执行续时操作
if (ttlRemaining == null) {
this.scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
}
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
原理：加锁机制
Lua脚本的详解
关于 lua脚本的参数解释：
{
8743c9c0-0795-4907-87fd-6c719a6b4586:1 1
}
锁互斥机制
可重入加锁机制
lock.lock();
//业务代码
lock.lock();
//业务代码
lock.unlock();
lock.unlock();
{
8743c9c0-0795-4907-87fd-6c719a6b4586:1 2
}
释放锁机制
unlock 源码
public void unlock() {
try {
get(unlockAsync(Thread.currentThread().getId()));
} catch (RedisException e) {
if (e.getCause() instanceof IllegalMonitorStateException) {
throw (IllegalMonitorStateException) e.getCause();
} else {
throw e;
}
}
// Future<Void> future = unlockAsync();
// future.awaitUninterruptibly();
// if (future.isSuccess()) {
// return;
// }
// if (future.cause() instanceof IllegalMonitorStateException) {
// throw (IllegalMonitorStateException)future.cause();
// }
// throw commandExecutor.convertException(future);
}
unlockInnerAsync方法
原理：Redision 解锁机制
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
通过redis Channel 解锁订阅
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return;
}
// 订阅
RFuture<RedissonLockEntry> future = subscribe(threadId);
commandExecutor.syncSubscription(future);
try {
while (true) {
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (ttl >= 0) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().acquire();
}
}
} finally {
unsubscribe(future, threadId);
}
// get(lockAsync(leaseTime, unit));
}
protected static final LockPubSub PUBSUB = new LockPubSub();
protected RFuture<RedissonLockEntry> subscribe(long threadId) {
return PUBSUB.subscribe(getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}
protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) {
PUBSUB.unsubscribe(future.getNow(), getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}
watch dog自动延期机制
使用watchDog机制实现锁的续期
redisson watchdog 使用和原理
lockWatchdogTimeout（监控锁的看门狗超时，单位：毫秒）
30000
leaseTimeout
参数的情况。如果该看门狗未使用
lockWatchdogTimeout
去重新调整一个分布式锁的
lockWatchdogTimeout
超时，那么这个锁将变为失效状态。这个参数可以用来避免由Redisson客户端节点宕机或其他原因造成死锁的情况。
tryAcquireAsync原理
//如果指定了加锁时间，会直接去加锁
if (leaseTime != -1) {
return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
//没有指定加锁时间 会先进行加锁，并且默认时间就是 LockWatchdogTimeout的时间
//这个是异步操作 返回RFuture 类似netty中的future
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
//这里也是类似netty Future 的addListener，在future内容执行完成后执行
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired
if (ttlRemaining == null) {
//这里是定时执行 当前锁自动延期的动作
scheduleExpirationRenewal(threadId);
}
});
return ttlRemainingFuture;
}
RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry();
RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
this.renewExpiration();
}
}
renewExpiration执行延期动作
//从容器中去获取要被续期的锁
RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
//容器中没有要续期的锁，直接返回null
if (ee != null) {
//创建定时任务
//并且执行的时间为 30000/3 毫秒，也就是 10 秒后
Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
public void run(Timeout timeout) throws Exception {
//从容器中取出线程
RedissonLock.ExpirationEntry ent = (RedissonLock.ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName());
if (ent != null) {
Long threadId = ent.getFirstThreadId();
if (threadId != null) {
//Redis进行锁续期
//这个方法的作用其实底层也是去执行LUA脚本
RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
//同理去处理Redis续命结果
future.onComplete((res, e) -> {
if (e != null) {
RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e);
} else {
//如果成功续期，递归继续创建下一个 10S 后的任务
if (res) {
//递归继续创建下一个10S后的任务
RedissonLock.this.renewExpiration();
}
}
});
}
}
}
}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
}
renewExpirationAsync
return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.singletonList(getName()),
internalLockLeaseTime, getLockName(threadId));
}
watchLog总结
总结：Redis分布式锁过期怎么办？
说在最后：有问题找老架构取经
被裁之后， 空窗1年/空窗2年， 如何 起死回生 ？
被裁之后，100W 年薪 到手， 如何 人生逆袭？
100W案例，100W年薪的底层逻辑是什么？ 如何实现年薪百万？ 如何远离 中年危机？
如何 逆天改命，包含AI、大数据、golang、Java 等
关注职业救助站公众号，获取每天职业干货
助您实现职业转型、职业升级、极速上岸
---------------------------------
实现架构转型，再无中年危机
关注技术自由圈公众号，获取每天技术千货
一起成为牛逼的未来超级架构师
几十篇架构笔记、5000页面试宝典、20个技术圣经
请加尼恩个人微信 免费拿走
暗号，请在 公众号后台 发送消息：领电子书
如有收获，请点击底部的"在看"和"赞"，谢谢