【分布式缓存系列】Redis实现分布式锁的正确姿式

1、前言

　　在咱们平常工做中，除了Spring和Mybatis外，用到最多无外乎分布式缓存框架——Redis。可是不少工做不少年的朋友对Redis还处于一个最基础的使用和认识。因此我就像把本身对分布式缓存的一些理解和应用整理一个系列，但愿能够帮助到你们加深对Redis的理解。本系列的文章思路先从Redis的应用开始。再解析Redis的内部实现原理。最后以常常会问到Redist相关的面试题为结尾。html

2、分布式锁的实现要点

为了实现分布式锁，须要确保锁同时知足如下四个条件：git

互斥性。在任意时刻，只有一个客户端能持有锁
不会发送死锁。即便一个客户端持有锁的期间崩溃而没有主动释放锁，也须要保证后续其余客户端可以加锁成功
加锁和解锁必须是同一个客户端，客户端本身不能把别人加的锁给释放了。
容错性。只要大部分的Redis节点正常运行，客户端就能够进行加锁和解锁操做。

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

3.1 加锁错误姿式

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

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

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

/** * 实现加锁的错误姿式 * @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操做在一个事务中进行，保证其原子性。

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

　　刚刚介绍完了错误的姿式后，从上面错误姿式中，咱们能够知道，要使用Redis实现分布式锁。加锁操做的正确姿式为：

使用setnx命令保证互斥性
须要设置锁的过时时间，避免死锁
setnx和设置过时时间须要保持原子性，避免在设置setnx成功以后在设置过时时间客户端崩溃致使死锁
加锁的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; } @Override public String acquire() { try { // 获取锁的超时时间，超过这个时间则放弃获取锁
            long end = System.currentTimeMillis() + acquireTimeout; // 随机生成一个value
            String 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; } @Override public 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的递减是顺序的。

5、总结

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

　　本文全部源码下载地址：https://github.com/learninghard-lizhi/common-util

　　补充：为了暂时知足你们好奇心，这里先抛出两篇文章已供你们了解在集群环境下上面实现方式的问题。

《基于Redis的分布式锁到底安全吗（上）？》

《基于Redis的分布式锁到底安全吗（下）？》