一、前言

　　在我们日常工作中，除了Spring和Mybatis外，用到最多无外乎分布式缓存框架——Redis。但是很多工作很多年的朋友对Redis还处于一个最基础的使用和认识。所以我就像把自己对分布式缓存的一些理解和应用整理一个系列，希望可以帮助到大家加深对Redis的理解。本系列的文章思路先从Redis的应用开始。再解析Redis的内部实现原理。最后以经常会问到Redist相关的面试题为结尾。

二、分布式锁的实现要点

为了实现分布式锁，需要确保锁同时满足以下四个条件：

1. 互斥性。在任意时刻，只有一个客户端能持有锁

2. 不会发送死锁。即使一个客户端持有锁的期间崩溃而没有主动释放锁，也需要保证后续其他客户端能够加锁成功

3. 加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给释放了。

4. 容错性。只要大部分的Redis节点正常运行，客户端就可以进行加锁和解锁操作。

三、Redis实现分布式锁的错误姿势

3.1 加锁错误姿势

　　在讲解使用Redis实现分布式锁的正确姿势之前，我们有必要来看下错误实现方式。

　　首先，为了保证互斥性和不会发送死锁2个条件，所以我们在加锁操作的时候，需要使用SETNX指令来保证互斥性——只有一个客户端能够持有锁。为了保证不会发送死锁，需要给锁加一个过期时间，这样就可以保证即使持有锁的客户端期间崩溃了也不会一直不释放锁。

　　为了保证这2个条件，有些人错误的实现会用如下代码来实现加锁操作：

/*** 实现加锁的错误姿势* @param jedis* @param lockKey* @param requestId* @param expireTime*/public static void wrongGetLock1(Jedis jedis, String lockKey, String requestId, int expireTime) {Long result = jedis.setnx(lockKey, requestId);if (result == 1) {// 若在这里程序突然崩溃，则无法设置过期时间，将发生死锁jedis.expire(lockKey, expireTime);}}

　　可能一些初学者还没看出以上实现加锁操作的错误原因。这样我们解释下。setnx 和expire是两条Redis指令，不具备原子性，如果程序在执行完setnx之后突然崩溃，导致没有设置锁的过期时间，从而就导致死锁了。因为这个客户端持有的所有不会被其他客户端释放，持有锁的客户端又崩溃了，也不会主动释放。从而该锁永远不会释放，导致其他客户端也获得不能锁。从而其他客户端一直阻塞。所以针对该代码正确姿势应该保证setnx和expire原子性。

　　实现加锁操作的错误姿势2。具体实现如下代码所示

/*** 实现加锁的错误姿势2* @param jedis* @param lockKey* @param expireTime* @return*/public static boolean wrongGetLock2(Jedis jedis, String lockKey, int expireTime) {long expires = System.currentTimeMillis() + expireTime;String expiresStr = String.valueOf(expires);// 如果当前锁不存在，返回加锁成功if (jedis.setnx(lockKey, expiresStr) == 1) {return true;}// 如果锁存在，获取锁的过期时间String currentValueStr = jedis.get(lockKey);if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {// 锁已过期，获取上一个锁的过期时间，并设置现在锁的过期时间String oldValueStr = jedis.getSet(lockKey, expiresStr);if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {// 考虑多线程并发的情况，只有一个线程的设置值和当前值相同，它才有权利加锁return true;}}// 其他情况，一律返回加锁失败return false;}

　　这个加锁操作咋一看没有毛病对吧。那以上这段代码的问题毛病出在哪里呢？

　　1. 由于客户端自己生成过期时间，所以需要强制要求分布式环境下所有客户端的时间必须同步。

　　2. 当锁过期的时候，如果多个客户端同时执行jedis.getSet()方法，虽然最终只有一个客户端加锁，但是这个客户端的锁的过期时间可能被其他客户端覆盖。不具备加锁和解锁必须是同一个客户端的特性。解决上面这段代码的方式就是为每个客户端加锁添加一个唯一标示，已确保加锁和解锁操作是来自同一个客户端。

3.2 解锁错误姿势

　　分布式锁的实现无法就2个方法，一个加锁，一个就是解锁。下面我们来看下解锁的错误姿势。

　　错误姿势1.

/*** 解锁错误姿势1* @param jedis* @param lockKey*/public static void wrongReleaseLock1(Jedis jedis, String lockKey) {jedis.del(lockKey);}

　　上面实现是最简单直接的解锁方式，这种不先判断拥有者而直接解锁的方式，会导致任何客户端都可以随时解锁。即使这把锁不是它上锁的。

　　错误姿势2：

/*** 解锁错误姿势2* @param jedis* @param lockKey* @param requestId*/public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {// 判断加锁与解锁是不是同一个客户端if (requestId.equals(jedis.get(lockKey))) {// 若在此时，这把锁突然不是这个客户端的，则会误解锁jedis.del(lockKey);}

　　既然错误姿势1中没有判断锁的拥有者，那姿势2中判断了拥有者，那错误原因又在哪里呢？答案又是原子性上面。因为判断和删除不是一个原子性操作。在并发的时候很可能发生解除了别的客户端加的锁。具体场景有：客户端A加锁，一段时间之后客户端A进行解锁操作时，在执行jedis.del()之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del方法，则客户端A将客户端B的锁给解除了。从而不也不满足加锁和解锁必须是同一个客户端特性。解决思路就是需要保证GET和DEL操作在一个事务中进行，保证其原子性。

四、Redis实现分布式锁的正确姿势

　　刚刚介绍完了错误的姿势后，从上面错误姿势中，我们可以知道，要使用Redis实现分布式锁。加锁操作的正确姿势为：

1. 使用setnx命令保证互斥性

2. 需要设置锁的过期时间，避免死锁

3. setnx和设置过期时间需要保持原子性，避免在设置setnx成功之后在设置过期时间客户端崩溃导致死锁

4. 加锁的Value 值为一个唯一标示。可以采用UUID作为唯一标示。加锁成功后需要把唯一标示返回给客户端来用来客户端进行解锁操作

　　解锁的正确姿势为：

　　1. 需要拿加锁成功的唯一标示要进行解锁，从而保证加锁和解锁的是同一个客户端

　　2. 解锁操作需要比较唯一标示是否相等，相等再执行删除操作。这2个操作可以采用Lua脚本方式使2个命令的原子性。

　　Redis分布式锁实现的正确姿势的实现代码：

public interface DistributedLock {/*** 获取锁* @author zhi.li* @return 锁标识*/String acquire();/*** 释放锁* @author zhi.li* @param indentifier* @return*/boolean release(String indentifier);
}/*** @author zhi.li* @Description* @created 2019/1/1 20:32*/
@Slf4j
public class RedisDistributedLock implements DistributedLock{private static final String LOCK_SUCCESS = "OK";private static final Long RELEASE_SUCCESS = 1L;private static final String SET_IF_NOT_EXIST = "NX";private static final String SET_WITH_EXPIRE_TIME = "PX";/*** redis 客户端*/private Jedis jedis;/*** 分布式锁的键值*/private String lockKey;/*** 锁的超时时间 10s*/int expireTime = 10 * 1000;/*** 锁等待，防止线程饥饿*/int acquireTimeout  = 1 * 1000;/*** 获取指定键值的锁* @param jedis jedis Redis客户端* @param lockKey 锁的键值*/public RedisDistributedLock(Jedis jedis, String lockKey) {this.jedis = jedis;this.lockKey = lockKey;}/*** 获取指定键值的锁,同时设置获取锁超时时间* @param jedis jedis Redis客户端* @param lockKey 锁的键值* @param acquireTimeout 获取锁超时时间*/public RedisDistributedLock(Jedis jedis,String lockKey, int acquireTimeout) {this.jedis = jedis;this.lockKey = lockKey;this.acquireTimeout = acquireTimeout;}/*** 获取指定键值的锁,同时设置获取锁超时时间和锁过期时间* @param jedis jedis Redis客户端* @param lockKey 锁的键值* @param acquireTimeout 获取锁超时时间* @param expireTime 锁失效时间*/public RedisDistributedLock(Jedis jedis, String lockKey, int acquireTimeout, int expireTime) {this.jedis = jedis;this.lockKey = lockKey;this.acquireTimeout = acquireTimeout;this.expireTime = expireTime;}@Overridepublic String acquire() {try {// 获取锁的超时时间，超过这个时间则放弃获取锁long end = System.currentTimeMillis() + acquireTimeout;// 随机生成一个valueString requireToken = UUID.randomUUID().toString();while (System.currentTimeMillis() < end) {String result = jedis.set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);if (LOCK_SUCCESS.equals(result)) {return requireToken;}try {Thread.sleep(100);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}} catch (Exception e) {log.error("acquire lock due to error", e);}return null;}@Overridepublic boolean release(String identify) {if(identify == null){return false;}String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";Object result = new Object();try {result = jedis.eval(script, Collections.singletonList(lockKey),Collections.singletonList(identify));if (RELEASE_SUCCESS.equals(result)) {log.info("release lock success, requestToken:{}", identify);return true;}}catch (Exception e){log.error("release lock due to error",e);}finally {if(jedis != null){jedis.close();}}log.info("release lock failed, requestToken:{}, result:{}", identify, result);return false;}
}下面就以秒杀库存数量为场景，测试下上面实现的分布式锁的效果。具体测试代码如下：public class RedisDistributedLockTest {static int n = 500;public static void secskill() {System.out.println(--n);}public static void main(String[] args) {Runnable runnable = () -> {RedisDistributedLock lock = null;String unLockIdentify = null;try {Jedis conn = new Jedis("127.0.0.1",6379);lock = new RedisDistributedLock(conn, "test1");unLockIdentify = lock.acquire();System.out.println(Thread.currentThread().getName() + "正在运行");secskill();} finally {if (lock != null) {lock.release(unLockIdentify);}}};for (int i = 0; i < 10; i++) {Thread t = new Thread(runnable);t.start();}}
}

　　运行效果如下图所示。从图中可以看出，同一个资源在同一个时刻只能被一个线程获取，从而保证了库存数量N的递减是顺序的。

五、总结

　　这样是不是已经完美使用Redis实现了分布式锁呢？答案是并没有结束。上面的实现代码只是针对单机的Redis没问题。但是现实生产中大部分都是集群的或者是主备的。但上面的实现姿势在集群或者主备情况下会有相应的问题。这里先买一个关子，在后面一篇文章将详细分析集群或者主备环境下Redis分布式锁的实现方式。

在此我向大家推荐一个架构学习交流圈：830478757  帮助突破瓶颈 提升思维能力