分布式锁用 Redis 还是 Zookeeper?

作者：jianfeng来源：石杉的架构笔记

为什么用分布式锁？

基于Redis实现分布式锁

// 获取锁
// NX是指如果key不存在就成功，key存在返回false，PX可以指定过期时间
SET anyLock unique_value NX PX 30000


// 释放锁：通过执行一段lua脚本
// 释放锁涉及到两条指令，这两条指令不是原子性的
// 需要用到redis的lua脚本支持特性，redis执行lua脚本是原子性的
if redis.call(get,KEYS[1]) == ARGV[1] then
return redis.call(del,KEYS[1])
else
return 0
end

• 一定要用SET key value NX PX milliseconds 命令

  如果不用，先设置了值，再设置过期时间，这个不是原子性操作，有可能在设置过期时间之前宕机，会造成死锁(key永久存在)

• value要具有唯一性

  这个是为了在解锁的时候，需要验证value是和加锁的一致才删除key。

  这是避免了一种情况：假设A获取了锁，过期时间30s，此时35s之后，锁已经自动释放了，A去释放锁，但是此时可能B获取了锁。A客户端就不能删除B的锁了。

• 单机模式

• master-slave + sentinel选举模式

• redis cluster模式

• 获取当前时间戳，单位是毫秒

• 轮流尝试在每个master节点上创建锁，过期时间设置较短，一般就几十毫秒

• 尝试在大多数节点上建立一个锁，比如5个节点就要求是3个节点（n / 2 +1）

• 客户端计算建立好锁的时间，如果建立锁的时间小于超时时间，就算建立成功了

• 要是锁建立失败了，那么就依次删除这个锁

• 只要别人建立了一把分布式锁，你就得不断轮询去尝试获取锁

另一种方式：Redisson

• SET anyLock unique_value NX PX 30000

Config config = new Config();
config.useClusterServers()
.addNodeAddress(redis://192.168.31.101:7001)
.addNodeAddress(redis://192.168.31.101:7002)
.addNodeAddress(redis://192.168.31.101:7003)
.addNodeAddress(redis://192.168.31.102:7001)
.addNodeAddress(redis://192.168.31.102:7002)
.addNodeAddress(redis://192.168.31.102:7003);

RedissonClient redisson = Redisson.create(config);


RLock lock = redisson.getLock(anyLock);
lock.lock();
lock.unlock();

• redisson所有指令都通过lua脚本执行，redis支持lua脚本原子性执行

• redisson设置一个key的默认过期时间为30s,如果某个客户端持有一个锁超过了30s怎么办？

  redisson中有一个watchdog的概念，翻译过来就是看门狗，它会在你获取锁之后，每隔10秒帮你把key的超时时间设为30s

  这样的话，就算一直持有锁也不会出现key过期了，其他线程获取到锁的问题了。

• redisson的“看门狗”逻辑保证了没有死锁发生。

  (如果机器宕机了，看门狗也就没了。此时就不会延长key的过期时间，到了30s之后就会自动过期了，其他线程可以获取到锁)

// 加锁逻辑
private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 调用一段lua脚本，设置一些key、过期时间
    RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.addListener(new FutureListener() {
        @Override
        public void operationComplete(Future future) throws Exception {
            if (!future.isSuccess()) {
                return;
            }

            Long ttlRemaining = future.getNow();
            // lock acquired
            if (ttlRemaining == null) {
                // 看门狗逻辑
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}


 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              if (redis.call('exists', KEYS[1]) == 0) then  +
                  redis.call('hset', KEYS[1], ARGV[2], 1);  +
                  redis.call('pexpire', KEYS[1], ARGV[1]);  +
                  return nil;  +
              end;  +
              if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then  +
                  redis.call('hincrby', KEYS[1], ARGV[2], 1);  +
                  redis.call('pexpire', KEYS[1], ARGV[1]);  +
                  return nil;  +
              end;  +
              return redis.call('pttl', KEYS[1]);,
                Collections.