网站建设预算申请,网页上的视频怎么保存到本地,绥化市建设工程网站招投标,公司app开发收费价目表前言
本文将讨论的做一个高并发场景下避不开的话题#xff0c;即redis分布式锁。比如在淘宝 的秒杀场景、热点新闻和热搜排行榜等。可见分布式锁是一个程序员面向高级的一门必修课#xff0c;下面请跟着本篇文章好好学习。
redis分布式锁有哪些面试题
1.Redis做分布式的时…前言
本文将讨论的做一个高并发场景下避不开的话题即redis分布式锁。比如在淘宝 的秒杀场景、热点新闻和热搜排行榜等。可见分布式锁是一个程序员面向高级的一门必修课下面请跟着本篇文章好好学习。
redis分布式锁有哪些面试题
1.Redis做分布式的时候需要注意什么问题 2.你们公司自己实现的分布式锁是否用的setnx命令实现?这个是最合适的吗?你如何考虑分布式锁的可重入问题? 3.如果Redis是单点部署的会带来什么问题准备怎么解决单点问题呢 Redis集群模式下比如主从模式下CAP方面有没有什么问题
1.分布式锁是什么
1.2.锁的种类介绍
锁的种类锁的概念单机单机版同一个JVM虚拟机内synchronized或者lock接口分布式分布式多个不同的java虚拟机单机的线程锁机制不再起作用了资源类在不同的服务器之间共享了。
1.2一个正经的分布式锁具有哪些刚需 独占性任何时刻只能有且仅有一个线程持有 高可用若redis集群环境下不能因为某个节点挂了而出现获取锁或者释放锁失败。高并发请求下依旧能够保证良好使用。 防止死锁杜绝死锁必须有超时控制或者撤销操作有个兜底终止跳出方案 不乱抢防止张冠李戴不能私下uolock别人的锁只能自己加锁自己释放自己约的锁自己要释放可以设置过期时间或者业务代码执行完毕以后删除对一个的锁。 可重入同一个节点的同一个线程如果获得锁之后他也可以再次获得这个锁。
1.3 redis分布式锁
setnx key values1.4 java实现分布式锁的案例
先来个乞丐版的分布式锁并没有遵循上面五大原则。然后慢慢进行优化乞丐版分布锁案例如下代码所示
public String sale() {String resMessgae ;String key luojiaRedisLocak;String uuidValue IdUtil.simpleUUID() : Thread.currentThread().getId();Boolean flag stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue);// 抢不到的线程继续重试if (!flag) {// 线程休眠20毫秒进行递归重试try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}sale();} else {try {// 1 抢锁成功查询库存信息String result stringRedisTemplate.opsForValue().get(inventory01);// 2 判断库存书否足够Integer inventoryNum result null ? 0 : Integer.parseInt(result);// 3 扣减库存每次减少一个库存if (inventoryNum 0) {stringRedisTemplate.opsForValue().set(inventory01, String.valueOf(--inventoryNum));resMessgae 成功卖出一个商品库存剩余 inventoryNum \t 服务端口号 port;log.info(resMessgae);} else {resMessgae 商品已售罄。 \t 服务端口号 port;log.info(resMessgae);}} finally {stringRedisTemplate.delete(key);}}return resMessgae;
}请看看以上代码有哪些问题既没有删除过期时间 也没有判断redis获取的redis值进行删除有可能删除错锁。如果进一步优化可以redis可以存一个流水号业务代码执行完了以后判断流水号是否相等然后进行删除。可重入问题可以通过递归实现重试但是依旧有问题手工设置5000个线程来抢占锁压测OK但是容易导致StackOverflowError在高并发不推荐使用需要进一步完善。改进获取重试方法代码如下所示
public String sale() {String resMessgae ;String key luojiaRedisLocak;// 标记线程id知道使哪个线程在执行String uuidValue IdUtil.simpleUUID() : Thread.currentThread().getId();// 不用递归了高并发容易出错我们用自旋代替递归方法重试调用也不用if用while代替while (!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue)) {// 线程休眠20毫秒进行递归重试try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}}try {// 1 抢锁成功查询库存信息String result stringRedisTemplate.opsForValue().get(inventory01);// 2 判断库存书否足够Integer inventoryNum result null ? 0 : Integer.parseInt(result);// 3 扣减库存每次减少一个库存if (inventoryNum 0) {stringRedisTemplate.opsForValue().set(inventory01, String.valueOf(--inventoryNum));resMessgae 成功卖出一个商品库存剩余 inventoryNum \t 服务端口号 port;log.info(resMessgae);} else {resMessgae 商品已售罄。 \t 服务端口号 port;log.info(resMessgae);}} finally {stringRedisTemplate.delete(key);}return resMessgae;
}为了防止出现死锁需要给锁设置过期时关键点在于过期时间设置以避免代码异常出现而该线程持续占有该锁。其java代码如下所示
while (!stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue, 30L, TimeUnit.SECONDS)) {// 线程休眠20毫秒进行递归重试try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}}为了防止误删key在执行完了业务代码以后需要删掉锁在try-catch-finally 中添加如下删除锁的代码 try {//和上一个代码块重复省略掉了} finally {// v5.0 改进点判断加锁与解锁是不同客户端自己只能删除自己的锁不误删别人的锁if (stringRedisTemplate.opsForValue().get(key).equalsIgnoreCase(uuidValue)) {stringRedisTemplate.delete(key);}}1.5 优化分布式锁
本次优化主要解决的问题有宕机防止死锁、防止误删key、Lua保证原子性。设置 过期时间的同时当业务执行时间大于过期时间自动续锁功能等。java代码如下所示 自动续锁的Lua脚本
// 自动续期的LUA脚本
if redis.call(hexists, KEYS[1], ARGV[1]) 1 thenreturn redis.call(expire, KEYS[1], ARGV[2])
elsereturn 0
end新增续锁功能java代码如下所示
package com.luojia.redislock.mylock;import cn.hutool.core.util.IdUtil;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;/*** 自研的分布式锁实现了Lock接口*/
public class RedisDistributedLock implements Lock {private StringRedisTemplate stringRedisTemplate;private String lockName; // KEYS[1]private String uuidValule; // ARGV[1]private long expireTime; // ARGV[2]public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName) {this.stringRedisTemplate stringRedisTemplate;this.lockName lockName;this.uuidValule IdUtil.simpleUUID() : Thread.currentThread().getId();this.expireTime 50L;}Overridepublic void lock() {tryLock();}Overridepublic boolean tryLock() {try {tryLock(-1L, TimeUnit.SECONDS);} catch (InterruptedException e) {e.printStackTrace();}return false;}Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {if (-1 time) {String script if redis.call(exists, KEYS[1]) 0 or redis.call(hexists, KEYS[1], ARGV[1]) 1 then redis.call(hincrby, KEYS[1], ARGV[1], 1) redis.call(expire, KEYS[1], ARGV[2]) return 1 else return 0 end;System.out.println(lockName: lockName \t uuidValue: uuidValule);// 加锁失败需要自旋一直获取锁while (!stringRedisTemplate.execute(new DefaultRedisScript(script, Boolean.class),Arrays.asList(lockName),uuidValule,String.valueOf(expireTime))) {// 休眠60毫秒再来重试try {TimeUnit.MILLISECONDS.sleep(60);} catch (InterruptedException e) {e.printStackTrace();}}return true;}return false;}Overridepublic void unlock() {String script if redis.call(hexists, KEYS[1], ARGV[1]) 0 then return nil elseif redis.call(hincrby, KEYS[1], ARGV[1], -1) 0 then return redis.call(del, KEYS[1]) else return 0 end;System.out.println(lockName: lockName \t uuidValue: uuidValule);// LUA脚本由C语言编写nil - false; 0 - false; 1 - true;// 所以此处DefaultRedisScript构造函数返回值不能是BooleanBoolean没有nilLong flag stringRedisTemplate.execute(new DefaultRedisScript(script, Long.class),Arrays.asList(lockName),uuidValule);if (null flag) {throw new RuntimeException(this lock does not exists.);}}// 下面两个暂时用不到不用重写Overridepublic void lockInterruptibly() throws InterruptedException {}Overridepublic Condition newCondition() {return null;}
}完整的分布式锁java代码如下所示
// v7.0 使用自研的lock/unlockLUA脚本自研的Redis分布式锁
Lock redisDistributedLock new RedisDistributedLock(stringRedisTemplate, luojiaRedisLock);
public String sale() {String resMessgae ;redisDistributedLock.lock();try {// 1 抢锁成功查询库存信息String result stringRedisTemplate.opsForValue().get(inventory01);// 2 判断库存书否足够Integer inventoryNum result null ? 0 : Integer.parseInt(result);// 3 扣减库存每次减少一个库存if (inventoryNum 0) {stringRedisTemplate.opsForValue().set(inventory01, String.valueOf(--inventoryNum));resMessgae 成功卖出一个商品库存剩余 inventoryNum \t 服务端口号 port;log.info(resMessgae);} else {resMessgae 商品已售罄。 \t 服务端口号 port;log.info(resMessgae);}} finally {redisDistributedLock.unlock();}return resMessgae;
}总结
synchronized单机版OK - v1.0 Nginx分布式微服务轮询多台服务器单机锁不行- v2.0 取消单机锁上redis分布式锁setnx中小企业使用没问题- v3.1 只是加锁了没有释放锁出异常的话可能无法释放锁必须要在代码层面finally释放锁 - v3.2 如果服务宕机部署了微服务代码层面根本就没有走到finally这块没办法保证解锁这个Key没有被删除需要对锁设置过期时间 - v3.2 为redis的分布式锁key增加过期时间还必须要保证setnx过期时间在同一行保证原子性 - v4.1 程序由于执行超过锁的过期时间所以在finally中必须规定只能自己删除自己的锁不能把别人的锁删除了防止张冠李戴 - v5.0 将Lock、unlock变成LUA脚本保证原子性 - v6.0 保证锁的可重入性hset替代setnxLock变成LUA脚本保障可重入性 - v7.0 锁的自动续期 - v8.0
