Redisson如何解决Redis分布式锁提前释放问题
来源：脚本之家
时间：2023-01-07 12:04:52 215浏览 收藏
来到golang学习网的大家，相信都是编程学习爱好者，希望在这里学习数据库相关编程知识。下面本篇文章就来带大家聊聊《Redisson如何解决Redis分布式锁提前释放问题》，介绍一下释放、Redis分布式锁，希望对大家的知识积累有所帮助，助力实战开发！
前言：
在分布式场景下，相信你或多或少需要使用分布式锁来访问临界资源，或者控制耗时操作的并发性。
当然，实现分布式锁的方案也比较多，比如数据库、redis、zk 等等。本文主要结合一个线上案例，讲解 redis 分布式锁的相关实现。
一、问题描述：
某天线上出现了数据重复处理问题，经排查后发现，竟然是单次处理时间较长，redis 分布式锁提前释放导致相同请求并发处理。
其实，这是一个锁续约的问题，对于一把分布式锁，我们需要考虑，设置锁多长时间过期、出现异常如何释放锁？
以上问题便是本文要讨论的主题。
二、原因分析：
项目采用较简单的自定义 redis 分布式锁，为避免死锁定义默认过期时间 10s，如下：
override fun lock() {
while (true) {
//尝试获取锁
if (tryLock()) {
return
}
try {
Thread.sleep(10)
} catch (e: InterruptedException) {
e.printStackTrace()
}
}
}
override fun tryLock(): Boolean {
val value = getUniqueSign() // 随机串
val flag = redisTemplate!!.opsForValue().setIfAbsent(name, value, 10000, TimeUnit.MILLISECONDS)
if (flag != null && flag) {
VALUE_lOCAL.set(value)
INTO_NUM_LOCAL.set(if (INTO_NUM_LOCAL.get() != null) INTO_NUM_LOCAL.get() + 1 else 1)
return true
}
return false
}
缺乏对锁自动续期等实现。
三、解决方案：
1、思考：
针对这种场景，可以考虑的是如何给锁自动续期-当业务没有执行结束的情况下，当然也可以自定义实现 比如开一个后台线程定时的给这些拿到锁的线程续期。
Redisson 也正是基于这种思路实现自动续期的分布式锁，各种异常情况也考虑的更加完善，综合考虑采用 Redisson 的分布式锁解决方案优化。
2、Redisson简单配置：
@Configuration
@EnableConfigurationProperties(RedissonProperties::class)
class RedissonConfig {
@Bean
fun redissonClient(redissonProperties: RedissonProperties): RedissonClient {
val config = Config()
val singleServerConfig = redissonProperties.singleServerConfig!!
config.useSingleServer().setAddress(singleServerConfig.address)
.setDatabase(singleServerConfig.database)
.setUsername(singleServerConfig.username)
.setPassword(singleServerConfig.password)
.setConnectionPoolSize(singleServerConfig.connectionPoolSize)
.setConnectionMinimumIdleSize(singleServerConfig.connectionMinimumIdleSize)
.setConnectTimeout(singleServerConfig.connectTimeout)
.setIdleConnectionTimeout(singleServerConfig.idleConnectionTimeout)
.setRetryInterval(singleServerConfig.retryInterval)
.setRetryAttempts(singleServerConfig.retryAttempts)
.setTimeout(singleServerConfig.timeout)
return Redisson.create(config)
}
}
@ConfigurationProperties(prefix = "xxx.redisson")
class RedissonProperties {
var singleServerConfig: SingleServerConfig? = null
}
Redis 服务使用的腾讯云的哨兵模式架构，此架构对外开放一个代理地址访问，因此这里配置单机模式配置即可。
如果你是自己搭建的 redis 哨兵模式架构，需要按照文档配置相关必要参数
3、使用样例：
...
@Autowired
lateinit var redissonClient: RedissonClient
...
fun xxx() {
...
val lock = redissonClient.getLock("mylock")
lock.lock()
try {
...
} finally {
lock.unlock()
}
...
}
使用方式和JDK提供的锁是不是很像？是不是很简单？
正是Redisson这类优秀的开源产品的出现，才让我们将更多的时间投入到业务开发中...
四、源码分析
下面来看看 Redisson 对常规分布式锁的实现，主要分析 RedissonLock
1、lock加锁操作
@Override
public void lock() {
try {
lock(-1, null, false);
} catch (InterruptedException e) {
throw new IllegalStateException();
}
}
// 租约期限, 也就是expire时间, -1代表未设置 将使用系统默认的30s
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
// 尝试拿锁, 如果能拿到就直接返回
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return;
}
RFuture future = subscribe(threadId);
if (interruptibly) {
commandExecutor.syncSubscriptionInterrupted(future);
} else {
commandExecutor.syncSubscription(future);
}
// 如果拿不到锁就尝试一直轮循, 直到成功获取锁或者异常终止
try {
while (true) {
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
...
}
} finally {
unsubscribe(future, threadId);
}
}
1.1、tryAcquire
private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}
private RFuture tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
RFuture ttlRemainingFuture;
// 调用真正获取锁的操作
if (leaseTime != -1) {
ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
}
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired
// 这里是成功获取了锁, 尝试给锁续约
if (ttlRemaining == null) {
if (leaseTime != -1) {
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
// 通过lua脚本真正执行加锁的操作
RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
// 如果key不存在, 那正好, 直接set并设置过期时间
// 如果key存在, 就有两种情况需要考虑
// - 同一线程获取重入锁，直接将field(也就是getLockName(threadId))对应的value值+1
// - 不同线程竞争锁, 此次加锁失败, 并直接返回此key对应的过期时间
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hincrby', 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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}
1.2、续约
通过 scheduleExpirationRenewal 给锁续约
protected void scheduleExpirationRenewal(long threadId) {
ExpirationEntry entry = new ExpirationEntry();
ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
// 续约操作
renewExpiration();
}
}
private void renewExpiration() {
ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ee == null) {
return;
}
// 设置延迟任务task, 在时长internalLockLeaseTime/3之后执行, 定期给锁续期
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ent == null) {
return;
}
Long threadId = ent.getFirstThreadId();
if (threadId == null) {
return;
}
// 真正执行续期命令操作
RFuture future = renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
log.error("Can't update lock " + getRawName() + " expiration", e);
EXPIRATION_RENEWAL_MAP.remove(getEntryName());
return;
}
// 这次续期之后, 继续schedule自己, 达到持续续期的效果
if (res) {
// reschedule itself
renewExpiration();
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
// 所谓续期, 就是将expire过期时间再延长
protected RFuture renewExpirationAsync(long threadId) {
// 如果key以及当前线程存在, 则延长expire时间, 并返回1代表成功；否则返回0代表失败
return evalWriteAsync(getRawName(), 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(getRawName()),
internalLockLeaseTime, getLockName(threadId));
}
2、unlock解锁操作
public void unlock() {
try {
get(unlockAsync(Thread.currentThread().getId()));
} catch (RedisException e) {
...
}
}
public RFuture unlockAsync(long threadId) {
RPromise result = new RedissonPromise();
// 执行解锁操作
RFuture future = unlockInnerAsync(threadId);
// 操作成功之后做的事
future.onComplete((opStatus, e) -> {
// 取消续约task
cancelExpirationRenewal(threadId);
...
});
return result;
}
protected RFuture unlockInnerAsync(long threadId) {
// 如果key以及当前线程对应的记录已经不存在, 直接返回空
// 否在将field(也就是getLockName(threadId))对应的value减1
// - 如果减去1之后值还大于0, 那么重新延长过期时间
// - 如果减去之后值小于等于0, 那么直接删除key, 并发布订阅消息
return evalWriteAsync(getRawName(), 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.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
以上便是 redisson 客户端工具对 redis 分布式锁的加/解锁具体实现，主要解决了以下几个问题
1、死锁问题：设置过期时间
2、可重入问题：重入+1, 释放锁-1，当值=0时代表完全释放锁
3、续约问题：可解决锁提前释放问题
4、锁释放：谁加锁就由谁来释放
总结：
本文由一个线上问题做引子，通过 redis 分布式锁的常用实现方案，最终选定 redisson 的解决方案； 并分析 redisson 的具体实现细节
相关参考:
本篇关于《Redisson如何解决Redis分布式锁提前释放问题》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于数据库的相关知识，请关注golang学习网公众号！
声明：本文转载于：脚本之家 如有侵犯，请联系study_golang@163.com删除
相关阅读
更多>
-
182 收藏
-
367 收藏
-
348 收藏
-
302 收藏
-
214 收藏
最新阅读
更多>
-
342 收藏
-
361 收藏
-
159 收藏
-
164 收藏
-
221 收藏
-
156 收藏
课程推荐
更多>
-
- 前端进阶之JavaScript设计模式
- 设计模式是开发人员在软件开发过程中面临一般问题时的解决方案，代表了最佳的实践。本课程的主打内容包括JS常见设计模式以及具体应用场景，打造一站式知识长龙服务，适合有JS基础的同学学习。
- 立即学习 542次学习
-
- GO语言核心编程课程
- 本课程采用真实案例，全面具体可落地，从理论到实践，一步一步将GO核心编程技术、编程思想、底层实现融会贯通，使学习者贴近时代脉搏，做IT互联网时代的弄潮儿。
- 立即学习 507次学习
-
- 简单聊聊mysql8与网络通信
- 如有问题加微信：Le-studyg；在课程中，我们将首先介绍MySQL8的新特性，包括性能优化、安全增强、新数据类型等，帮助学生快速熟悉MySQL8的最新功能。接着，我们将深入解析MySQL的网络通信机制，包括协议、连接管理、数据传输等，让
- 立即学习 497次学习
-
- JavaScript正则表达式基础与实战
- 在任何一门编程语言中,正则表达式,都是一项重要的知识,它提供了高效的字符串匹配与捕获机制,可以极大的简化程序设计。
- 立即学习 487次学习
-
- 从零制作响应式网站—Grid布局
- 本系列教程将展示从零制作一个假想的网络科技公司官网，分为导航，轮播，关于我们，成功案例，服务流程，团队介绍，数据部分，公司动态，底部信息等内容区块。网站整体采用CSSGrid布局，支持响应式，有流畅过渡和展现动画。
- 立即学习 484次学习
评论列表
-
- 无情的白猫
- 受益颇多，一直没懂这个问题，但其实工作中常常有遇到...不过今天到这，帮助很大，总算是懂了，感谢大佬分享文章内容！
- 2023-03-20 09:02:01
-
- 啊啊啊啊
- 这篇博文真是及时雨啊，好细啊，受益颇多，码起来，关注大佬了！希望大佬能多写数据库相关的文章。
- 2023-02-10 13:51:33