一、简介

单机器环境下，可以通过锁来解决共享资源的竞争问题；而在分布式集群环境下，机器与机器之间的资源竞争则需要依赖Redis、ZooKeeper等中间件去协调。

简单总结一下自己对Redis分布式锁的一些理解

二、代码实现

第一步先是获取锁，通过setnx操作，设置指定key及其过期时间。较新的版本支持setnx和过期时间的原子性操作，如果是较老的版本，只能通过Lua脚本来完成这一步。

若setnx返回true，则代表成功获取到锁，否则没有获取到锁。

boolean setIfAbsent = false;
try {
		setIfAbsent = redisTemplate.opsForValue().setIfAbsent(lockKey, clientId, lockTimeout, TimeUnit.MILLISECONDS);
} catch (Exception e) {
		log.error("redis lock occur error", e);
}
if (!setIfAbsent) {
		return;
}

获取到锁以后开始执行task任务

执行完成后需要释放锁

try {
		redisTemplate.delete(lockKey);
} catch (Exception e) {
		logger.error("release lock occur error", e);
}

乍一看没什么问题，但是在某种情况下是有问题的，假如A获取到了锁开始执行task，但是task执行时间很长，超过了lockTimeout时间，key过期了，此时B尝试获取锁，成功获取到了，开始执行task，A完成了task执行了redisTemplate.delete操作，把B的锁给删除了，那么锁的功能就失效了。

在这种情况下就需要保证获取锁的那一方的锁不会被其他方释放，锁只能被获取方释放或过期自动释放，不能有其他被释放的情况发生。因此可以通过锁key时设置指定的value，只有相同的value才能释放锁。而这个value的比较与锁的释放必须保证原子性，需要通过Lua脚本来实现。

try {
		redisTemplate.execute(
				RedisScript.of(
						"if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
      			Long.class
    		),
    		Collections.singletonList(lockKey),
    		clientId);
} catch (Exception e) {
  	log.error("release lock occur error", e);
}

完整的代码示例如下：

public class RedisLock {

    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * @param lockKey     锁key
     * @param clientId    锁value
     * @param lockTimeout 锁超时时间（毫秒）
     * @param task        执行类
     */
    public void lock(String lockKey, String clientId, long lockTimeout, Task task) {
        boolean setIfAbsent = false;
        try {
            setIfAbsent = redisTemplate.opsForValue().
                    setIfAbsent(lockKey, clientId, lockTimeout, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            log.error("redis lock occur error", e);
        }
        if (!setIfAbsent) {
            return;
        }
        try {
            task.action();
        } catch (Exception e) {
            log.error("task action occur error", e);
        }
        try {
            redisTemplate.execute(
                    RedisScript.of(
                            "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
                            Long.class
                    ),
                    Collections.singletonList(lockKey),
                    clientId);
        } catch (Exception e) {
            log.error("release lock occur error", e);
        }
    }

}

三、不足

• Redis锁并没有极高的可靠性，Redis一般采用主从方式部署，如果在加锁过程中，Redis主节点挂掉，加锁的指令只在主节点执行完成，但是还未同步到从节点，此时从节点变更为主节点，但是锁信息并没有得到同步，锁的功能就失效了。

• 上述的方案，如果task执行出现耗时超过锁的超时时间，仍然会出现task并发执行的情况，这种情况需要根据实际业务场景决定，是否允许此类情况发生，如果不允许是否需要增加超时时间，或者提供锁延长超时时间的机制。