前言
分布式锁在日常开发中，用处非常的多。包括但不限于抢红包，秒杀，支付下单，幂等，等等场景。
分布式锁的实现方式有多种，包括redis实现，MySQL实现，zookeeper实现等等。而其中redis非常适合作为分布式锁使用，并且在各个公司都大规模的使用。
本文将由浅入深的探究Redis分布式锁的实现，最终实现一个可工业使用的Redis分布式锁。欢迎大家一步一步跟读，一起学习一起进步。
什么是分布式锁
分布式锁其实就是，控制分布式系统不同进程共同访问共享资源的一种锁的实现。如果不同的系统或同一个系统的不同主机之间共享了某个临界资源，往往需要互斥来防止彼此干扰，以保证一致性。
举一个最简单的例子。有一个数据库字段status=0，表示初始状态。只有在status=0初始状态。才能修改这个值。现在有两个人，张三和李四。
- 张三，发起请求将status=0 修改为 status=1
- 李四，发起请求将status=0 修改为 status=2
因为只有status=0才会修改，代码在修改之前都会去查询status的值，并且判断是否为0。如果为0才会去更新，不为0，则拒绝更新。这其实就是一个幂等的实现。
- 假如没有分布式锁。短时间内请求两次，此时两次都获取status=0，一个修改成了1，一个修改成了2。破坏代码逻辑，有问题
- 假如加上分布式锁。短时间内请求两次，只有第一笔请求结束之后，第二笔才会执行。也就是第二笔获取status，只能获取到最新的值，比如status=1，则不修改。
Redis分布式锁方案一：SETNX （不推荐）
public String lockA(String key) {
String val = UUID.randomUUID().toString();
// set k v nx 如果不存在则设置成功，如果存在则设置失败
boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val);
if (success) {
log.info("lock success");
try {
// do something
} finally {
stringRedisTemplate.delete(key);
}
} else {
log.info("lock fAIl");
}
return "lockA";
}
这个方案有一个最大的问题就是，如果线程A获取锁成功，并没有设置过期时间。那么如果此时doSomething里面是一个死循环或者程序在期间重启了，就会导致这个锁就不会被释放，那么别的线程永远获取不到锁啦。这个问题非常严重，对业务影响极大。不推荐使用。
Redis分布式锁方案二：SETNX + expire （不推荐）
那既然没有过去时间，我就设置一个过期时间不就行了，代码如下。
public String lockB(String key) {
String val = UUID.randomUUID().toString();
// set k v nx 如果不存在则设置成功，如果存在则设置失败
boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val);
stringRedisTemplate.expire(key, 60, TimeUnit.SECONDS);
if (success) {
log.info("lock success");
try {
// do something
} finally {
stringRedisTemplate.delete(key);
}
} else {
log.info("lock fail");
}
return "lockB";
}
这个方案2和方案1有同样的问题。setnx 和 expire不是一个原子执行。在获取锁成功之后，准备执行expire的时候，程序重启，也会导致同样方案1的问题，此处不再赘述。不推荐使用。
Redis分布式锁方案三：SET EX NX （不推荐）
那既然不是原子性，我们就用原子性就好了。从redis 2.6.12开始，set方法支持 set ex nx
public String lockB(String key) {
String val = UUID.randomUUID().toString();
// set k v ex nx 如果不存在则设置成功，如果存在则设置失败
boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val, 60, TimeUnit.SECONDS);
if (success) {
log.info("lock success");
try {
// do something
} finally {
stringRedisTemplate.delete(key);
}
} else {
log.info("lock fail");
}
return "lockB";
}
从方案三开始，此代码就比较有健壮性了。有部分公司使用的就是方案三，但仍然存在两个问题
- doSomething还没执行完，锁过期就被自动释放了。那么其他线程就可以获取此锁了。就会导致此代码块可能被多个线程执行。当然使用的时候可以把过期时间设置大一点，比如60分钟，3个小时等等，但总归不太好。
- 线程A获取锁，没执行完成，锁过期了。此时线程B获取锁执行了。然后A执行完成去释放锁的时候，但他释放的是线程B获取的锁，此时是有问题的，并且问题还不小。同样不推荐使用。
Redis分布式锁方案四： （推荐）
既然时间太短，我就设置过期时间长一点。既然会被误删，我们就判断一下。代码如下
public String lockD(String key) {
String val = UUID.randomUUID().toString();
// set k v nx 如果不存在则设置成功，如果存在则设置失败
boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val, 60, TimeUnit.MINUTES);
if (success) {
log.info("lock success");
try {
// do something
} finally {
if (val.equals(stringRedisTemplate.opsForValue().get(key))) {
stringRedisTemplate.delete(key);
}
// String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
// DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
// redisScript.setScriptText(script);
// redisScript.setResultType(Long.class);
// return stringRedisTemplate.execute(redisScript, Collections.singletonList(key));
}
} else {
log.info("lock fail");
}
return "lockD";
}
大部分公司，我相信使用的都是方案四。方案四正常来说，在使用过程中极大概率不会出现任何问题，除非你们的量非常的大。但其仍有问题，finally删除锁的那块不是原子性。
比如线程A获取锁成功uuid=123, 释放成功。线程B获取锁，uuid=456,锁过期，自动释放。
此时A再次获取锁，uuid=456（恰巧是456,概率非常低）。那么A就会释放B的锁。因此为了更加严谨一点，我们使用lua脚本来保证，判断+删除的原子性。
方案四已经符合绝大多数公司的使用了，但其不好估计的过期时间，以及释放的原子性，仍 概率性的存在问题。所以社区为了解决此问题，有了以下方案。
Redis分布式锁方案五： Redission方案 （推荐）
Redisson官网介绍： Easy Redis Java client with features of an in-memory data grid（易于使用的 Redis Java 客户端，具备内存数据网格的特性）
Redisson 是一个基于 Java 的 Redis 客户端库，它提供了一系列的高级功能，使得在 Java 应用程序中使用 Redis 变得更加方便和强大。Redisson 的目标是充分利用 Redis 的各种特性，同时提供易于使用的 Java 接口。
RedissonClient 是 Java 中 Redisson 库提供的一个接口，它封装了对 Redis 数据库的各种操作，提供了丰富的方法来与 Redis 进行交互。Redisson 是一个在 Redis 的基础上实现的 Java 内存数据网格（In-Memory Data Grid）。它不仅提供了对基本数据结构的操作，还提供了分布式的 Java 对象和服务，例如分布式锁、集合、映射、发布/订阅、计数器等。
我们这次使用到的是redission的分布式锁。
// 获取锁
public String lockE(String key) {
// 获取锁
RLock lock = redissonClient.getLock(key);
try {
// 获取锁。此处30s不是指执行30s，而是获取锁的超时时间
if (lock.tryLock(30, TimeUnit.SECONDS)) {
log.info("lock success");
}
} catch (Exception e) {
} finally {
if (lock != null && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
return "lockE";
}
此方案基本适用于99.99%的公司，当然可能会出现Redlock的问题，此处不过多讨论，感兴趣的同学可以网上自行搜索。
只要线程加锁成功，默认过期时间是30s。后台会自动启动一个watch dog看门狗，它是一个后台线程，会每隔10秒检查一下，如果线程还持有锁，那么就会不断的延长锁key的生存时间。因此，Redisson就是使用Redisson解决了锁过期释放，业务没执行完问题。
具体Redission常见问题，以及源码分析，可以详见： Redis分布式锁实现Redisson 15问（面试常问）
最后
本文由浅入深的介绍了分布式锁。解释了为什么大部分公司用的都是方案四以及方案五的实现，而不是方案1,2,3。我们需要知道每个方案的优劣势，从而选出最适合我们业务的一种技术方案，这是每个架构师都应该具备的一种能力。
