redisson分布式锁redLock源码解析【未完】

r囧r小猫 2022-06-09 02:00 389阅读 0赞

## 一、准备阶段 ##

### 1、原理 ###

> 一个客户端需要做如下操作来获取锁：
> 
> 1.获取当前时间（单位是毫秒）
> 
> 2.轮流用相同的key和随机值在N个节点上请求锁，在这一步里，客户端在每个master上请求锁时会有一个和总的锁释放时间相比小的多的超时时间。比如如果锁自动释放时间是10秒钟，那每个节点锁请求的超时时间可能是5-50毫秒的范围，这个可以防止一个客户端在某个宕掉的master节点上阻塞过长时间，如果一个master节点不可用了，我们应该尽快尝试下一个master节点
> 
> 3.客户端计算第二步中获取锁所花的时间，只有当客户端在大多数master节点上成功获取了锁（在这里是3个），而且总共消耗的时间不超过锁释放时间，这个锁就认为是获取成功了
> 
> 4.如果锁获取成功了，那现在锁自动释放时间就是最初的锁释放时间减去之前获取锁所消耗的时间
> 
> 5.如果锁获取失败了，不管是因为获取成功的锁不超过一半（N/2+1)还是因为总消耗时间超过了锁释放时间，客户端都会到每个master节点上释放锁，即便是那些他认为没有获取成功的锁。
> 
> In order to acquire the lock, the client performs the following operations:
> 
> 1、It gets the current time in milliseconds.
> 
> 2、It tries to acquire the lock in all the N instances sequentially, using the same key name and random value in all the instances. During step 2, when setting the lock in each instance, the client uses a timeout which is small compared to the total lock auto-release time in order to acquire it. For example if the auto-release time is 10 seconds, the timeout could be in the ~ 5-50 milliseconds range. This prevents the client from remaining blocked for a long time trying to talk with a Redis node which is down: if an instance is not available, we should try to talk with the next instance ASAP.
> 
> 3、The client computes how much time elapsed in order to acquire the lock, by subtracting from the current time the timestamp obtained in step 1. If and only if the client was able to acquire the lock in the majority of the instances (at least 3), and the total time elapsed to acquire the lock is less than lock validity time, the lock is considered to be acquired.
> 
> 4、If the lock was acquired, its validity time is considered to be the initial validity time minus the time elapsed, as computed in step 3.
> 
> 5、If the client failed to acquire the lock for some reason (either it was not able to lock N/2+1 instances or the validity time is negative), it will try to unlock all the instances (even the instances it believed it was not able to lock).

### 2、依赖包和版本 ###

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.3.2</version>
    </dependency>

### 3、源码中使用到的 Redis 命令 ###

SETNX key value (SET if Not eXists):当且仅当 key 不存在，将 key 的值设为 value ，并返回1；若给定的 key 已经存在，则 SETNX 不做任何动作，并返回0。详见：SETNX commond
    
    GETSET key value:将给定 key 的值设为 value ，并返回 key 的旧值 (old value)，当 key 存在但不是字符串类型时，返回一个错误，当key不存在时，返回nil。详见：GETSET commond
    
    GET key:返回 key 所关联的字符串值，如果 key 不存在那么返回 nil 。详见：GET Commond
    
    DEL key [KEY …]:删除给定的一个或多个 key ,不存在的 key 会被忽略,返回实际删除的key的个数（integer）。详见：DEL Commond
    
    HSET key field value：给一个key 设置一个{field=value}的组合值，如果key没有就直接赋值并返回1，如果field已有，那么就更新value的值，并返回0.详见：HSET Commond
    
    HEXISTS key field:当key 中存储着field的时候返回1，如果key或者field至少有一个不存在返回0。详见HEXISTS Commond
    
    HINCRBY key field increment:将存储在 key 中的哈希（Hash）对象中的指定字段 field 的值加上增量 increment。如果键
    key 不存在，一个保存了哈希对象的新建将被创建。如果字段 field 不存在，在进行当前操作前，其将被创建，且对应的值被置为 0。返回值是增量之后的值。详见：HINCRBY Commond
    
    PEXPIRE key milliseconds：设置存活时间，单位是毫秒。expire操作单位是秒。详见:PEXPIRE Commond
    
    PUBLISH channel message:向channel post一个message内容的消息，返回接收消息的客户端数。详见PUBLISH Commond

### 4、入口类 ###

public class RedisLockTest { 
    
        public static void main(String[] args) throws InterruptedException {
            Config config = new Config();
            config.useSentinelServers().addSentinelAddress("127.0.0.1:6479", "127.0.0.1:6489").setMasterName("master")
                    .setPassword("password").setDatabase(0);
            RedissonClient redissonClient = Redisson.create(config);
    
            RLock lock = redissonClient.getLock("LOCKER_PREFIX" + "test_lock");
            try {
                boolean isLock = lock.tryLock();
                // isLock = lock.tryLock(100, 1000, TimeUnit.SECONDS);
                if (isLock) {
                    //doBusiness();
                }
            } catch (Exception e) {
            } finally {
                lock.unlock();
            }
    
        }
    }

### 4、大体分两种，一种是无参，另一种是带过期时间的 ###

lock.tryLock() -> tryAcquireOnceAsync - tryLockInnerAsync

lock.tryLock(100, 1000, TimeUnit.SECONDS) ->tryLock(long waitTime, long leaseTime, TimeUnit unit)

## 二、 源码分析-无参 ##

### 1、RedissonLock.tryAcquireOnceAsync ###

private RFuture<Boolean> tryAcquireOnceAsync(long leaseTime, TimeUnit unit, final long threadId) {
            if (leaseTime != -1) {
                return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
            }
            RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
            ttlRemainingFuture.addListener(new FutureListener<Boolean>() {
                @Override
                public void operationComplete(Future<Boolean> future) throws Exception {
                    if (!future.isSuccess()) {
                        return;
                    }
    
                    Boolean ttlRemaining = future.getNow();
                    // lock acquired
                    if (ttlRemaining) {
                        scheduleExpirationRenewal(threadId);
                    }
                }
            });
            return ttlRemainingFuture;
        }

#### 1.1 、RedissonLock.tryLockInnerAsync ####

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> 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中添加一个key为test_lock的set，并且向set中添加一个field为线程id，值=1的键值对，表示此线程的重入次数为1
                          "redis.call('pexpire', KEYS[1], ARGV[1]); " +//设置set的过期时间，防止当前服务器出问题后导致死锁，return nil; end;返回nil 结束
                          "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; " + //返回nil，结束
                      "end; " +
                      "return redis.call('pttl', KEYS[1]);", //锁存在, 但不是当前线程加的锁，则返回锁的过期时间
                        Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
        }
    ````
    
    其中
    KEYS[1] 表示的是 getName() ，代表的是锁名 test_lock
    ARGV[1] 表示的是 internalLockLeaseTime 默认值是30s
    ARGV[2] 表示的是 getLockName(threadId) 代表的是 id:threadId 用锁对象id+线程id， 表示当前访问线程，用于区分不同服务器上的线程.
    
    
    
    
    
    <div class="se-preview-section-delimiter"></div>
    
    #### 1.2、RedissonLock.scheduleExpirationRenewal
    
    
    
    
    <div class="se-preview-section-delimiter"></div>
    
    ```java
    private void scheduleExpirationRenewal(final long threadId) {
            if (expirationRenewalMap.containsKey(getEntryName())) {
                return;
            }
    
            Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
    
                    RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                            "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                "return 1; " +
                            "end; " +
                            "return 0;",
                              Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    
                    future.addListener(new FutureListener<Boolean>() {
                        @Override
                        public void operationComplete(Future<Boolean> future) throws Exception {
                            expirationRenewalMap.remove(getEntryName());
                            if (!future.isSuccess()) {
                                log.error("Can't update lock " + getName() + " expiration", future.cause());
                                return;
                            }
    
                            if (future.getNow()) {
                                // reschedule itself
                                scheduleExpirationRenewal(threadId);
                            }
                        }
                    });
                }
            }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    
            if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
                task.cancel();
            }
        }

## 三、 源码分析-过期时间 ##

### 　1、RedissonLock.tryLock ###

@Override
        public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
            long time = unit.toMillis(waitTime);
            long current = System.currentTimeMillis();
            final long threadId = Thread.currentThread().getId();
            Long ttl = tryAcquire(leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                return true;
            }
    
            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
    
            current = System.currentTimeMillis();
            final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
            if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
                if (!subscribeFuture.cancel(false)) {
                    subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
                        @Override
                        public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                            if (subscribeFuture.isSuccess()) {
                                unsubscribe(subscribeFuture, threadId);
                            }
                        }
                    });
                }
                acquireFailed(threadId);
                return false;
            }
    
            try {
                time -= (System.currentTimeMillis() - current);
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
    
                while (true) {
                    long currentTime = System.currentTimeMillis();
                    ttl = tryAcquire(leaseTime, unit, threadId);
                    // lock acquired
                    if (ttl == null) {
                        return true;
                    }
    
                    time -= (System.currentTimeMillis() - currentTime);
                    if (time <= 0) {
                        acquireFailed(threadId);
                        return false;
                    }
    
                    // waiting for message
                    currentTime = System.currentTimeMillis();
                    if (ttl >= 0 && ttl < time) {
                        getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } else {
                        getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                    }
    
                    time -= (System.currentTimeMillis() - currentTime);
                    if (time <= 0) {
                        acquireFailed(threadId);
                        return false;
                    }
                }
            } finally {
                unsubscribe(subscribeFuture, threadId);
            }
    // return get(tryLockAsync(waitTime, leaseTime, unit));
        }

#### 1.1 tryAcquire ####

> tryLock -> tryAcquire -> tryAcquireAsync -> tryAcquireAsync -> tryLockInnerAsync

#### 1.2 subscribe ####

> tryLock -> subscribe(threadId) -> PUBSUB.subscribe(final String entryName, final String channelName, final ConnectionManager connectionManager)

PublishSubscribe.subscribe

public RFuture<E> subscribe(final String entryName, final String channelName, final ConnectionManager connectionManager) {
            final AtomicReference<Runnable> listenerHolder = new AtomicReference<Runnable>();
            final AsyncSemaphore semaphore = connectionManager.getSemaphore(channelName);
            final RPromise<E> newPromise = new PromiseDelegator<E>(connectionManager.<E>newPromise()) {
                @Override
                public boolean cancel(boolean mayInterruptIfRunning) {
                    return semaphore.remove(listenerHolder.get());
                }
            };
    
            Runnable listener = new Runnable() {
    
                @Override
                public void run() {
                    E entry = entries.get(entryName);
                    if (entry != null) {
                        entry.aquire();
                        semaphore.release();
                        entry.getPromise().addListener(new TransferListener<E>(newPromise));
                        return;
                    }
    
                    E value = createEntry(newPromise);
                    value.aquire();
    
                    E oldValue = entries.putIfAbsent(entryName, value);
                    if (oldValue != null) {
                        oldValue.aquire();
                        semaphore.release();
                        oldValue.getPromise().addListener(new TransferListener<E>(newPromise));
                        return;
                    }
                    // 1.2.1
                    RedisPubSubListener<Object> listener = createListener(channelName, value);
                    // 1.2.2
                    connectionManager.subscribe(LongCodec.INSTANCE, channelName, listener, semaphore);
                }
            };
            semaphore.acquire(listener);
            listenerHolder.set(listener);
    
            return newPromise;
        }

##### 1.2.1 PublishSubscribe.createListener #####

> tryLock -> subscribe(threadId) -> PUBSUB.subscribe -> PublishSubscribe.createListener

PublishSubscribe.createListener

private RedisPubSubListener<Object> createListener(final String channelName, final E value) {
            RedisPubSubListener<Object> listener = new BaseRedisPubSubListener() {
    
                @Override
                public void onMessage(String channel, Object message) {
                    if (!channelName.equals(channel)) {
                        return;
                    }
    
                    PublishSubscribe.this.onMessage(value, (Long)message);
                }
    
                @Override
                public boolean onStatus(PubSubType type, String channel) {
                    if (!channelName.equals(channel)) {
                        return false;
                    }
    
                    if (type == PubSubType.SUBSCRIBE) {
                        value.getPromise().trySuccess(value);
                        return true;
                    }
                    return false;
                }
    
            };
            return listener;
        }

##### 1.2.2 PublishSubscribe.createListener #####

> tryLock -> subscribe(threadId) -> PUBSUB.subscribe -> PublishSubscribe.createListener -> LockPubSub.onMessage

public class LockPubSub extends PublishSubscribe<RedissonLockEntry> { 
    
        public static final Long unlockMessage = 0L;
    
        @Override
        protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) {
            return new RedissonLockEntry(newPromise);
        }
    
        @Override
        protected void onMessage(RedissonLockEntry value, Long message) {
            if (message.equals(unlockMessage)) {
                // 释放一个许可，唤醒等待的entry.getLatch().tryAcquire去再次尝试获取锁。
                value.getLatch().release();
                // 如果entry还有其他Listeners回调，也唤醒执行。
                while (true) {
                    Runnable runnableToExecute = null;
                    synchronized (value) {
                        Runnable runnable = value.getListeners().poll();
                        if (runnable != null) {
                            if (value.getLatch().tryAcquire()) {
                                runnableToExecute = runnable;
                            } else {
                                value.addListener(runnable);
                            }
                        }
                    }
    
                    if (runnableToExecute != null) {
                        runnableToExecute.run();
                    } else {
                        return;
                    }
                }
            }
        }
    }

##### 1.2.2 MasterSlaveConnectionManager.subscribe #####

> tryLock -> subscribe(threadId) -> PUBSUB.subscribe -> MasterSlaveConnectionManager.subscribe

public RFuture<PubSubConnectionEntry> subscribe(Codec codec, String channelName, RedisPubSubListener<?> listener, AsyncSemaphore semaphore) {
            RPromise<PubSubConnectionEntry> promise = newPromise();
            subscribe(codec, channelName, listener, promise, PubSubType.SUBSCRIBE, semaphore);
            return promise;
        }

##### 1.2.2.1 MasterSlaveConnectionManager.subscribe #####

> tryLock -> subscribe(threadId) -> PUBSUB.subscribe -> MasterSlaveConnectionManager.subscribe -> MasterSlaveConnectionManager.subscribe(final Codec codec, final String channelName, final RedisPubSubListener

private void subscribe(final Codec codec, final String channelName, final RedisPubSubListener<?> listener,
                final RPromise<PubSubConnectionEntry> promise, final PubSubType type, final AsyncSemaphore lock) {
            final PubSubConnectionEntry connEntry = name2PubSubConnection.get(channelName);
            if (connEntry != null) {
                connEntry.addListener(channelName, listener);
                connEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                    @Override
                    public void operationComplete(Future<Void> future) throws Exception {
                        lock.release();
                        promise.trySuccess(connEntry);
                    }
                });
                return;
            }
    
            freePubSubLock.acquire(new Runnable() {
    
                @Override
                public void run() {
                    if (promise.isDone()) {
                        return;
                    }
    
                    final PubSubConnectionEntry freeEntry = freePubSubConnections.peek();
                    if (freeEntry == null) {
                        connect(codec, channelName, listener, promise, type, lock);
                        return;
                    }
    
                    int remainFreeAmount = freeEntry.tryAcquire();
                    if (remainFreeAmount == -1) {
                        throw new IllegalStateException();
                    }
    
                    final PubSubConnectionEntry oldEntry = name2PubSubConnection.putIfAbsent(channelName, freeEntry);
                    if (oldEntry != null) {
                        freeEntry.release();
                        freePubSubLock.release();
    
                        oldEntry.addListener(channelName, listener);
                        oldEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                            @Override
                            public void operationComplete(Future<Void> future) throws Exception {
                                lock.release();
                                promise.trySuccess(oldEntry);
                            }
                        });
                        return;
                    }
    
                    if (remainFreeAmount == 0) {
                        freePubSubConnections.poll();
                    }
                    freePubSubLock.release();
    
                    freeEntry.addListener(channelName, listener);
                    freeEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener<Void>() {
                        @Override
                        public void operationComplete(Future<Void> future) throws Exception {
                            lock.release();
                            promise.trySuccess(freeEntry);
                        }
                    });
    
                    if (PubSubType.PSUBSCRIBE == type) {
                        freeEntry.psubscribe(codec, channelName);
                    } else {
                        freeEntry.subscribe(codec, channelName);
                    }
                }
    
            });
        }

## 三、RedissonLock解锁 unlock源码 ##

@Override
        public void unlock() {
            Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                            "if (redis.call('exists', KEYS[1]) == 0) then " +//如果锁已经不存在(可能是因为过期导致不存在，也可能是因为已经解锁)
                                "redis.call('publish', KEYS[2], ARGV[1]); " +//则发布锁解除的消息
                                "return 1; " + //返回1结束
                            "end;" +
                            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + //如果锁存在，但是若果当前线程不是加锁的线
                                "return nil;" + //则直接返回nil 结束
                            "end; " +
                            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + //如果是锁是当前线程所添加，定义变量counter，表示当前线程的重入次数-1,即直接将重入次数-1
                            "if (counter > 0) then " + //如果重入次数大于0，表示该线程还有其他任务需要执行
                                "redis.call('pexpire', KEYS[1], ARGV[2]); " + //则重新设置该锁的有效时间
                                "return 0; " + //返回0结束
                            "else " +
                                "redis.call('del', KEYS[1]); " + //否则表示该线程执行结束，删除该锁
                                "redis.call('publish', KEYS[2], ARGV[1]); " + //并且发布该锁解除的消息
                                "return 1; "+ //返回1结束
                            "end; " +
                            "return nil;", //其他情况返回nil并结束
                            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(Thread.currentThread().getId()));
            if (opStatus == null) {
                throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + Thread.currentThread().getId());
            }
            if (opStatus) {
                cancelExpirationRenewal();
            }
        }

其中  
KEYS\[1\] 表是的是getName() 代表锁名test\_lock  
KEYS\[2\] 表示getChanelName() 表示的是发布订阅过程中使用的Chanel  
ARGV\[1\] 表示的是LockPubSub.unLockMessage 是解锁消息，实际代表的是数字 0，代表解锁消息  
ARGV\[2\] 表示的是internalLockLeaseTime 默认的有效时间 30s  
ARGV\[3\] 表示的是getLockName(thread.currentThread().getId())，是当前锁id+线程id

## RedissonLock强制解锁源码 ##

@Override
        public void forceUnlock() {
            get(forceUnlockAsync());
        }
    
        @Override
        public RFuture<Boolean> forceUnlockAsync() {
            cancelExpirationRenewal();
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                    "if (redis.call('del', KEYS[1]) == 1) then "
                    + "redis.call('publish', KEYS[2], ARGV[1]); "
                    + "return 1 "
                    + "else "
                    + "return 0 "
                    + "end",
                    Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage);
        }

以上是强制解锁的源码,在源码中并没有找到forceUnlock()被调用的痕迹(也有可能是我没有找对),但是forceUnlockAsync()方法被调用的地方很多，大多都是在清理资源时删除锁。此部分比较简单粗暴，删除锁成功则并发布锁被删除的消息，返回1结束，否则返回0结束。

参考：  
[http://tech.lede.com/2017/03/08/rd/server/Redisson/][http_tech.lede.com_2017_03_08_rd_server_Redisson]

[http_tech.lede.com_2017_03_08_rd_server_Redisson]: http://tech.lede.com/2017/03/08/rd/server/Redisson/