⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 blog.csdn.net/xuan_lu/article/details/111600302 「xuan_lu」欢迎转载，保留摘要，谢谢！

分布式锁

基于redis实现分布式锁思考几个问题？？？

synchronized锁为什么不能应用于分布式锁？

synchronized虽然能够解决同步问题，但是每次只有一个线程访问，并且synchronized锁属于JVM锁，仅适用于单点部署；然而分布式需要部署多台实例，属于不同的JVM线程对象

使用redis中setnx实现分布式锁。

//设置分布式锁
String lockKey = "product_001_key";
//语义：如何不存在则存入缓存中，且返回true；
//否则已存在，则返回false即加锁失败
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "product_001_lock");
        if (!result) {
        //没有加锁成功，则返回提示等
        }
        try{

        }catch() {

        }finally{
        //释放锁
        stringRedisTemplate.delete(lockKey);
        }

针对以上设置分布式锁思考一下问题？

1.如果突然服务器宕机，那么必然造成锁无法释放，即造成死锁？

解决方案：设置超时时间。

//设置分布式锁
String lockKey = "product_001_key";
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "product_001_lock");
//设置锁超时时间30s
stringRedisTemplate.expire(lockKey,30, TimeUnit.SECONDS);
if (!result) {
  //没有加锁成功，则返回提示等
}
try{

}catch() {

}finally{
  //释放锁
  stringRedisTemplate.delete(lockKey);
}

2.加锁和设置超时时间中间引起服务器宕机，则一样会导致死锁。

Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "product_001_lock");
//------服务器宕机，则超时时间未设置成功-------
//设置锁超时时间30s
stringRedisTemplate.expire(lockKey,30, TimeUnit.SECONDS);

解决方案：原子性操作，即同时加锁和设置超时时间；

即上面的代码合并成一句操作：

Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"product_001_lock", 30, TimeUnit.SECONDS)

3.思考超时时间设置是否合理呢？即线程执行时间和锁超时时间并非一致。

场景：假设设置加锁超时时间10s；

高并发场景下，线程A执行时间为15s，redis依据超时时间，将其线程A加的锁释放掉；然后线程B获取锁，并加锁成功，此时线程A执行结束，执行finally代码块就会将线程B加的锁释放。

解决方案：设置线程随机ID，释放锁时判断是否为当前线程加的锁，即使存在线程A因线程执行时间超时被动释放其锁，但至少保证当前超时线程不会释放其他线程加的锁。但是面对线程执行时间大于设置的超时时间，也是会存在并发问题。

String lockKey = "product_001";
String clientId = UUID.randomUUID().toString();
//设置超时时间，且加锁和设置线程ID
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,clientId, 30, TimeUnit.SECONDS)`

if (!result) {
  //没有加锁成功，则返回提示等
}
try{

}catch() {

}finally{
  //释放锁：加锁线程ID和当前执行线程ID相同，才允许释放锁
 if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
        stringRedisTemplate.delete(lockKey);
    }
}

4.上面场景解决方案：加锁续命即续线程锁超时时间

解决方案：加锁成功时，开启一个后台线程，每隔10s（自定义）判断当前线程是否还持有锁，持有锁则再续命30s等

Redission实现分布式锁

实现原理流程：

String lockKey = "product_001";
     //获取锁对象，并未加锁
     RLock redissonLock = redisson.getLock(lockKey);
     try {
     // **此时加锁**，实现锁续命功能
     redissonLock.lock();
     int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
     if (stock > 0) {
     int realStock = stock - 1;
     stringRedisTemplate.opsForValue().set("stock", realStock + "");
     System.out.println("扣减成功，剩余库存:" + realStock + "");
     } else {
     System.out.println("扣减失败，库存不足");
     }
     }finally {
     //释放锁
     redissonLock.unlock();
     }

总结

综上，设计实现分布式锁需要满足一下条件：

互斥性；在任意时刻，只有一个客户端能持有锁。
不能发生死锁；即使存在一个线程持有锁的期间崩溃而没有主动解锁，也能保证后续其他线程能加锁。
加锁和解锁必须是同一个线程。

芋道源码 —— 纯源码解析博客

基于 Redis 实现分布式锁思考