-如有不严谨或者错误之处&＃xff0c;还望不吝赐教&＃xff0c;轻点怼&＃xff0c;人家还是个孩子&＃xff0c;嘤嘤嘤~

引言&＃xff1a;

最近项目上线的频率颇高&＃xff0c;连着几天加班熬夜&＃xff0c;身体有点吃不消精神也有些萎靡&＃xff0c;无奈业务方催的紧&＃xff0c;工期就在眼前只能硬着头皮上了。脑子浑浑噩噩的时候&＃xff0c;写的就不能叫代码&＃xff0c;可以直接叫做Bug。我就熬夜写了一个bug被骂惨了。

由于是做商城业务&＃xff0c;要频繁的对商品库存进行扣减&＃xff0c;应用是集群部署&＃xff0c;为避免并发造成库存超买超卖等问题&＃xff0c;采用 redis 分布式锁加以控制。本以为给扣库存的代码加上锁lock.tryLock就万事大吉了

/** * &＃64;author xiaofu * &＃64;description 扣减库存 * &＃64;date 2020/4/21 12:10 */ public String stockLock() { RLock lock &＃61; redissonClient.getLock("stockLock"); try { /** * 获取锁 */ if (lock.tryLock(10, TimeUnit.SECONDS)) { /** * 扣减库存 */ 。。。。。。 } else { LOGGER.info("未获取到锁业务结束.."); } } catch (Exception e) { LOGGER.info("处理异常", e); } return "ok"; }

结果业务代码执行完以后我忘了释放锁lock.unlock()&＃xff0c;导致redis线程池被打满&＃xff0c;redis服务大面积故障&＃xff0c;造成库存数据扣减混乱&＃xff0c;被领导一顿臭骂&＃xff0c;这个月绩效~ 哎·~。

随着 使用redis 锁的时间越长&＃xff0c;我发现 redis 锁的坑远比想象中要多。就算在面试题当中redis分布式锁的出镜率也比较高&＃xff0c;比如&＃xff1a;“用锁遇到过哪些问题&＃xff1f;” &＃xff0c;“又是如何解决的&＃xff1f;” 基本都是一套连招问出来的。

今天就分享一下我用redis 分布式锁的踩坑日记&＃xff0c;以及一些解决方案&＃xff0c;和大家一起共勉。

一、锁未被释放

这种情况是一种低级错误&＃xff0c;就是我上边犯的错&＃xff0c;由于当前线程 获取到redis 锁&＃xff0c;处理完业务后未及时释放锁&＃xff0c;导致其它线程会一直尝试获取锁阻塞&＃xff0c;例如&＃xff1a;用Jedis客户端会报如下的错误信息

redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool

redis线程池已经没有空闲线程来处理客户端命令。

解决的方法也很简单&＃xff0c;只要我们细心一点&＃xff0c;拿到锁的线程处理完业务及时释放锁&＃xff0c;如果是重入锁未拿到锁后&＃xff0c;线程可以释放当前连接并且sleep一段时间。

public void lock() { while (true) { boolean flag &＃61; this.getLock(key); if (flag) { TODO ......... } else { // 释放当前redis连接 redis.close(); // 休眠1000毫秒 sleep(1000); } } }

二、B的锁被A给释放了

我们知道Redis实现锁的原理在于 SETNX命令。当 key不存在时将 key的值设为 value &＃xff0c;返回值为 1&＃xff1b;若给定的 key 已经存在&＃xff0c;则 SETNX不做任何动作&＃xff0c;返回值为 0 。

SETNX key value

我们来设想一下这个场景&＃xff1a;A、B两个线程来尝试给key myLock加锁&＃xff0c;A线程先拿到锁(假如锁3秒后过期)&＃xff0c;B线程就在等待尝试获取锁&＃xff0c;到这一点毛病没有。

那如果此时业务逻辑比较耗时&＃xff0c;执行时间已经超过redis锁过期时间&＃xff0c;这时A线程的锁自动释放(删除key)&＃xff0c;B线程检测到myLock这个key不存在&＃xff0c;执行 SETNX命令也拿到了锁。

但是&＃xff0c;此时A线程执行完业务逻辑之后&＃xff0c;还是会去释放锁(删除key)&＃xff0c;这就导致B线程的锁被A线程给释放了。

为避免上边的情况&＃xff0c;一般我们在每个线程加锁时要带上自己独有的value值来标识&＃xff0c;只释放指定value的key&＃xff0c;否则就会出现释放锁混乱的场景。

三、数据库事务超时

emm~ 聊redis锁咋还扯到数据库事务上来了&＃xff1f;别着急往下看&＃xff0c;看下边这段代码&＃xff1a;

&＃64;Transaction public void lock() { while (true) { boolean flag &＃61; this.getLock(key); if (flag) { insert(); } }

给这个方法添加一个&＃64;Transaction注解开启事务&＃xff0c;如代码中抛出异常进行回滚&＃xff0c;要知道数据库事务可是有超时时间限制的&＃xff0c;并不会无条件的一直等一个耗时的数据库操作。

比如&＃xff1a;我们解析一个大文件&＃xff0c;再将数据存入到数据库&＃xff0c;如果执行时间太长&＃xff0c;就会导致事务超时自动回滚。

一旦你的key长时间获取不到锁&＃xff0c;获取锁等待的时间远超过数据库事务超时时间&＃xff0c;程序就会报异常。

一般为解决这种问题&＃xff0c;我们就需要将数据库事务改为手动提交、回滚事务。

&＃64;Autowired DataSourceTransactionManager dataSourceTransactionManager; &＃64;Transaction public void lock() { //手动开启事务 TransactionStatus transactionStatus &＃61; dataSourceTransactionManager.getTransaction(transactionDefinition); try { while (true) { boolean flag &＃61; this.getLock(key); if (flag) { insert(); //手动提交事务 dataSourceTransactionManager.commit(transactionStatus); } } } catch (Exception e) { //手动回滚事务 dataSourceTransactionManager.rollback(transactionStatus); } }

四、锁过期了&＃xff0c;业务还没执行完

这种情况和我们上边提到的第二种比较类似&＃xff0c;但解决思路上略有不同。

同样是redis分布式锁过期&＃xff0c;而业务逻辑没执行完的场景&＃xff0c;不过&＃xff0c;这里换一种思路想问题&＃xff0c;把redis锁的过期时间再弄长点不就解决了吗&＃xff1f;

那还是有问题&＃xff0c;我们可以在加锁的时候&＃xff0c;手动调长redis锁的过期时间&＃xff0c;可这个时间多长合适&＃xff1f;业务逻辑的执行时间是不可控的&＃xff0c;调的过长又会影响操作性能。

要是redis锁的过期时间能够自动续期就好了。

为了解决这个问题我们使用redis客户端redisson&＃xff0c;redisson很好的解决了redis在分布式环境下的一些棘手问题&＃xff0c;它的宗旨就是让使用者减少对Redis的关注&＃xff0c;将更多精力用在处理业务逻辑上。

redisson对分布式锁做了很好封装&＃xff0c;只需调用API即可。

RLock lock &＃61; redissonClient.getLock("stockLock");

redisson在加锁成功后&＃xff0c;会注册一个定时任务监听这个锁&＃xff0c;每隔10秒就去查看这个锁&＃xff0c;如果还持有锁&＃xff0c;就对过期时间进行续期。默认过期时间30秒。这个机制也被叫做&＃xff1a;“看门狗”&＃xff0c;这名字。。。

举例子&＃xff1a;假如加锁的时间是30秒&＃xff0c;过10秒检查一次&＃xff0c;一旦加锁的业务没有执行完&＃xff0c;就会进行一次续期&＃xff0c;把锁的过期时间再次重置成30秒。

通过分析下边redisson的源码实现可以发现&＃xff0c;不管是加锁、解锁、续约都是客户端把一些复杂的业务逻辑&＃xff0c;通过封装在Lua脚本中发送给redis&＃xff0c;保证这段复杂业务逻辑执行的原子性。

&＃64;Slf4j&＃64;Servicepublic class RedisDistributionLockPlus { /** * 加锁超时时间&＃xff0c;单位毫秒&＃xff0c; 即&＃xff1a;加锁时间内执行完操作&＃xff0c;如果未完成会有并发现象 */ private static final long DEFAULT_LOCK_TIMEOUT &＃61; 30; private static final long TIME_SECONDS_FIVE &＃61; 5 ; /** * 每个key的过期时间 {&＃64;link LockContent} */ private Map lockContentMap &＃61; new ConcurrentHashMap<>(512); /** * redis执行成功的返回 */ private static final Long EXEC_SUCCESS &＃61; 1L; /** * 获取锁lua脚本&＃xff0c; k1&＃xff1a;获锁key, k2&＃xff1a;续约耗时key, arg1:requestId&＃xff0c;arg2&＃xff1a;超时时间 */ private static final String LOCK_SCRIPT &＃61; "if redis.call(&＃39;exists&＃39;, KEYS[2]) &＃61;&＃61; 1 then ARGV[2] &＃61; math.floor(redis.call(&＃39;get&＃39;, KEYS[2]) &＃43; 10) end " &＃43; "if redis.call(&＃39;exists&＃39;, KEYS[1]) &＃61;&＃61; 0 then " &＃43; "local t &＃61; redis.call(&＃39;set&＃39;, KEYS[1], ARGV[1], &＃39;EX&＃39;, ARGV[2]) " &＃43; "for k, v in pairs(t) do " &＃43; "if v &＃61;&＃61; &＃39;OK&＃39; then return tonumber(ARGV[2]) end " &＃43; "end " &＃43; "return 0 end"; /** * 释放锁lua脚本, k1&＃xff1a;获锁key, k2&＃xff1a;续约耗时key, arg1:requestId&＃xff0c;arg2&＃xff1a;业务耗时 arg3: 业务开始设置的timeout */ private static final String UNLOCK_SCRIPT &＃61; "if redis.call(&＃39;get&＃39;, KEYS[1]) &＃61;&＃61; ARGV[1] then " &＃43; "local ctime &＃61; tonumber(ARGV[2]) " &＃43; "local biz_timeout &＃61; tonumber(ARGV[3]) " &＃43; "if ctime > 0 then " &＃43; "if redis.call(&＃39;exists&＃39;, KEYS[2]) &＃61;&＃61; 1 then " &＃43; "local avg_time &＃61; redis.call(&＃39;get&＃39;, KEYS[2]) " &＃43; "avg_time &＃61; (tonumber(avg_time) * 8 &＃43; ctime * 2)/10 " &＃43; "if avg_time >&＃61; biz_timeout - 5 then redis.call(&＃39;set&＃39;, KEYS[2], avg_time, &＃39;EX&＃39;, 24*60*60) " &＃43; "else redis.call(&＃39;del&＃39;, KEYS[2]) end " &＃43; "elseif ctime > biz_timeout -5 then redis.call(&＃39;set&＃39;, KEYS[2], ARGV[2], &＃39;EX&＃39;, 24*60*60) end " &＃43; "end " &＃43; "return redis.call(&＃39;del&＃39;, KEYS[1]) " &＃43; "else return 0 end"; /** * 续约lua脚本 */ private static final String RENEW_SCRIPT &＃61; "if redis.call(&＃39;get&＃39;, KEYS[1]) &＃61;&＃61; ARGV[1] then return redis.call(&＃39;expire&＃39;, KEYS[1], ARGV[2]) else return 0 end"; private final StringRedisTemplate redisTemplate; public RedisDistributionLockPlus(StringRedisTemplate redisTemplate) { this.redisTemplate &＃61; redisTemplate; ScheduleTask task &＃61; new ScheduleTask(this, lockContentMap); // 启动定时任务 ScheduleExecutor.schedule(task, 1, 1, TimeUnit.SECONDS); } /** * 加锁 * 取到锁加锁&＃xff0c;取不到锁一直等待知道获得锁 * * &＃64;param lockKey * &＃64;param requestId 全局唯一 * &＃64;param expire 锁过期时间, 单位秒 * &＃64;return */ public boolean lock(String lockKey, String requestId, long expire) { log.info("开始执行加锁, lockKey &＃61;{}, requestId&＃61;{}", lockKey, requestId); for (; ; ) { // 判断是否已经有线程持有锁&＃xff0c;减少redis的压力 LockContent lockContentOld &＃61; lockContentMap.get(lockKey); boolean unLocked &＃61; null &＃61;&＃61; lockContentOld; // 如果没有被锁&＃xff0c;就获取锁 if (unLocked) { long startTime &＃61; System.currentTimeMillis(); // 计算超时时间 long bizExpire &＃61; expire &＃61;&＃61; 0L ? DEFAULT_LOCK_TIMEOUT : expire; String lockKeyRenew &＃61; lockKey &＃43; "_renew"; RedisScript script &＃61; RedisScript.of(LOCK_SCRIPT, Long.class); List keys &＃61; new ArrayList<>(); keys.add(lockKey); keys.add(lockKeyRenew); Long lockExpire &＃61; redisTemplate.execute(script, keys, requestId, Long.toString(bizExpire)); if (null !&＃61; lockExpire && lockExpire > 0) { // 将锁放入map LockContent lockContent &＃61; new LockContent(); lockContent.setStartTime(startTime); lockContent.setLockExpire(lockExpire); lockContent.setExpireTime(startTime &＃43; lockExpire * 1000); lockContent.setRequestId(requestId); lockContent.setThread(Thread.currentThread()); lockContent.setBizExpire(bizExpire); lockContent.setLockCount(1); lockContentMap.put(lockKey, lockContent); log.info("加锁成功, lockKey &＃61;{}, requestId&＃61;{}", lockKey, requestId); return true; } } // 重复获取锁&＃xff0c;在线程池中由于线程复用&＃xff0c;线程相等并不能确定是该线程的锁 if (Thread.currentThread() &＃61;&＃61; lockContentOld.getThread() && requestId.equals(lockContentOld.getRequestId())){ // 计数 &＃43;1 lockContentOld.setLockCount(lockContentOld.getLockCount()&＃43;1); return true; } // 如果被锁或获取锁失败&＃xff0c;则等待100毫秒 try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { // 这里用lombok 有问题 log.error("获取redis 锁失败, lockKey &＃61;{}, requestId&＃61;{}", lockKey, requestId, e); return false; } } } /** * 解锁 * * &＃64;param lockKey * &＃64;param lockValue */ public boolean unlock(String lockKey, String lockValue) { String lockKeyRenew &＃61; lockKey &＃43; "_renew"; LockContent lockContent &＃61; lockContentMap.get(lockKey); long consumeTime; if (null &＃61;&＃61; lockContent) { consumeTime &＃61; 0L; } else if (lockValue.equals(lockContent.getRequestId())) { int lockCount &＃61; lockContent.getLockCount(); // 每次释放锁&＃xff0c; 计数 -1&＃xff0c;减到0时删除redis上的key if (--lockCount > 0) { lockContent.setLockCount(lockCount); return false; } consumeTime &＃61; (System.currentTimeMillis() - lockContent.getStartTime()) / 1000; } else { log.info("释放锁失败&＃xff0c;不是自己的锁。"); return false; } // 删除已完成key&＃xff0c;先删除本地缓存&＃xff0c;减少redis压力, 分布式锁&＃xff0c;只有一个&＃xff0c;所以这里不加锁 lockContentMap.remove(lockKey); RedisScript script &＃61; RedisScript.of(UNLOCK_SCRIPT, Long.class); List keys &＃61; new ArrayList<>(); keys.add(lockKey); keys.add(lockKeyRenew); Long result &＃61; redisTemplate.execute(script, keys, lockValue, Long.toString(consumeTime), Long.toString(lockContent.getBizExpire())); return EXEC_SUCCESS.equals(result); } /** * 续约 * * &＃64;param lockKey * &＃64;param lockContent * &＃64;return true:续约成功&＃xff0c;false:续约失败(1、续约期间执行完成&＃xff0c;锁被释放 2、不是自己的锁&＃xff0c;3、续约期间锁过期了(未解决)) */ public boolean renew(String lockKey, LockContent lockContent) { // 检测执行业务线程的状态 Thread.State state &＃61; lockContent.getThread().getState(); if (Thread.State.TERMINATED &＃61;&＃61; state) { log.info("执行业务的线程已终止,不再续约 lockKey &＃61;{}, lockContent&＃61;{}", lockKey, lockContent); return false; } String requestId &＃61; lockContent.getRequestId(); long timeOut &＃61; (lockContent.getExpireTime() - lockContent.getStartTime()) / 1000; RedisScript script &＃61; RedisScript.of(RENEW_SCRIPT, Long.class); List keys &＃61; new ArrayList<>(); keys.add(lockKey); Long result &＃61; redisTemplate.execute(script, keys, requestId, Long.toString(timeOut)); log.info("续约结果&＃xff0c;True成功&＃xff0c;False失败 lockKey &＃61;{}, result&＃61;{}", 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 &＃61; unit.toMillis(initialDelay); long period_ &＃61; unit.toMillis(period); // 定时执行 new Timer("Lock-Renew-Task").schedule(task, delay, period_); } } static class ScheduleTask extends TimerTask { private final RedisDistributionLockPlus redisDistributionLock; private final Map lockContentMap; public ScheduleTask(RedisDistributionLockPlus redisDistributionLock, Map lockContentMap) { this.redisDistributionLock &＃61; redisDistributionLock; this.lockContentMap &＃61; lockContentMap; } &＃64;Override public void run() { if (lockContentMap.isEmpty()) { return; } Set> entries &＃61; lockContentMap.entrySet(); for (Map.Entry entry : entries) { String lockKey &＃61; entry.getKey(); LockContent lockContent &＃61; entry.getValue(); long expireTime &＃61; lockContent.getExpireTime(); // 减少线程池中任务数量 if ((expireTime - System.currentTimeMillis())/ 1000 { boolean renew &＃61; redisDistributionLock.renew(lockKey, lockContent); if (renew) { long expireTimeNew &＃61; lockContent.getStartTime() &＃43; (expireTime - lockContent.getStartTime()) * 2 - TIME_SECONDS_FIVE * 1000; lockContent.setExpireTime(expireTimeNew); } else { // 续约失败&＃xff0c;说明已经执行完 OR redis 出现问题 lockContentMap.remove(lockKey); } }); } } } }}

五、redis主从复制的坑

redis高可用最常见的方案就是主从复制(master-slave)&＃xff0c;这种模式也给redis分布式锁挖了一坑。

redis cluster集群环境下&＃xff0c;假如现在A客户端想要加锁&＃xff0c;它会根据路由规则选择一台master节点写入key mylock&＃xff0c;在加锁成功后&＃xff0c;master节点会把key异步复制给对应的slave节点。

如果此时redis master节点宕机&＃xff0c;为保证集群可用性&＃xff0c;会进行主备切换&＃xff0c;slave变为了redis master。B客户端在新的master节点上加锁成功&＃xff0c;而A客户端也以为自己还是成功加了锁的。

此时就会导致同一时间内多个客户端对一个分布式锁完成了加锁&＃xff0c;导致各种脏数据的产生。

至于解决办法嘛&＃xff0c;目前看还没有什么根治的方法&＃xff0c;只能尽量保证机器的稳定性&＃xff0c;减少发生此事件的概率。

总结

上面就是我在使用Redis 分布式锁时遇到的一些坑&＃xff0c;有点小感慨&＃xff0c;经常用一个方法填上这个坑&＃xff0c;没多久就发现另一个坑又出来了&＃xff0c;其实根本没有什么十全十美的解决方案&＃xff0c;哪有什么银弹&＃xff0c;只不过是在权衡利弊后&＃xff0c;选一个在接受范围内的折中方案而已。

作者&＃xff1a;程序员内点事
链接&＃xff1a;https://juejin.im/post/5e9ff1766fb9a03c3b2bddab
来源&＃xff1a;掘金