一、分布式锁介绍

之前我们都是使用本地锁（synchronize、lock等）来避免共享资源并发操作导致数据问题，这种是锁在当前进程内。

那么在集群部署下，对于多个节点，我们要使用分布式锁来避免共享资源并发操作导致数据问题，虽然还是锁，但是是多个进程共用的锁标记，可以用Redis、Zookeeper、Mysql等都可以实现。

案例：优惠券领劵限制张数、商品库存超卖。

我们设计分布式锁应该要考虑的东西：

• 排他性：在分布式应用集群中，同一个方法在同一时间只能被一台机器上的一个线程执行。

• 容错性：分布式锁一定能得到释放，比如客户端奔溃或者网络中断，可能会导致锁一直不被释放，从而导致死锁，我们可以设置锁的过期时间。

• 满足可重入、高性能、高可用（集群部署）。

• 注意分布式锁的开销、锁粒度。

二、分布式锁的实现

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

分布式锁离不开 key - value 设置，key 是锁的唯一标识，一般按业务来决定命名，比如想要给一种优惠券活动加锁，key 命名为 “coupon:id” 。value就可以使用固定值，比如设置成1。

基于redis实现分布式锁：

（1）、加锁 setnx key value：

setnx 的含义就是 set if not exists，有两个参数 setnx(key, value)，该方法是原子性操作，如果 key 不存在，则设置当前 key 成功，返回 1；如果当前 key 已经存在，则设置当前 key 失败，返回 0。

（2）、解锁 del (key)：

得到锁的线程执行完任务，需要释放锁，以便其他线程可以进入,调用 del(key)。

（3)、配置锁超时 expire (key，30s）：

客户端奔溃或者网络中断，资源将会永远被锁住,即死锁，因此需要给key配置过期时间，以保证即使没有被显式释放，这把锁也要在一定时间后自动释放。

综合的伪代码：

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

三、 基于Redis实现分布式锁的几种坑

上面我们写的伪代码中有几个坑，我们分别来分析一下。

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

解决方法：使用原子命令来设置和配置过期时间 setnx / setex，在java里面是

redisTemplate.opsForValue().setIfAbsent("key","value",30,TimeUnit.MILLISECONDS)

成功了返回true，失败了返回false。 

2、业务超时，存在其他线程勿删，设置key30秒过期，假如线程A执行很慢超过30秒，则key就被释放了，其他线程B就得到了锁，这个时候线程A执行完成，而B还没执行完成，结果就是线程A删除了线程B加的锁，所以我们的value不能单单只是1。

解决方法：可以在 del 释放锁之前做一个判断，验证当前的锁是不是自己加的锁, 那 value 应该是当前线程的标识或者uuid。

String key = "coupon:id"
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毫秒，然后自旋调用本方法
​
}

 3、进一步细化误删，当线程A获取到正常值value时，返回带代码中判断期间锁过期了，线程B刚好重新设置了新值，线程A那边有判断value是自己的标识，然后调用del方法，结果就是删除了新设置的线程B的值。

解决办法：由于redis没有相关的原子性api，所以采用 lua脚本+redis来实现多个命令的原子性。由于【判断和删除】是lua脚本执行，所以要么全成功，要么全失败。

总结：核心是保证多个指令原子性，加锁使用setnx setex 可以保证原子性，解锁采用 lua脚本+redis来保证原子性。

【判断和删除】的lua脚本：

//获取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是参数
Long result = redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockKey), uuid);

 四、原生分布式锁的具体实现

@RestController
@RequestMapping("/api/v1/coupon")
public class CouponController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@GetMapping("add")public JsonData saveCoupon(@RequestParam (value = "coupon_id",required = true)int couponId){//防止其他线程误删String uuid = UUID.randomUUID().toString();String lockKey = "lock:coupon:" + couponId;lock(couponId,uuid,lockKey);return JsonData.buildSuccess();}private void lock(int couponId,String uuid,String lockKey){//lua脚本String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";Boolean nativeLock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,uuid, Duration.ofSeconds(30));System.out.println(uuid+"加锁状态:"+nativeLock);if(nativeLock){//加锁成功try{//TODO 做相关业务逻辑TimeUnit.SECONDS.sleep(10L);} catch (InterruptedException e) {} finally {//解锁Long result = stringRedisTemplate.execute( new DefaultRedisScript<>(script,Long.class), Arrays.asList(lockKey),uuid);System.out.println("解锁状态:"+result);}}else {//自旋操作try {System.out.println("加锁失败，睡眠5秒 进行自旋");TimeUnit.MILLISECONDS.sleep(5000);} catch (InterruptedException e) { }//睡眠一会再尝试获取锁lock(couponId,uuid,lockKey);}}}

运行结果：

d124ae03-5de6-4e25-82b8-fb0b30d7c7fc加锁状态:true
54041d23-ab3c-492e-977b-99c9b531534f加锁状态:false
加锁失败，睡眠5秒 进行自旋
51f16a96-45cd-476b-95ff-2ee6cc398e37加锁状态:false
加锁失败，睡眠5秒 进行自旋
54041d23-ab3c-492e-977b-99c9b531534f加锁状态:false
加锁失败，睡眠5秒 进行自旋
51f16a96-45cd-476b-95ff-2ee6cc398e37加锁状态:false
加锁失败，睡眠5秒 进行自旋
解锁状态:1
54041d23-ab3c-492e-977b-99c9b531534f加锁状态:true
51f16a96-45cd-476b-95ff-2ee6cc398e37加锁状态:false
加锁失败，睡眠5秒 进行自旋
51f16a96-45cd-476b-95ff-2ee6cc398e37加锁状态:false
加锁失败，睡眠5秒 进行自旋
解锁状态:1
51f16a96-45cd-476b-95ff-2ee6cc398e37加锁状态:true
解锁状态:1