21| 原生分布式锁-Redisson分布式锁防止个人超领优惠券

· · · ·

### 超领业务问题和效果

- 优惠券限制1⼈限制1张，有些⼈却领了2张
- 优惠券限制1⼈限制2张，有些⼈却领了3或者4张

举例：

有个洗发水100元，有个10元优惠券，每⼈限制领劵1张，老王使用时间暂停思维来发现问题，并发领劵。A线程原先查询出来没有领劵，要再插⼊领劵记录前暂停然后B线程原先查询出来也没有领劵，则插⼊领劵记录，然后A线程也插⼊领劵记录。⽼王就有了两个优惠券

`问题来源核⼼：对资源的修改没有加锁，导致多个线程可以同时操作，从⽽导致数据不正确`

`解决问题：分布式锁 或者 细粒度分布式锁`。

避免单⼈超领劵：

- 加锁
  - `本地锁：synchronize、lock`等，锁在当前进程内，集群部署下依旧存在问题
  - `分布式锁：redis、zookeeper`等实现，虽然还是锁，但是`多个进程共⽤的锁标记`，可以⽤`Redis、Zookeeper、Mysql等都可以`

设计分布式锁应该考虑的东⻄

- `排他性`：在分布式应⽤集群中，同⼀个⽅法在同⼀时间`只能被⼀台机器上的⼀个线程执⾏`。

- `容错性`：分布式锁`⼀定能得到释放`，⽐如客户端奔溃或者⽹络中断。

- 满⾜`可重⼊、⾼性能、⾼可⽤`

- 注意分布式锁的`开销、锁粒度`

### 基于Redis实现分布式锁的方式及问题

实现分布式锁 可以⽤`Redis、Zookeeper、Mysql`数据库这⼏种 , `性能最好的是Redis`且是最容易理解。

分布式锁离不开`key - value`设置

-  key 是`锁的唯⼀标识`，⼀般按业务来决定命名，⽐如想要给⼀种商品的秒杀活动加锁，key 命名为 “`seckill`商品ID”。
-  value就可以使⽤`固定值`，⽐如设置成1

基于redis实现分布式锁，⽂档：http://www.redis.cn/commands.html#string

- 加锁 `SETNX key value`
  - setnx 的含义就是 `SET if Not Exists`，有两个参数`setnx(key, value)`，该⽅法是原⼦性操作
  - 如果 key 不存在，则设置当前 key 成功，返回 1；
  - 如果当前 key 已经存在，则设置当前 key 失败，返回 0

- 解锁 `del (key)`
  - 得到锁的线程执⾏完任务，需要`释放锁`，以便其他线程可以进⼊,调⽤`del(key)`配置`锁超时expire(key,30s)`
  - 客户端奔溃或者`⽹络中断，资源将会永远被锁住,即死锁`，因此需要给key`配置过期时间`，以保证即使没有被显式释放，这把锁也要在⼀定时间后`⾃动释放`。

实现的伪代码：

```java
methodA(){
  String key = "coupon_66"
  if（setnx（key，1） == 1）{
    expire(key,30,TimeUnit.MILLISECONDS)
    try {
     //做对应的业务逻辑
     //查询⽤户是否已经领券
     //如果没有则扣减库存
     //新增领劵记录
     } finally {
       del（key）
     }
   }else{
    //睡眠100毫秒，然后⾃旋调⽤本⽅法
     methodA()
  }
}
```

存在什么问题？

- **多个命令之间不是原⼦性操作，如setnx和expire之间，如果setnx成功，但是expire失败，且宕机了，则这个资源就是死锁**。

- 业务超时，`存在其他线程勿删`，key 30秒过期，假如线程A执⾏很慢超过30秒，则key就被释放了，其他线程B就得到了锁，这个时候线程A执⾏完成，⽽B还没执⾏完成，结果就是线程`A删除了线程B加的锁`。(**设置超时时间太短导致业务还未执行完就释放了的问题**)

```java
// 可以在 del 释放锁之前做⼀个判断，验证当前的锁是不是⾃⼰加的锁, 那value 应该是存当前线程的标识或者uuid
String key = "coupon_66"
String value =Thread.currentThread().getId()
if（setnx（key，value） == 1）{
   expire(key,30,TimeUnit.MILLISECONDS)
   try {
     //做对应的业务逻辑
    } finally {
      //删除锁，判断是否是当前线程加的
      if(get(key).equals(value)){
          //还存在时间间隔
           del（key）
        }
    }
}else{ 
 //睡眠100毫秒，然后⾃旋调⽤本⽅法
}
```

进⼀步细化误删:

- 当线程A获取到正常值时，返回带代码中判断期间锁过期了，线程B刚好重新设置了新值，线程A那边有判断value是⾃⼰的标识，然后调⽤del⽅法，结果就是删除了新设置的线程B的值。

- 核⼼还是`判断和删除命令不是原⼦性操作导致`。

### 分布式锁lua脚本+redis解决非原子操作导致死锁问题

redis做分布式锁存在的问题

- 核⼼是`保证多个指令原⼦性`，加锁使⽤`setnx setex`可以保证原⼦性，那解锁使⽤判断和删除怎么保证原⼦性

⽂档：http://www.redis.cn/commands/set.html (官方不推荐使用)

- **多个命令的原⼦性：采⽤ lua脚本+redis, 由于【判断和删除】是lua脚本执⾏，所以要么全成功，要么全失败**

```java
//获取lock的值和传递的值⼀样，调⽤删除操作返回1，否则返回0
String script = "if redis.call('get',KEYS[1])== ARGV[1] then return
redis.call('del',KEYS[1]) else return 0 end";
//Arrays.asList(lockKey)是key列表，uuid是参数
Integer result = redisTemplate.execute(newDefaultRedisScript<>(script, Integer.class),Arrays.asList(lockKey), uuid);
```

代码实现：

```java
       /**
         * 原⽣分布式锁 开始
         * 1、原⼦加锁 设置过期时间，防⽌宕机死锁
         * 2、原⼦解锁：需要判断是不是⾃⼰的锁
         */
        String uuid = CommonUtil.generateUUID();
        String lockKey = "lock:coupon:"+couponId;
        Boolean nativeLock=redisTemplate.opsForValue().setIfAbsent(lockKey,uuid,Duration.ofSeconds(30));
        if(nativeLock){
            //加锁成功
            log.info("加锁：{}",nativeLock);
            try {
                //执⾏业务 TODO
            }finally {
                String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
                Integer result = redisTemplate.execute(new DefaultRedisScript<>(script, Integer.class), Arrays.asList(lockKey), uuid);
                log.info("解锁：{}",result);
            }
        }else {
            //加锁失败，睡眠100毫秒，⾃旋᯿试
            try {
                TimeUnit.MILLISECONDS.sleep(100L);
            } catch (InterruptedException e) {
            }
            return addCoupon( couponId, couponCategory);
        }
        //原⽣分布式锁 结束
    }
```

遗留⼀个问题，锁的过期时间，`如何实现锁的⾃动续期或者避免业务执⾏时间过⻓`，锁过期了？

- 原⽣⽅式的话，⼀般`把锁的过期时间设置久⼀点`，⽐如10分钟时间。

### redis官方推荐-分布式锁最佳实践Redisson

原⽣代码+redis实现分布式锁使⽤⽐较复杂，且有些锁续期问题更难处理。

官⽅推荐⽅式：https://redis.io/topics/distlock

- 多种实现客户端框架(`java中推荐Redisson`)

Redisson官⽅中⽂⽂档：https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95

**使用Redisson的redis必须要设置密码**

聚合⼯程锁定版本，common项⽬添加依赖（多个服务都会⽤到分布式锁）

```xml

<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson</artifactId>
  <version>3.10.1</version>
</dependency>
```

创建redisson客户端

```java
@Configuration
@Data
public class RedissonConfig {
    @Value("${spring.redis.host}")
    private String redisHost;
    @Value("${spring.redis.port}")
    private String redisPort;
    @Value("${spring.redis.password}")
    private String redisPwd;
    /**
     * 配置分布式锁的redisson
     */
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        //单机方式
        config.useSingleServer().setPassword(redisPwd).setAddress("redis://" + redisHost + ":" + redisPort);
        //集群
// config.useClusterServers().addNodeAddress("redis://127.0.0.1:6379","redis://127.0.0.2:6379")
        return Redisson.create(config);
    }
}
```

模拟controller接⼝测试

```JAVA
@GetMapping("lock")
public JsonData testLock() {
    RLock lock = redissonClient.getLock("lock:coupon:1");
    //阻塞等待
    //lock.lock(10,TimeUnit.MILLISECONDS);
    lock.lock();
    try {
        log.info("加锁成功，处理业务逻辑。。。。。。" + Thread.currentThread().getId());
        TimeUnit.SECONDS.sleep(20);
    } catch (Exception e) {
    } finally {
        log.info("解锁成功，其他线程可以进去。。。。。。" + Thread.currentThread().getId());
        lock.unlock();
    }
    return JsonData.buildSuccess();
}
```

领劵redisson分布式锁实现

```java
@Resource
private RedissonClient redissonClient;

@Transactional(rollbackFor=Exception.class,propagation= Propagation.REQUIRED)
@Override
public JsonData addCoupon(Long couponId, CouponCategoryEnum category) {
    LoginUser loginUser = LoginInterceptor.threadLocal.get();
    String lockKey = "lock:coupon:" + couponId;
    RLock rLock = redissonClient.getLock(lockKey);
    //多个线程进入，会阻塞等待释放锁
    rLock.lock();
    log.info("领劵接口加锁成功:{}", Thread.currentThread().getId());
    try {
        CouponEntity coupon = baseMapper.selectOne(
                Wrappers.<CouponEntity>lambdaQuery()
                        .eq(CouponEntity::getId, couponId).eq(CouponEntity::getCategory, category.name()));
        //优惠券是否可以领取
        this.checkCoupon(coupon, loginUser.getId());
        //构建领劵记录
        CouponRecordEntity couponRecord = new CouponRecordEntity();
        BeanUtils.copyProperties(coupon, couponRecord);
        couponRecord.setCreateTime(new Date());
        couponRecord.setUseState(CouponStateEnum.NEW.name());
        couponRecord.setUserId(loginUser.getId());
        couponRecord.setUserName(loginUser.getName());
        couponRecord.setCouponId(couponId);
        couponRecord.setId(null);
        //扣减库存
        int rows = baseMapper.reduceStock(couponId);
        if (rows == 1) {
            //库存扣减成功才保存
            couponRecordMapper.insert(couponRecord);
        } else {
            log.warn("发放优惠券失败:{},⽤户:{}", coupon, loginUser);
            throw new BizException(BizCodeEnum.COUPON_NO_STOCK);
        }
    } finally {
        rLock.unlock();
        log.info("解锁成功");
    }
    return JsonData.buildSuccess();
}
```

### redisson解决分布式锁里面的坑

Redis锁的过期时间⼩于业务的执⾏时间该如何续期？

- `watch dog看⻔狗机制`

负责储存这个分布式锁的`Redisson节点宕机`以后，⽽且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。或者业务`执⾏时间过⻓导致锁过期`，为了避免这种情况的发⽣，Redisson内部提供了⼀个`监控锁的看⻔狗`，它的作⽤是在`Redisson实例被关闭前，不断的延⻓锁的有效期`。

`Redisson中客户端⼀旦加锁成功，就会启动⼀个watchdog看⻔狗`。watch dog是⼀个后台线程，会每隔10秒检查⼀下，如果客户端还持有锁key，那么就会不断的延⻓锁key的⽣存时间

默认情况下，`看⻔狗的检查锁的超时时间是30秒钟`，也可以通过修改`Config.lockWatchdogTimeout`来另⾏指定

指定加锁时间

```java
// 加锁以后10秒钟⾃动解锁
// ⽆需调⽤unlock⽅法⼿动解锁
lock.lock(10, TimeUnit.SECONDS);
// 尝试加锁，最多等待100秒，上锁以后10秒⾃动解锁
boolean res = lock.tryLock(100, 10,TimeUnit.SECONDS);
if (res) {
 try {
 ...
 } finally {
 lock.unlock();
 }
}
```

21| 原生分布式锁-Redisson分布式锁防止个人超领优惠券

编码规范参考文档

关于Vue和Python相关的资料文档

Git Fork后的仓库内容和原仓库保持一致

JDK 源码解析 —— 调试环境搭建（二）进阶

JDK 源码解析 —— 集合（一）数组 ArrayList

Dubbo 源码分析 —— 调试环境搭建