redis 分布式锁的 5个坑,真是又大又深
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了redis 分布式锁的 5个坑,真是又大又深,小编现在分享给大家,供广大互联网技能从业者学习和参考。文章包含13211字,纯文字阅读大概需要19分钟。
内容图文
![redis 分布式锁的 5个坑,真是又大又深](/upload/InfoBanner/zyjiaocheng/875/94c17573cb4648f2b997e23fb1f51303.jpg)
引言
最近项目上线的频率颇高,连着几天加班熬夜,身体有点吃不消精神也有些萎靡,无奈业务方催的紧,工期就在眼前只能硬着头皮上了。脑子浑浑噩噩的时候,写的就不能叫代码,可以直接叫做Bug
。我就熬夜写了一个bug
被骂惨了。
由于是做商城业务,要频繁的对商品库存进行扣减,应用是集群部署,为避免并发造成库存超买超卖
等问题,采用 redis
分布式锁加以控制。本以为给扣库存的代码加上锁lock.tryLock
就万事大吉了
/**
* @author xiaofu
* @description 扣减库存
* @date 2020/4/21 12:10
*/
public String stockLock() {
RLock lock = redissonClient.getLock("stockLock");
try {
/**
* 获取锁
*/
if (lock.tryLock(10, TimeUnit.SECONDS)) {
/**
* 查询库存数
*/
Integer stock = Integer.valueOf(stringRedisTemplate.opsForValue().get("stockCount"));
/**
* 扣减库存
*/
if (stock > 0) {
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stockCount", stock.toString());
LOGGER.info("库存扣减成功,剩余库存数量:{}", stock);
} else {
LOGGER.info("库存不足~");
}
} else {
LOGGER.info("未获取到锁业务结束..");
}
} catch (Exception e) {
LOGGER.info("处理异常", e);
} finally {
lock.unlock();
}
return "ok";
}
结果业务代码执行完以后我忘了释放锁lock.unlock()
,导致redis
线程池被打满,redis
服务大面积故障,造成库存数据扣减混乱,被领导一顿臭骂,这个月绩效~ 哎·~。
随着 使用redis
锁的时间越长,我发现 redis
锁的坑远比想象中要多。就算在面试题当中redis
分布式锁的出镜率也比较高,比如:“用锁遇到过哪些问题?” ,“又是如何解决的?” 基本都是一套连招问出来的。
今天就分享一下我用redis
分布式锁的踩坑日记,以及一些解决方案,和大家一起共勉。
一、锁未被释放
这种情况是一种低级错误,就是我上边犯的错,由于当前线程 获取到redis
锁,处理完业务后未及时释放锁,导致其它线程会一直尝试获取锁阻塞,例如:用Jedis
客户端会报如下的错误信息
redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool
redis线程池
已经没有空闲线程来处理客户端命令。
解决的方法也很简单,只要我们细心一点,拿到锁的线程处理完业务及时释放锁,如果是重入锁未拿到锁后,线程可以释放当前连接并且sleep
一段时间。
public void lock() {
while (true) {
boolean flag = this.getLock(key);
if (flag) {
TODO .........
} else {
// 释放当前redis连接
redis.close();
// 休眠1000毫秒
sleep(1000);
}
}
}
二、B的锁被A给释放了
我们知道Redis
实现锁的原理在于 SETNX
命令。当 key
不存在时将 key
的值设为 value
,返回值为 1
;若给定的 key
已经存在,则 SETNX
不做任何动作,返回值为 0
。
SETNX key value
我们来设想一下这个场景:A
、B
两个线程来尝试给key
myLock
加锁,A线程
先拿到锁(假如锁3秒
后过期),B线程
就在等待尝试获取锁,到这一点毛病没有。
那如果此时业务逻辑比较耗时,执行时间已经超过redis
锁过期时间,这时A线程
的锁自动释放(删除key
),B线程
检测到myLock
这个key
不存在,执行 SETNX
命令也拿到了锁。
但是,此时A线程
执行完业务逻辑之后,还是会去释放锁(删除key
),这就导致B线程
的锁被A线程
给释放了。
为避免上边的情况,一般我们在每个线程加锁时要带上自己独有的value
值来标识,只释放指定value
的key
,否则就会出现释放锁混乱的场景。
三、数据库事务超时
emm~ 聊redis
锁咋还扯到数据库事务上来了?别着急往下看,看下边这段代码:
@Transaction
public void lock() {
while (true) {
boolean flag = this.getLock(key);
if (flag) {
insert();
}
}
}
给这个方法添加一个@Transaction
注解开启事务,如代码中抛出异常进行回滚,要知道数据库事务可是有超时时间限制的,并不会无条件的一直等一个耗时的数据库操作。
比如:我们解析一个大文件,再将数据存入到数据库,如果执行时间太长,就会导致事务超时自动回滚。
一旦你的key
长时间获取不到锁,获取锁等待的时间
远超过数据库事务超时时间
,程序就会报异常。
一般为解决这种问题,我们就需要将数据库事务改为手动提交、回滚事务。
@Autowired
DataSourceTransactionManager dataSourceTransactionManager;
@Transaction
public void lock() {
//手动开启事务
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
try {
while (true) {
boolean flag = this.getLock(key);
if (flag) {
insert();
//手动提交事务
dataSourceTransactionManager.commit(transactionStatus);
}
}
} catch (Exception e) {
//手动回滚事务
dataSourceTransactionManager.rollback(transactionStatus);
}
}
四、锁过期了,业务还没执行完
这种情况和我们上边提到的第二种比较类似,但解决思路上略有不同。
同样是redis
分布式锁过期,而业务逻辑没执行完的场景,不过,这里换一种思路想问题,把redis
锁的过期时间再弄长点不就解决了吗?
那还是有问题,我们可以在加锁的时候,手动调长redis
锁的过期时间,可这个时间多长合适?业务逻辑的执行时间是不可控的,调的过长又会影响操作性能。
要是redis
锁的过期时间能够自动续期就好了。
为了解决这个问题我们使用redis
客户端redisson
,redisson
很好的解决了redis
在分布式环境下的一些棘手问题,它的宗旨就是让使用者减少对Redis
的关注,将更多精力用在处理业务逻辑上。
redisson
对分布式锁做了很好封装,只需调用API
即可。
RLock lock = redissonClient.getLock("stockLock");
redisson
在加锁成功后,会注册一个定时任务监听这个锁,每隔10秒就去查看这个锁,如果还持有锁,就对过期时间
进行续期。默认过期时间30秒。这个机制也被叫做:“看门狗
”,这名字。。。
举例子:假如加锁的时间是30秒,过10秒检查一次,一旦加锁的业务没有执行完,就会进行一次续期,把锁的过期时间再次重置成30秒。
通过分析下边redisson
的源码实现可以发现,不管是加锁
、解锁
、续约
都是客户端把一些复杂的业务逻辑,通过封装在Lua
脚本中发送给redis
,保证这段复杂业务逻辑执行的原子性
。
@Slf4j
@Service
public class RedisDistributionLockPlus {
/**
* 加锁超时时间,单位毫秒, 即:加锁时间内执行完操作,如果未完成会有并发现象
*/
private static final long DEFAULT_LOCK_TIMEOUT = 30;
private static final long TIME_SECONDS_FIVE = 5 ;
/**
* 每个key的过期时间 {@link LockContent}
*/
private Map<String, LockContent> lockContentMap = new ConcurrentHashMap<>(512);
/**
* redis执行成功的返回
*/
private static final Long EXEC_SUCCESS = 1L;
/**
* 获取锁lua脚本, k1:获锁key, k2:续约耗时key, arg1:requestId,arg2:超时时间
*/
private static final String LOCK_SCRIPT = "if redis.call('exists', KEYS[2]) == 1 then ARGV[2] = math.floor(redis.call('get', KEYS[2]) + 10) end " +
"if redis.call('exists', KEYS[1]) == 0 then " +
"local t = redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) " +
"for k, v in pairs(t) do " +
"if v == 'OK' then return tonumber(ARGV[2]) end " +
"end " +
"return 0 end";
/**
* 释放锁lua脚本, k1:获锁key, k2:续约耗时key, arg1:requestId,arg2:业务耗时 arg3: 业务开始设置的timeout
*/
private static final String UNLOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
"local ctime = tonumber(ARGV[2]) " +
"local biz_timeout = tonumber(ARGV[3]) " +
"if ctime > 0 then " +
"if redis.call('exists', KEYS[2]) == 1 then " +
"local avg_time = redis.call('get', KEYS[2]) " +
"avg_time = (tonumber(avg_time) * 8 + ctime * 2)/10 " +
"if avg_time >= biz_timeout - 5 then redis.call('set', KEYS[2], avg_time, 'EX', 24*60*60) " +
"else redis.call('del', KEYS[2]) end " +
"elseif ctime > biz_timeout -5 then redis.call('set', KEYS[2], ARGV[2], 'EX', 24*60*60) end " +
"end " +
"return redis.call('del', KEYS[1]) " +
"else return 0 end";
/**
* 续约lua脚本
*/
private static final String RENEW_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end";
private final StringRedisTemplate redisTemplate;
public RedisDistributionLockPlus(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
ScheduleTask task = new ScheduleTask(this, lockContentMap);
// 启动定时任务
ScheduleExecutor.schedule(task, 1, 1, TimeUnit.SECONDS);
}
/**
* 加锁
* 取到锁加锁,取不到锁一直等待知道获得锁
*
* @param lockKey
* @param requestId 全局唯一
* @param expire 锁过期时间, 单位秒
* @return
*/
public boolean lock(String lockKey, String requestId, long expire) {
log.info("开始执行加锁, lockKey ={}, requestId={}", lockKey, requestId);
for (; ; ) {
// 判断是否已经有线程持有锁,减少redis的压力
LockContent lockContentOld = lockContentMap.get(lockKey);
boolean unLocked = null == lockContentOld;
// 如果没有被锁,就获取锁
if (unLocked) {
long startTime = System.currentTimeMillis();
// 计算超时时间
long bizExpire = expire == 0L ? DEFAULT_LOCK_TIMEOUT : expire;
String lockKeyRenew = lockKey + "_renew";
RedisScript<Long> script = RedisScript.of(LOCK_SCRIPT, Long.class);
List<String> keys = new ArrayList<>();
keys.add(lockKey);
keys.add(lockKeyRenew);
Long lockExpire = redisTemplate.execute(script, keys, requestId, Long.toString(bizExpire));
if (null != lockExpire && lockExpire > 0) {
// 将锁放入map
LockContent lockContent = new LockContent();
lockContent.setStartTime(startTime);
lockContent.setLockExpire(lockExpire);
lockContent.setExpireTime(startTime + lockExpire * 1000);
lockContent.setRequestId(requestId);
lockContent.setThread(Thread.currentThread());
lockContent.setBizExpire(bizExpire);
lockContent.setLockCount(1);
lockContentMap.put(lockKey, lockContent);
log.info("加锁成功, lockKey ={}, requestId={}", lockKey, requestId);
return true;
}
}
// 重复获取锁,在线程池中由于线程复用,线程相等并不能确定是该线程的锁
if (Thread.currentThread() == lockContentOld.getThread()
&& requestId.equals(lockContentOld.getRequestId())){
// 计数 +1
lockContentOld.setLockCount(lockContentOld.getLockCount()+1);
return true;
}
// 如果被锁或获取锁失败,则等待100毫秒
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
// 这里用lombok 有问题
log.error("获取redis 锁失败, lockKey ={}, requestId={}", lockKey, requestId, e);
return false;
}
}
}
/**
* 解锁
*
* @param lockKey
* @param lockValue
*/
public boolean unlock(String lockKey, String lockValue) {
String lockKeyRenew = lockKey + "_renew";
LockContent lockContent = lockContentMap.get(lockKey);
long consumeTime;
if (null == lockContent) {
consumeTime = 0L;
} else if (lockValue.equals(lockContent.getRequestId())) {
int lockCount = lockContent.getLockCount();
// 每次释放锁, 计数 -1,减到0时删除redis上的key
if (--lockCount > 0) {
lockContent.setLockCount(lockCount);
return false;
}
consumeTime = (System.currentTimeMillis() - lockContent.getStartTime()) / 1000;
} else {
log.info("释放锁失败,不是自己的锁。");
return false;
}
// 删除已完成key,先删除本地缓存,减少redis压力, 分布式锁,只有一个,所以这里不加锁
lockContentMap.remove(lockKey);
RedisScript<Long> script = RedisScript.of(UNLOCK_SCRIPT, Long.class);
List<String> keys = new ArrayList<>();
keys.add(lockKey);
keys.add(lockKeyRenew);
Long result = redisTemplate.execute(script, keys, lockValue, Long.toString(consumeTime),
Long.toString(lockContent.getBizExpire()));
return EXEC_SUCCESS.equals(result);
}
/**
* 续约
*
* @param lockKey
* @param lockContent
* @return true:续约成功,false:续约失败(1、续约期间执行完成,锁被释放 2、不是自己的锁,3、续约期间锁过期了(未解决))
*/
public boolean renew(String lockKey, LockContent lockContent) {
// 检测执行业务线程的状态
Thread.State state = lockContent.getThread().getState();
if (Thread.State.TERMINATED == state) {
log.info("执行业务的线程已终止,不再续约 lockKey ={}, lockContent={}", lockKey, lockContent);
return false;
}
String requestId = lockContent.getRequestId();
long timeOut = (lockContent.getExpireTime() - lockContent.getStartTime()) / 1000;
RedisScript<Long> script = RedisScript.of(RENEW_SCRIPT, Long.class);
List<String> keys = new ArrayList<>();
keys.add(lockKey);
Long result = redisTemplate.execute(script, keys, requestId, Long.toString(timeOut));
log.info("续约结果,True成功,False失败 lockKey ={}, result={}", lockKey, EXEC_SUCCESS.equals(result));
return EXEC_SUCCESS.equals(result);
}
static class ScheduleExecutor {
public static void schedule(ScheduleTask task, long initialDelay, long period, TimeUnit unit) {
long delay = unit.toMillis(initialDelay);
long period_ = unit.toMillis(period);
// 定时执行
new Timer("Lock-Renew-Task").schedule(task, delay, period_);
}
}
static class ScheduleTask extends TimerTask {
private final RedisDistributionLockPlus redisDistributionLock;
private final Map<String, LockContent> lockContentMap;
public ScheduleTask(RedisDistributionLockPlus redisDistributionLock, Map<String, LockContent> lockContentMap) {
this.redisDistributionLock = redisDistributionLock;
this.lockContentMap = lockContentMap;
}
@Override
public void run() {
if (lockContentMap.isEmpty()) {
return;
}
Set<Map.Entry<String, LockContent>> entries = lockContentMap.entrySet();
for (Map.Entry<String, LockContent> entry : entries) {
String lockKey = entry.getKey();
LockContent lockContent = entry.getValue();
long expireTime = lockContent.getExpireTime();
// 减少线程池中任务数量
if ((expireTime - System.currentTimeMillis())/ 1000 < TIME_SECONDS_FIVE) {
//线程池异步续约
ThreadPool.submit(() -> {
boolean renew = redisDistributionLock.renew(lockKey, lockContent);
if (renew) {
long expireTimeNew = lockContent.getStartTime() + (expireTime - lockContent.getStartTime()) * 2 - TIME_SECONDS_FIVE * 1000;
lockContent.setExpireTime(expireTimeNew);
} else {
// 续约失败,说明已经执行完 OR redis 出现问题
lockContentMap.remove(lockKey);
}
});
}
}
}
}
}
五、redis主从复制的坑
redis
高可用最常见的方案就是主从复制
(master-slave),这种模式也给redis分布式锁
挖了一坑。
redis cluster
集群环境下,假如现在A客户端
想要加锁,它会根据路由规则选择一台master
节点写入key
mylock
,在加锁成功后,master
节点会把key
异步复制给对应的slave
节点。
如果此时redis master
节点宕机,为保证集群可用性,会进行主备切换
,slave
变为了redis master
。B客户端
在新的master
节点上加锁成功,而A客户端
也以为自己还是成功加了锁的。
此时就会导致同一时间内多个客户端对一个分布式锁完成了加锁,导致各种脏数据的产生。
至于解决办法嘛,目前看还没有什么根治的方法,只能尽量保证机器的稳定性,减少发生此事件的概率。
总结
上面就是我在使用Redis
分布式锁时遇到的一些坑,有点小感慨,经常用一个方法填上这个坑,没多久就发现另一个坑又出来了,其实根本没有什么十全十美的解决方案,哪有什么银弹,只不过是在权衡利弊后,选一个在接受范围内的折中方案而已。
小福利:
有一些付费课程 ,嘘~,免费 送给小伙伴们。关注我的公号,回复【666】,无套路自行领取哦
内容总结
以上是互联网集市为您收集整理的redis 分布式锁的 5个坑,真是又大又深全部内容,希望文章能够帮你解决redis 分布式锁的 5个坑,真是又大又深所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错,欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 gblab@vip.qq.com 举报,一经查实,本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。