（五）Redis分布式锁的问题和实现

关于分布式的锁，大概有几个热点问题

关键点一：原子命令加锁。

对于 Redis 的加锁操作先set key，再设置 key 的过期时间，这样的话不是原子性操作。不是原子操作会带来什么问题，就不用我说了吧？

而在 2.6.12 版本后，可以通过向 Redis 发送下面的命令，实现原子性的加锁操作：

SET key random_value NX PX 30000

我们常用的redis客户端，Jedis、Lettuce都实现了这一命令方法，比如jedis的setnx()，Lettuce的setIfAbsent()。

关键点二：设置值的时候，放的是random_value。而不是你随便扔个“OK”进去。

先解释一下上面的命令中的几个参数的含义：

random_value：是由客户端生成的一个随机字符串，它要保证在足够长的一段时间内在所有客户端的所有获取锁的请求中都是唯一的。比如我们使用UUID。

NX：表示只有当要设置的 key 值不存在的时候才能 set 成功。这保证了只有第一个请求的客户端才能获得锁，而其它客户端在锁被释放之前都无法获得锁。

PX 30000：表示这个锁有一个 30 秒的自动过期时间。当然，这里 30 秒只是一个例子，客户端可以选择合适的过期时间。

那么为什么要使用 PX 30000 去设置一个超时时间？是怕进程 A 不讲道理啊，锁没等释放呢，万一崩了，直接原地把锁带走了，导致系统中谁也拿不到锁。

就算这样，还是不能保证万无一失。如果进程 A 又不讲道理，操作锁内资源超过笔者设置的超时时间，那么就会导致其他进程拿到锁，等进程 A 回来了，回手就是把其他进程的锁删了。这就引出了下一个关键点。

再解释一下为什么 value 需要设置为一个随机字符串。这也是第三个关键点。

关键点三：value 的值设置为随机数主要是为了更安全的释放锁，释放锁的时候需要检查 key 是否存在，且 key 对应的值是否和我指定的值一样，是一样的才能释放锁。所以可以看到这里有获取、判断、删除三个操作。

释放锁伪代码

String uuid = xxxx;
// 伪代码，具体实现看项目中用的连接工具
// 有的提供的方法名为set 有的叫setIfAbsent
set Test uuid NX PX 3000
try{
// biz handle....
} finally {
    // unlock
    if(uuid.equals(redisTool.get('key')){
        redisTool.del('key');
    }
}

这回看起来是不是稳了？相反，这回的问题更明显了，在 Finally 代码块中，Get 和 Del 并非原子操作，还是有进程安全问题。

进程安全问题会在什么场景下出现？我就先不回答了，感兴趣的可以在留言中交流下

为了保障原子性，我们需要用 lua 脚本。
那么删除锁的正确姿势之一，就是可以使用 Lua 脚本，通过 Redis 的 eval/evalsha 命令来。

下面简单看下一个释放锁有问题的代，相信也是很多人使用最多的方法

/**
     * 获取锁
     * @param key
     * @param expireSecond
     * @return
     */
    public boolean lock(String key, int expireSecond) {

        Jedis Jedis = null;
        try {
            Jedis = jedisPool.getResource();
            Long result = Jedis.setnx(key, "1");
            if(result == 1){
                Jedis.expire(key, expireSecond);
                return true;
            }
            return false;
        } catch (Exception ex) {
            logger.error("lock-setnx error:", ex);
            returnBrokenResource(Jedis);
            throw new TraweServiceException("获取锁出现异常：" + ex.getMessage(), ex);
        } finally {
            returnResource(Jedis);
        }
    }

    /**
     * 获取锁，如果没有获取到会再去尝试几次
     *
     * @param key 锁键值
     * @param expireSecond 锁过期时间
     * @param tryCount 尝试次数
     * @return 是否获得锁
     */
    @SuppressWarnings("static-access")
    public boolean tryLock(String key, int expireSecond, int tryCount) {
        if (lock(key, expireSecond)) {
            return true;
        }

        for (int i=0; i<tryCount; i++) {
            int sleepMills = RandomUtils.nextInt(20, 200);
            try {
                Thread.currentThread().sleep(sleepMills);
            } catch (InterruptedException e) {
                logger.error(e, e);
            }

            if (lock(key, expireSecond)) {
                return true;
            }
        }

        return false;
    }

    /**
     * 释放锁
     * @param key
     */
    public void unlock(String key) {
        del(key);
    }

实现Redis锁

网上大佬用jedis写的锁简单修改下使用lettuce实现

@Component
public class RedisDistributedLock {
    
    @Autowired
    @Resource(name="redisTemplateMaster")
    private RedisTemplate<Object,Object> redisTemplate;
    
    private final Logger logger = LoggerFactory.getLogger(RedisDistributedLock.class);
    
    private ThreadLocal<String> lockFlag = new ThreadLocal<String>();
    
    public static final String UNLOCK_LUA;

    static {
        StringBuilder sb = new StringBuilder();
        sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
        sb.append("then ");
        sb.append("    return redis.call(\"del\",KEYS[1]) ");
        sb.append("else ");
        sb.append("    return 0 ");
        sb.append("end ");
        UNLOCK_LUA = sb.toString();
    }


    public boolean lock(String key, long expire, int retryTimes, long sleepMillis) {
        boolean result = setRedis(key, expire);
        // 如果获取锁失败，按照传入的重试次数进行重试
        while((!result) && retryTimes-- > 0){
            try {
                logger.debug("lock failed, retrying..." + retryTimes);
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                return false;
            }
            result = setRedis(key , expire);
        }
        return result;
    }
    
    private boolean setRedis(final String key, final long expire ) {
        try{
            
            RedisCallback<Boolean> callback = (connection) -> {
                String uuid = UUID.randomUUID().toString();
                lockFlag.set(uuid);
                return connection.set(key.getBytes(Charset.forName("UTF-8")), uuid.getBytes(Charset.forName("UTF-8")), Expiration.milliseconds(expire), RedisStringCommands.SetOption.SET_IF_ABSENT);
            };
            return (Boolean)redisTemplate.execute(callback);
        } catch (Exception e) {
            logger.error("redis lock error.", e);
        }
        return false;
    }
    
    
    public boolean releaseLock(String key) {
        // 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
        try {
             RedisCallback<Boolean> callback = (connection) -> {
                String value = lockFlag.get();
                return connection.eval(UNLOCK_LUA.getBytes(), ReturnType.BOOLEAN ,1, key.getBytes(Charset.forName("UTF-8")), value.getBytes(Charset.forName("UTF-8")));
            };
             return (Boolean)redisTemplate.execute(callback);
        } catch (Exception e) {
            logger.error("release lock occured an exception", e);
        } finally {
            // 清除掉ThreadLocal中的数据，避免内存溢出
            lockFlag.remove();
        }
        return false;
    }
    
}