「这是我参与11月更文挑战的第13天，活动详情查看：2021最后一次更文挑战」。

减库存场景

分布式锁，锁的是资源，我们先设定一个场景，按照这个场景去一步步说这个分布式锁。

假设我们有个减库存的场景，将库存存到redis当中，每当来个减库存的请求，就进行减库存，更新redis

如果是单机场景就是只用一个服务器部署一个单机项目，在高并发场景下多个线程同时去减库存，这时候

可以用redis的setnx命令，这个命令会返回true或false,如果当前key已经存在，就会返回false,不存在就会返回true，通过这个可以实现锁的机制，但是也会出现问题，如果执行过程出现异常，没有来及删除锁，就会出现死锁,所以这时可以设置一个过期时间，在这个时间内如果还没释放锁，锁就会自动被释放，为了保证设置锁和设置过期时间是原子性的，可以使用lua脚本来实现，也可以用syrnchronized来实现，但是syrnchronized在单实例场景下没有问题，在分布式场景，集群部署场景下就不可以了。

这里我们用springboot 集成redis来演示下基于redis的setnx加过期时间的实现

csharp复制代码
@GetMapping("/lock")

public String TestRedis() {

String clientId = UUID.randomUUID().toString();

try {

Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent("stock", clientId, 10, TimeUnit.SECONDS);

if (!aBoolean) {

return "error";

}

redisslock.lock(30, TimeUnit.SECONDS);

int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));

if (stock > 0) {

stock = stock - 1;

redisTemplate.opsForValue().set("stock", stock + "");

System.out.println("扣减成功，剩余库存" + stock);

} else {

System.out.println("扣减失败，剩余库存" + stock);

}

} catch (Exception e) {

e.printStackTrace();

} finally {

if (clientId.equals(redisTemplate.opsForValue().get("stock"))) {

// 释放锁

redisTemplate.delete("stock");

}

}

这种方式在设置锁时，value用了一个uuid生成的随机数，为什么要这样做呢，假设一个场景，key设置的过期时间到了，但是线程1还没执行完，这时候线程2就会拿到锁，此时线程1执行完，释放锁，释放的就是线程2的锁，为了让每个线程释放自己的锁，所以会设置一个惟一的value，每次释放锁的时候判断value是不是自己的value，是的话再释放锁。

这种方式实现分布式锁，存在过期时间到线程还没执行完的情况，所以为了让线程1执行完前锁不会过期，我们可以采用Redission来实现分布式锁。

Redisson

首先看下具体的实例代码，看下Redisson怎么实现分布式锁。

csharp复制代码
@Autowired

private Redisson redisson;

@GetMapping("/lock")

public String TestRedis() {

String clientId = UUID.randomUUID().toString();

RLock redisslock = redisson.getLock("stocks");

try {

// Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent("stock", clientId, 10, TimeUnit.SECONDS);

// if (!aBoolean) {

// return "error";

// }

redisslock.lock(30, TimeUnit.SECONDS);

int stock = Integer.parseInt(redisTemplate.opsForValue().get("stock"));

if (stock > 0) {

stock = stock - 1;

redisTemplate.opsForValue().set("stock", stock + "");

System.out.println("扣减成功，剩余库存" + stock);

} else {

System.out.println("扣减失败，剩余库存" + stock);

}

} catch (Exception e) {

e.printStackTrace();

} finally {

redisslock.unlock();

// if (clientId.equals(redisTemplate.opsForValue().get("stock"))) {

// // 释放锁

// redisTemplate.delete("stock");

// }

}

很简单，几步就完成了分布式锁的实现，来看下它的运行流程

加锁机制

客户端要加锁，如果该客户端面对的是一个redis cluster集群，首先会根据hash节点来选择一台机器。

紧接着会发送一段lua脚本到redis上。

使用lua是因为将一堆业务封装在lua脚本发送给redis，会保证这段复杂的业务逻辑执行的原子性。

来看看这段lua的意思。

KEYS[1]代表加锁的那个key:

RLock lock=redisson.getLock(“mylock”);

这里自己设置了加锁的那个锁key就是“mylock”，

ARGV[1] 代表的是锁key默认的生存时间，默认30秒。
ARGV[2] 代表1加锁客户端ID,是个唯一值，用来保证只释放自己设置的锁key.

第一段if语句是判断加锁的那个key存不存在，如果不存在，就进行加锁。

加锁命令:

hset myLock 8743c9c0-0795-4907-87fd-6c719a6b4586:1 1

可以看到是使用hash来进行设置加锁key,最后这个1代表该线程对这个key加锁的次数，下面讲到可重入锁会用到。

接着会执行 “pexpire myLock 30000”命令，对key设置过期时间。

这个加锁过程就算完成了。

锁互斥机制

这时候，如果客户端2来尝试加锁，执行了同样一段lua脚本，首先会判断当前的key是否存在，如果存在，接着判断当前key的value是否是客户端2的客户ID。

如果不是，接着获取这个锁key的过期时间，然后该客户端会进入一个while循环，不停的尝试加锁。

watch log 自动延期机制

客户端1加锁默认过期时间是30秒，如果想在超过了30秒，客户端1还想一直持有锁，这时就可以用这个自动延期机制，只要客户端1加锁成功，redis就会启动一个wattch log，它是一个后台线程，每隔10秒检查一下，这个具多少时间检查一下，一般是根据过期时间的三分之一，如果客户端1还持有锁，就会不断延长锁key的生存时间。

可重入加锁机制

如果客户端1已经持有这把锁，可重入的加锁如何处理的呢？

我们再回过头看下那段lua脚本，第二个if判断如果成立，因为mylock会包含客户端的ID,如果ID相等，此时就会执行可重入锁逻辑，

incrby mylock,通过这个命令对客户端1加锁次数加1

可以看到最后那个2就代表加锁次数。

是否锁机制

lock.unlock来释放锁，就是对mylock的那个加锁次数减1，㝉加锁次数变为0了，就会 del mylock，来删除这个key.

Redisson缺点

如果Redis是主从结构，如果对主加点写入了这个mylock锁，此时会异步复制给从节点，如果在这个过程主节点宕机，还没来得及复制给从节点，主备切换，从节点变为主节点，就会导致客户端2也会加锁成功，而客户端1也以为自己成功加了锁，就会导致多个客户端对一个分布式锁完成加锁。

RedLock

现在假设有5个Redis主节点（大于3的奇数个），这样基本保证了他们不会同时都宕机，获取锁和释放锁的过程，客户端会执行以下操作。

获取当前Unix时间，以毫秒为单位
依次尝试从5个实例，使用相同的key和具有唯一性的value获取锁

当向Redis请求获取锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间，这样可以避免客户端死等

客户端使用当前时间减去开始获取锁时间就得到获取锁使用的时间，当且仅当从半数以上的Redis节点取到锁，并且使用的时间小于锁失效时间，锁才算获取成功
如果获取到锁，key真正的有效时间等于有效时间减去获取锁所使用的时间
如果因为某些原因，获取锁失败（没有在半数以上实例取到锁或者取锁时间已经超过了有效时间）

，客户端会在所有Redis实例上进行解锁，无论Redis实例是否加锁成功，都会去解锁。

失败的重试

当一个客户端获取锁失败时，这个客户端应该在一个随机延时后进行重试，之所以采用随机延时是为了避免不同客户端同时重试导致谁都无法拿到锁的情况出现。同样的道理客户端越快尝试在大多数Redis节点获取锁，出现多个客户端同时竞争锁和重试的时间窗口越小，可能性就越低，所以最完美的情况下，客户端应该用多路传输的方式同时向所有Redis节点发送SET命令。这里非常有必要强调一下客户端如果没有在多数节点获取到锁，一定要尽快在获取锁成功的节点上释放锁，这样就没必要等到key超时后才能重新获取这个锁（但是如果网络分区的情况发生而且客户端无法连接到Redis节点时，会损失等待key超时这段时间的系统可用性）