Redis系列二 - 通过redis命令和lua实现分布式锁

r囧r小猫 2021-06-10 20:42 298阅读 0赞

## 1. 概述 ##

在分布式系统，如果涉及到对相同资源的操作，则会经常涉及到使用分布锁。Redis为单进程单线程模式，通过Redis的命令SETNX，GET可以方便实现分布式锁。  
本文先通过redis命令实现分布式锁，介绍实现的主要业务逻辑，并指出其存在的不足之处。然后通过lua脚本实现分布式锁，弥补其存在的不足。最后通过ab对两者实现的锁进行压力测试，比较两者的性能。

## 2. 使用redis命令实现分布锁 ##

### 2.1. SETNX ###

语法： SETNX key value

*  如果key不存在，则存储(key:value）值，返回1
 *  如果key已经不存在，则不执行操作，返回0

因为这个命令的性质，多个线程竞争时只有一个线程能修改key的值。利用这一点可以实现锁的互斥功能。

### 2.2. ILock 和 DistributeLock ###

定义锁：主要方法有两个lock和unlock

/** 1. 定义锁 2. @author hry 3. */
    public interface ILock { 
        /** * 获取锁 * @param lock 锁名称 */
        void lock(String lock);
    
        /** * 释放锁 * @param lock 锁名称 */
        void unlock(String lock);
    }

ILock 具体实现类DistributeLock ：

1.  ThreadLocal threadId：通过threadId保存每个线程锁的UUID值，用于区分当前锁是否为自己所有，并且锁的value也存储此值
2.  lock主要逻辑：通过BoundValueOperations的setIfAbsent设置lockKey值(setIfAbsent其实就是封装了SETNX的命令），如果返回true，则表示已经获取锁;如果返回false，则进入等待
3.  unlock主要逻辑：通过redisTemplate.delete释放锁。在释放锁前，需要判断当前锁被当前线程所有，如果是，才执行释放锁，否则不执行
4.  避免死锁：如果线程A拿到锁后，在执行释放锁前，突然死掉了，则其它线程都无法再次获取锁，从而出现死锁。为了避免死锁，我们获取锁后，需要为锁设置一个有效期，即使锁的拥有者死掉了，此锁也可以被自动释放
5.  锁可重入：线程A拿到锁后，如果他再次执行lock，也可以再次拿到锁，而不是出现在等待锁的队列中; 如果当前线程已经获取锁，则再次请求锁则一定可以获取锁，否则会出现自己等待自己释放锁，从而出现死锁

详细的实现见代码：

/** * 通过redis实现分布锁 * @author hry * */
    public class DistributeLock implements ILock { 
        private static final Logger logger  = LoggerFactory.getLogger(DistributeLock.class);
    
        private static final int LOCK_MAX_EXIST_TIME = 5;  // 单位s，一个线程持有锁的最大时间
        private static final String LOCK_PREX = "lock_"; // 作为锁的key的前缀
    
        private StringRedisTemplate redisTemplate;
        private String lockPrex; // 做为锁key的前缀
        private int lockMaxExistTime; // 单位s，一个线程持有锁的最大时间
    
        private ThreadLocal<String> threadId = new ThreadLocal<String>();  // 线程变量
    
        public DistributeLock(StringRedisTemplate redisTemplate){
            this(redisTemplate, LOCK_PREX, LOCK_MAX_EXIST_TIME);
        }
    
        public DistributeLock(StringRedisTemplate redisTemplate, String lockPrex, int lockMaxExistTime){
            this.redisTemplate = redisTemplate;
            this.lockPrex = lockPrex;
            this.lockMaxExistTime = lockMaxExistTime;
        }
    
        @Override
        public void lock(String lock){
            Assert.notNull(lock, "lock can't be null!");
            String lockKey = getLockKey(lock);
            BoundValueOperations<String,String> keyBoundValueOperations = redisTemplate.boundValueOps(lockKey);     
            while(true){
                // 如果上次拿到锁的是自己，则本次也可以拿到锁：实现可重入
                String value = keyBoundValueOperations.get();
                // 根据传入的值，判断用户是否持有这个锁
                if(value != null && value.equals(String.valueOf(threadId.get()))){
                    // 重置过期时间
                    keyBoundValueOperations.expire(lockMaxExistTime, TimeUnit.SECONDS);
                    break;
                }
    
                if(keyBoundValueOperations.setIfAbsent(lockKey)){
                    // 每次获取锁时，必须重新生成id值
                    String keyUniqueId = UUID.randomUUID().toString(); // 生成key的唯一值
                    threadId.set(keyUniqueId);
                    // 显设置value，再设置过期日期，否则过期日期无效
                    keyBoundValueOperations.set(String.valueOf(keyUniqueId));
                    // 为了避免一个用户拿到锁后，进行过程中没有正常释放锁，这里设置一个默认过期实际，这段非常重要，如果没有，则会造成死锁
                    keyBoundValueOperations.expire(lockMaxExistTime, TimeUnit.SECONDS);
                    // 拿到锁后，跳出循环
                    break;
                }else{
                    try {
                        // 短暂休眠，nano避免出现活锁 
                        Thread.sleep(10, (int)(Math.random() * 500));
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            }
        }
    
    
        /** * 释放锁，同时要考虑当前锁是否为自己所有，以下情况会导致当前线程失去锁：线程执行的时间超过超时的时间，导致此锁被其它线程拿走; 此时用户不可以执行删除 * * 以上方法的缺陷： * a. 在本线程获取值，判断锁本线程所有，但是在执行删除前，锁超时被释放同时被另一个线程获取，则本操作释放锁 * * 最终解决方案 * a. 使用lua脚本，保证检测和删除在同一事物中 * */
        @Override
        public void unlock(final String lock) {
            final String lockKey = getLockKey(lock);
            BoundValueOperations<String,String> keyBoundValueOperations = redisTemplate.boundValueOps(lockKey);
            String lockValue = keyBoundValueOperations.get();
            if(!StringUtils.isEmpty(lockValue) && lockValue.equals(threadId.get())){
                redisTemplate.delete(lockKey);
            }else{
                logger.warn("key=[{}]已经变释放了，本次不执行释放. 线程Id[{}] ", lock, lockValue);  
            }
        }
    
        /** * 生成key * @param lock * @return */
        private String getLockKey(String lock){
            StringBuilder sb = new StringBuilder();
            sb.append(lockPrex).append(lock);
            return sb.toString();
        }
    
    }

### 2.3. ILockManager和SimpleRedisLockManager ###

ILockManager: 封装分布锁使用

public interface ILockManager { 
        /** * 通过加锁安全执行程序，无返回的数据 * @param lockKeyName key名称 * @param callback */
        void lockCallBack(String lockKeyName, SimpleCallBack callback);
        /** * 通过加锁安全执行程序，有返回数据 * @param lockKeyName * @param callback * @return */
        <T> T lockCallBackWithRtn(String lockKeyName, ReturnCallBack<T> callback);
    }

SimpleRedisLockManager  
ILockManager 的实现类，初始化上面实现的锁;  
此类封装了使用锁的公共代码，简化分布锁的使用。  
定义了两个回调方法，用于用户真正的业务逻辑实现

1.  SimpleCallBack: 无返回值的回调函数
2.  ReturnCallBack:有返回数据的回调函数

@Component
    public class SimpleRedisLockManager implements ILockManager {   
    
        @Autowired
        protected StringRedisTemplate redisTemplate;
    
        protected ILock distributeLock; // 分布锁
    
        @PostConstruct
        public void init(){
            // 初始化锁
            distributeLock = new DistributeLock(redisTemplate, "mylock_", 5);
        }
    
        @Override
        public void lockCallBack(String lockKeyName, SimpleCallBack callback){
            Assert.notNull("lockKeyName","lockKeyName 不能为空");
            Assert.notNull("callback","callback 不能为空");
            try{
                // 获取锁
                distributeLock.lock(lockKeyName);
                callback.execute();
            }finally{
                // 必须释放锁
                distributeLock.unlock(lockKeyName);
            }
        }
    
        @Override
        public <T> T lockCallBackWithRtn(String lockKeyName, ReturnCallBack<T> callback){
            Assert.notNull("lockKeyName","lockKeyName 不能为空");
            Assert.notNull("callback","callback 不能为空");
            try{
                // 获取锁
                distributeLock.lock(lockKeyName);
                return callback.execute();
            }finally{
                // 必须释放锁
                distributeLock.unlock(lockKeyName);
            }
        }
    }
    
    /** * 无返回值的回调函数 * @author hry * */
    public interface SimpleCallBack {
        void execute();
    }
    
    /** * 有返回数据的回调函数 * * @author hry * * @param <T> */
    public interface ReturnCallBack<T> {
        T execute();
    }

### 2.4. 真正使用锁的代码TestCtrl ###

使用非常简单

@Autowired
    private SimpleRedisLockManager simpleRedisLockManager;
    
     simpleRedisLockManager.lockCallBack("distributeLock" + ThreadLocalRandom.current().nextInt(1000), new SimpleCallBack() {
        @Override
        public void execute() {
            System.out.println("lockCallBack");
        }
    });

### 2.5. 以上锁实现依然存在不足之处 ###

1.  如果线程A拿到锁超过规定的时间还没有结束，则此时redis会自动释放锁。此时线程B拿到锁，则同时线程A和线程B同时拿到锁。对于这种情况，可以通过设置合理的超时时间解决。
2.  如果并发量很大，则可能出现多个线程同时拥有锁。这是因为在DistributeLock的lock和unlock方法都执行多条语句且这些语句不是事务的。比如线程A在unlock时，通过get方法得知自己拥有锁，然后他执行释放锁操作。在这两个操作之间，redis发现锁到期，自动删除锁，此时线程B申请并且得到锁。这时线程A才执行删除锁操作，则另外线程C也可以得到锁，此时线程B,C同时得到锁。这种情况可以通过下文的lua方法解决

## 3. 使用lua脚本实现分布锁 ##

上面的锁的实现，之所有有问题，关键是执行的多条语句不在一个事务中。而本节介绍的lua正好可以解决这个问题。  
redis 2.6.0之后的版本开始支持lua脚本。lua在redis使用详细见[这里][Link 1]。在redis中执行一个lua脚本时redis会将整个脚本作为一个整体执行，中间不会被其他命令插入，解决多个命令事物的问题。

### 3.1. lua锁脚本 ###

lock脚本:lock.lua

-- Set a lock
    -- 如果获取锁成功，则返回 1
    local key     = KEYS[1]
    local content = KEYS[2]
    local ttl     = ARGV[1]
    local lockSet = redis.call('setnx', key, content) if lockSet == 1 then redis.call('pexpire', key, ttl) -- redis.call('incr', "count") else -- 如果value相同，则认为是同一个线程的请求，则认为重入锁 local value = redis.call('get', key) if(value == content) then lockSet = 1;
        redis.call('pexpire', key, ttl) end end return lockSet

unlock脚本: unlock.lua

-- unlock key
    local key     = KEYS[1]
    local content = KEYS[2]
    local value = redis.call('get', key) if value == content then -- redis.call('decr', "count") return redis.call('del', key);
    end return 0

### 3.2. LuaDistributeLock ###

实现ILock接口  
LuaDistributeLock 实现业务逻辑和DistributeLock基本相同。这里在创建LuaDistributeLock时会调用init方法初始化lock和unlock脚本，生成相应的DefaultRedisScript对象，这两个对象可以被重复使用，不需要每次执行lock/unlock就需要初始化一个对象

public class LuaDistributeLock implements ILock { 
        private static final int LOCK_MAX_EXIST_TIME = 5;  // 单位s，一个线程持有锁的最大时间
        private static final String LOCK_PREX = "lock_"; // 作为锁的key的前缀
    
        private StringRedisTemplate redisTemplate;
        private String lockPrex; // 做为锁key的前缀
        private int lockMaxExistTime; // 单位s，一个线程持有锁的最大时间
        private DefaultRedisScript<Long> lockScript; // 锁脚本
        private DefaultRedisScript<Long> unlockScript; // 解锁脚本
    
        // 线程变量
        private ThreadLocal<String> threadKeyId = new ThreadLocal<String>(){
            @Override
            protected String initialValue() {
                return UUID.randomUUID().toString();
            }
        };  
    
        public LuaDistributeLock(StringRedisTemplate redisTemplate){
            this(redisTemplate, LOCK_PREX, LOCK_MAX_EXIST_TIME);
        }
    
        public LuaDistributeLock(StringRedisTemplate redisTemplate, String lockPrex, int lockMaxExistTime){
            this.redisTemplate = redisTemplate;
            this.lockPrex = lockPrex;
            this.lockMaxExistTime = lockMaxExistTime;
            // init
            init();
        }
    
        /** * 生成 */
        public void init() {
            // Lock script
            lockScript = new DefaultRedisScript<Long>();
            lockScript.setScriptSource(
                new ResourceScriptSource(new ClassPathResource("com/hry/spring/redis/distributedlock/lock/lock.lua")));
            lockScript.setResultType(Long.class);
            // unlock script
            unlockScript = new DefaultRedisScript<Long>();
            unlockScript.setScriptSource(
                new ResourceScriptSource(new ClassPathResource("com/hry/spring/redis/distributedlock/lock/unlock.lua")));
            unlockScript.setResultType(Long.class);
        }
    
        @Override
        public void lock(String lock2){
            Assert.notNull(lock2, "lock2 can't be null!");
            String lockKey = getLockKey(lock2);
            while(true){
                List<String> keyList = new ArrayList<String>();
                keyList.add(lockKey);
                keyList.add(threadKeyId.get());
                if(redisTemplate.execute(lockScript, keyList, String.valueOf(lockMaxExistTime * 1000)) > 0){
                    break;
                }else{
                    try {
                        // 短暂休眠，nano避免出现活锁 
                        Thread.sleep(10, (int)(Math.random() * 500));
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            }
        }
    
    
        /** * 释放锁，同时要考虑当前锁是否为自己所有，以下情况会导致当前线程失去锁：线程执行的时间超过超时的时间，导致此锁被其它线程拿走; 此时用户不可以执行删除 */
        @Override
        public void unlock(final String lock) {
            final String lockKey = getLockKey(lock);
            List<String> keyList = new ArrayList<String>();
            keyList.add(lockKey);
            keyList.add(threadKeyId.get());
            redisTemplate.execute(unlockScript, keyList);
        }
    
        /** * 生成key * @param lock * @return */
        private String getLockKey(String lock){
            StringBuilder sb = new StringBuilder();
            sb.append(lockPrex).append(lock);
            return sb.toString();
        }
    
    }

### 3.3. LuaLockRedisLockManager ###

继承上文的SimpleRedisLockManager ，重写init，初始化刚写的锁

@Component
    public class LuaLockRedisLockManager extends SimpleRedisLockManager { 
        @PostConstruct
        public void init(){
            // 初始化锁
            distributeLock = new LuaDistributeLock(redisTemplate, "mylock_", 5);
        }
    }

### 3.4. 真正使用锁的代码TestCtrl ###

用法和SimpleRedisLockManager相同

@Autowired
    private LuaLockRedisLockManager luaLockRedisLockManager;
    
    luaLockRedisLockManager.lockCallBack("distributeLock2" + ThreadLocalRandom.current().nextInt(1000), new SimpleCallBack() {
        @Override
        public void execute() {
            System.out.println("distributeLock2");
        }
    });

## 4. 性能比较 ##

下面通过压力测试工具ab，对这两种实现进行压力测试：100个并发线程，总共发送1000个请求  
simpleRedisLockManager： ab -n 1000 -c 100 [http://192.168.188.4:8080/distributeLock2][http_192.168.188.4_8080_distributeLock2]  
luaLockRedisLockManager： ab -n 1000 -c 100 [http://192.168.188.4:8080/distributeLock][http_192.168.188.4_8080_distributeLock]

详细数据如下  
![这里写图片描述][SouthEast]  
分析： lua脚本比redis的实现快很多，lua脚本的速度比使用普通命令快一倍。越是在高压力的情况下，lua的表现越好

## 5. 总结 ##

为了更好的使用锁，建议满足以下条件

1.  使用lua实现分布式锁
2.  根据业务逻辑设置合理的超时时间
3.  锁的粒度尽可能小，减少冲突

## 6. 代码 ##

[详细见代码][Link 2]

[Link 1]: https://redis.io/commands/eval
[http_192.168.188.4_8080_distributeLock2]: http://192.168.188.4:8080/distributeLock2
[http_192.168.188.4_8080_distributeLock]: http://192.168.188.4:8080/distributeLock
[SouthEast]: /images/20210520/c1b52ca480534f1c905cce4eb3977ddc.png
[Link 2]: https://github.com/hryou0922/spring_boot/tree/master/redis/src/main/java/com/hry/spring/redis/distributedlock