1.为什么需要分布式锁。
当我们的应用程序,有多个实例,部署在多个服务器上时。当我们进行对操作库存,秒杀等操作时,一定需要加锁保证我们的库存数量被正确的赋值,但是我们的Lock锁,Synchronized锁都是基于我们的JVM的,但是我们的项目时多实例,部署在多个应用服务器的,这时候我们的本地锁就不能保证我们操作的原子性了。这时候我们需要一个服务器级别的锁,也就是分布式锁。
2.分布式锁的实现基于Redis
分布式锁,其实原理是就是多台机器,去争抢一个资源,谁争抢成功,那么谁就持有了这把锁,然后去执行后续的业务逻辑,执行完毕后,把锁释放掉。通过redis的setnx方式(不存在则设置),往redis上设置一个带有过期时间的key,如果设置成功,则获得了分布式锁。
3.代码实现和会出现的问题我们一点点的看
我们就以对库存的加减为案例:
基础代码:
解析:单服务应用下好像没什么问题,第一个请求进来会执行库存扣除操作,第二个请求会等待等一个请求释放锁,拿到锁对象后进行操作。
问题:多个实例服务器时,Synchronied()锁就没有用了。
@Autowired
private RedisTemplate redisTemplate;
private static final String STOCK = "stock";
/**
* @Description:redis 分布式锁模拟高并发下库存删减
* @Author: hdh
* @Date: 2021/2/1
* @return: java.lang.String
**/
@RequestMapping("stock")
private synchronized String stock() {
//从redis中获取库存
int stock = (Integer) redisTemplate.opsForValue().get(STOCK);
if (stock > 0) {
redisTemplate.opsForValue().set(STOCK, stock - 1);
return "扣除库存成功";
}
return "扣除库存失败,库存不足";
}
解析:redis是单线程操作,能保证操作的原子性
redisTemplate.opsForValue().setIfAbsent(key,value)判断redis 中是否存在key存在不进行操作返回false,不存在保存key返回true;
问题:解决了锁的问题,出现了新的问题,如果在扣除库存抛出了异常redis锁没有及时释放,那么该请求会被死锁
@RequestMapping("stock")
private String stock() {
Boolean redis_lock = redisTemplate.opsForValue().setIfAbsent("REDIS_LOCK", REDIS_LOCK);
if (!redis_lock) {
return "业务繁忙,请稍后再试";
}
//从redis中获取库存
int stock = (Integer) redisTemplate.opsForValue().get(STOCK);
if (stock > 0) {
redisTemplate.opsForValue().set(STOCK, stock - 1);
redisTemplate.delete(REDIS_LOCK);
return "扣除库存成功";
}
redisTemplate.delete(REDIS_LOCK);
return "扣除库存失败,库存不足";
}
解析:通过给锁setIfAbsent()设置过期时间可以防止被锁一直被占用
将释放锁放在finally中,可以保证不会死锁。
REDIS_LOCK + uuid:实现了自己设置的锁只有自己可以销毁
问题:setIfAbsent()设置过期时间,如果在过期时间内业务代码没有执行完,那么下一个请求就会进来执行,而上一个请求执行完就会释放锁,导致锁直接失效了。但是时间又不能设置过长,如果在执行业务代码的时候宕机了,那么会导致多个服务的库存接口直接死锁。设置多少秒的锁合适呢。
@RequestMapping("stock")
private String stock() {
String uuid = String.valueOf(UUID.randomUUID());
//加上过期时长,该锁10s后自动被释放
Boolean redis_lock = redisTemplate.opsForValue().setIfAbsent(REDIS_LOCK + uuid, REDIS_LOCK, 10, TimeUnit.SECONDS);
if (!redis_lock) {
return "业务繁忙,请稍后再试";
}
try {
int stock = (Integer) redisTemplate.opsForValue().get(STOCK);
if (stock > 0) {
redisTemplate.opsForValue().set(STOCK, stock - 1);
redisTemplate.delete(REDIS_LOCK+uuid);
return "扣除库存成功";
}
} finally {
//发生异常自动解锁
redisTemplate.delete(REDIS_LOCK+uuid);
}
//从redis中获取库存
return "扣除库存失败,库存不足";
}
4. redisson redis客户断的实现
<!--redisson redis客户端-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>
bean初始化
@Bean
public Redisson redisson() {
//单机模式
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("12345").setDatabase(0);
//添加多个实例
/*config.useClusterServers()
.addNodeAddress("redis://127.0.0.1:6379")
.addNodeAddress("redis://127.0.0.1:6379")
.addNodeAddress("redis://127.0.0.1:6379");*/
return (Redisson) Redisson.create(config);
}
5. redisson实现分布式锁
@RequestMapping("stock")
private String stock() {
RLock lock = redisson.getLock(LOCK_KEY);
try {
//设置锁
lock.lock();
int stock = (Integer) redisTemplate.opsForValue().get(STOCK);
if (stock > 0) {
redisTemplate.opsForValue().set(STOCK, stock - 1);
return "扣除库存成功";
}
} finally {
//解除锁
lock.unlock();
}
return "扣除库存失败,库存不足";
}
redisson底层其实是做了setIfAbsent(REDIS_LOCK + uuid, REDIS_LOCK, 10, TimeUnit.SECONDS);
但是redisson会使用后台线程检查是否继续持有锁,如果继续持有则延长锁的时间
6. 实例化不同的redis实例连接多个redisson客户端连接
@RequestMapping("stock")
private String stock() {
RLock lock1 = redisson1.getLock(LOCK_KEY);
RLock lock2 = redisson2.getLock(LOCK_KEY);
RLock lock3 = redisson3.getLock(LOCK_KEY);
//根据多个RLock对象创建RedissonRedLock;
RedissonRedLock redLock=newRedissonRedLock(lock1,lock2,lock3);
try {
//设置锁
boolen res=redLock.try.Lock(10,30,TimeUnit.SECONDS);
//判断锁是否设置成功
if(res){
int stock = (Integer)redisTemplate.opsForValue().get(STOCK);
if (stock > 0) {
redisTemplate.opsForValue().set(STOCK, stock - 1);
return "扣除库存成功";
}
}
} finally {
//解除锁
lock.unlock();
}
return "扣除库存失败,库存不足";
}