前言
分布式系统中，由于多个进程之间会存在操作共享数据的情况下，此时就需要一个协调系统进行各个进程之间的协调，避免多个进程之间同时修改数据导致互相影响的情况。通常可以采用数据库锁来实现数据不会再同一时间修改，但是数据库锁的悲观锁，比较影响整个系统的性能。并且如果修改的数据并非是数据库中的数据时，通过数据库锁就无法实现了。此时就需要一个分布式锁来进行分布式协调。
一、分布式锁
高可用的分布式锁需要达到以下几种特性：
1、分布式系统中，同一个方法在同一时间只能被一个节点上的一个线程执行
2、获取锁和释放锁性能较高
3、具备锁失效的机制，避免死锁
4、可重入性
5、非阻塞性，避免获取锁失败之后一直阻塞
二、redis实现分布式锁
由于redis是单线程工作的，所以redis天然就具备了同一时间只能有一个线程执行的条件。而redis分布式锁根据redis部署方式不同又分成两个版本，一个是redis单机部署版本，一个是redis集群部署版本。两种redis部署方式的分布式锁的实现也完全不同
2.1、redis单机模式分布式锁
redis的字符串操作API中提供了setnx key value方法，该方法的作用是只有在key不存在时才会设置value，redis的分布式锁主要就依赖此方法来实现。
方案一：采用命令setnx命令获取锁，设置值成功表示获取锁成功，如果设置失败表示key已经存在则表示获取锁失败。通过delete命令删除key的方式释放锁
步骤如下：
1、客户端执行setnx命令获取锁，返回值为1表示获取锁成功，返回为0表示当前key已经存在获取锁失败
2、获取到锁的客户端执行业务逻辑
3、客户端执行delete命令删除key释放锁
问题：当客户端执行setnx获取锁之后程序异常或崩溃，则无法通过delete命令来释放锁，此时就会导致锁一直无法释放导致死锁的情况
方案二：采用命令setnx命令获取锁，并通过expire命令给key添加过期时间，最后通过delete释放锁
步骤如下：
1、客户端执行setnx命令获取锁，返回值为1表示获取锁成功，返回为0表示当前key已经存在获取锁失败
2、获取到锁成功之后执行expire命令给key增加过期时间
3、获取到锁的客户端执行业务逻辑
4、客户端执行delete命令删除key释放锁
5、如果客户端异常或崩溃，可以通过key达到过期时间释放锁的效果
问题：setnx命令和expire命令是两个命令，所以两个操作不是原子操作，也就是说可能会存在setnx执行成功而expire执行失败的情况。
方案三：采用set方法加参数的方式达到加锁和过期时间原子操作
步骤如下：
1、客户端执行SET KEY VALUE [EX seconds] [PX milliseconds] [NX|XX]
其中EX表示过期时间单位为秒，seconds值为过期时长；PX表示过期时间单位为毫秒，millisseconds表示过期时长， NX表示只有当key不存在时才设置，XX表示只有当key存在时才设置
如 set test_key test_value PX 5 NX, 则表示设置test_key的值为test_value,并且设置了5秒的过期时间，同时只有当key不存在时才执行设置操作
2、获取到锁的客户端执行业务逻辑
3、客户端执行delete命令删除key释放锁
问题：由于delete命令没有做任何校验，所以会存在误删的情况，比如客户端A执行set方法获取锁成功然后执行业务逻辑，由于业务逻辑执行时间较长超过了过期时间，此时key会被删除，而此时客户端B执行set方法获取锁成功之后，客户端A的业务逻辑执行完成，开始执行delete操作释放锁，此时就会将客户端B获取到的锁给释放了，从而出现了误删锁的情况。
方案四：采用set方法加参数的方式加锁，在释放锁之前需要判断锁释放被当前线程占用
步骤如下：
1、客户端执行SET KEY VALUE PX 5 NX进行加锁
2、获取到锁的客户端执行业务逻辑
3、执行get(key)获取锁的值和之前设置过的值进行比较，如果期望的值和实际的值相同，则执行delete命令释放锁，如果期望的值和实际的值不相同，则不执行delete命令
问题：由于get操作和delete操作是两个命令，也就是说get和delete操作并非是原子操作，所以理论上还是会出现get操作的时候确实是期望的值，但是在delete之前实际的值已经变成了新的值了，此时还是会出现误删的情况。
方案五：采用set方法加参数的方式加锁，采用eval函数原子操作释放锁
redis的eval函数是执行一段Lua脚本，执行Lua脚本是原子执行的，所以可以通过eval函数在Lua脚本中判断当前锁是否可以释放并且最终释放锁。
步骤如下：
1、客户端执行SET KEY VALUE PX 5 NX进行加锁
2、获取到锁的客户端执行业务逻辑
3、执行Lua脚本来删除key，删除之前判断是否是原先的值，只不过eval方法是原子操作。
1 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
2 Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
上面的脚步含义是当前的值是否和期望的值相等，如果相等的话就执行delete操作删除key。
eval方法是将整个Lua脚本发送给redis执行，eval中的Lua会被当作原子操作交给redis来执行。
在单机部署redis的情况下，方案五可以达到原子性加锁和原子性释放锁的效果，从而可以达到分布式锁的效果。
2.2、redis集群模式分布式锁
在集群模式下，单机模式的方案五就会存在不可用的风险，比如以下场景：
客户端A从master节点获取锁成功，锁信息还没有同步到slave节点就宕机了，此时slave节点升级为新的master节点，客户端B从新的master节点获取锁成功，就导致同时有两个客户端都成功获取到了锁。显然就破坏了分布式锁同一时间只能有一个客户端获取锁的原则。
Redis官方提供了一种RedLock算法来实现了集群模式下的分布式锁的实现
2.2.1、RedLock算法
1、客户端获取当前Unix时间戳
2、客户端按顺序依次向N个redis节点设置相同的key和唯一的value获取锁的操作（获取锁的过程和单机模式获取锁的方式一样），且需要设置超时时间，超时时间需要小于锁过期的时间
3、如果满足以下几个条件，那么就表示获取锁成功，否则就表示获取锁失败
a、锁的可用时间 = (当前时间 - 请求时间) < 锁的过期时间；
b、超过半数以上的节点都获取锁成功，也就是满足a的要求
4、获取锁成功之后，锁的过期时间应该重新计算，锁的实际有效时间 = 锁的过期时间 - 获取锁消耗的时间
5、业务逻辑处理完成之后，需要将所有节点发送释放锁的命令，因为获取锁失败的节点可能已经占锁成功了，只是由于网络原因导致的失败。
RedLock理论上还是会存在一定的风险，比如当前有A、B、C三个redis节点，客户端1从A、B两个节点获取锁成功，C获取锁失败了；然后节点A宕机重启，且没有持久化。重启之后客户端2从A和C两个节点获取锁成功，就会出现客户端1和客户端2同时占有锁的情况。
对于这样的问题可以通过节点持久化来避免，或者宕机之后延迟重启，延迟时间为锁的过期时间，这样就可以保证重启之后锁已经完全被释放了。
2.2.2、RedLock算法的使用Redisson
目标Redis官方推荐的RedLock算法的实现为Redisson工具包，Redisson底层通过Netty框架实现，提供了redis不同部署环境下的分布式锁的实现。
1、Redisson的使用
a、添加maven依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.0</version>
</dependency>
b、Redisson支持单机模式、主从模式、哨兵模式、集群模式下的分布式锁的实现，只需要通过配置redis节点的主机信息即可。案例如下：
1 Config config = new Config(); 2 //1.单机模式 3 config.useSingleServer().setAddress("redis://122.51.172.201"); 4 //2.主从模式 5 Set<URI> slaveUrls = new HashSet<>(); 6 config.useMasterSlaveServers().setMasterAddress("redis://122.51.172.201").setSlaveAddresses(slaveUrls); 7 //3.哨兵模式 8 config.useSentinelServers().addSentinelAddress("redis://122.51.172.201").setMasterName("redis://122.51.172.201"); 9 //4.集群模式 10 config.useClusterServers().addNodeAddress("redis://122.51.172.201", "redis://122.51.172.201");
通过创建Config对象并使用指定的模式，然后添加对应模式下的redis节点信息即可
c、创建Redisson客户端并获取和释放锁
1 /**获取Redisson客户端 通过静态方法根据Config配置创建Redisson客户端*/ 2 RedissonClient client = Redisson.create(config); 3 /**设置分布式锁的key*/ 4 String lockKey = "test_lock_key"; 5 /**获取分布式锁*/ 6 RLock lock = client.getLock(lockKey); 7 8 boolean isLock = false; 9 try{ 10 while (!isLock) { 11 isLock = lock.tryLock(5, TimeUnit.SECONDS);//尝试获取分布式锁，有效期为5秒 12 if (isLock) { 13 //获取锁成功 14 System.out.println(Thread.currentThread().getName() + "获取锁成功"); 15 //TODO 执行业务逻辑 16 Thread.sleep(5000L); 17 } else { 18 System.out.println(Thread.currentThread().getName() + "获取锁失败"); 19 } 20 } 21 }catch (Exception e){ 22 e.printStackTrace(); 23 }finally { 24 if(isLock) { 25 System.out.println(Thread.currentThread().getName() + "释放锁"); 26 lock.unlock();//释放锁 27 } 28 }
可以发现使用起来十分便捷，对于客户端而言，只需要配置好redis集群的节点信息即可，通过客户端获取RLock对象，然后调用tryLock进行加锁，调用unlock进行释放锁即可。
2.2、Redisson的工作流程
Redisson整体工作流程如下图示：
加锁时，如果采用的是集群模式，会根据hash选择指定的节点，执行Lua脚本进行加锁操作，脚本如下：
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
通过Lua脚本执行可以保证命令的原子性，上述脚本的意思是：
如果key不存在则通过原子操作向hash中添加key并设置过期时间；
如果key已存在则通过原子操作向hash中的value执行自增操作并重新设置过期时间
该脚本一个中参数说明：
KEYS[1] : 表示解锁的key
ARGV[1] : 锁的key的过期时间
ARGV[2]：加锁的客户端唯一ID
1、首先第一个判断当前的key是否存在，如果存在表示锁被占用，如果锁不存在才会执行后面逻辑
2、再判断key对应的hash的value值是否是当前客户端的唯一ID标识，如果值则加锁成功，如果不是则加锁失败
3、当获取锁失败时会不断循环尝试获取锁
另外redisson实现的锁对应的数据结构时hash类型，key是锁的key，field为加锁的客户端标识，value为自增的数字，表示重入的次数。
释放锁是通过自减的方式修改value值，如果值为0了就直接删除key即可。