分布式锁—5.Redisson的读写锁二
大纲
1.Redisson读写锁RedissonReadWriteLock概述
2.读锁RedissonReadLock的获取读锁逻辑
3.写锁RedissonWriteLock的获取写锁逻辑
4.读锁RedissonReadLock的读读不互斥逻辑
5.RedissonReadLock和RedissonWriteLock的读写互斥逻辑
6.写锁RedissonWriteLock的写写互斥逻辑
7.写锁RedissonWriteLock的可重入逻辑
8.读锁RedissonReadLock的释放读锁逻辑
9.写锁RedissonWriteLock的释放写锁逻辑
6.写锁RedissonWriteLock的写写互斥逻辑
(1)不同客户端线程先加写锁的情况
(2)不同客户端线程再加写锁的情况
(1)不同客户端线程先加写锁的情况
假设客户端A(UUID1:ThreadID1)先加写锁：
//传入参数
KEYS[1] = myLock
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1:write
//执行结果
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1
}(2)不同客户端线程再加写锁的情况
假设客户端B(UUID2:ThreadID2)再加写锁：首先执行命令"hget myLock mode"发现mode = write，说明已有线程加了写锁。然后继续执行命令"hexists myLock UUID2:ThreadID2:write"，判断已加的写锁是否是当前客户端B(UUID2:ThreadID2)加的。由于已加的写锁是客户端A(UUID1:ThreadID1)加的，所以判断不通过。于是执行"pttl myLock"返回myLock的剩余过期时间。这样会导致客户端B加写锁失败，于是会在while循环阻塞和重试加写锁，从而实现不同客户端线程的写锁和写锁的互斥。
public class RedissonWriteLock extends RedissonLock implements RLock {
...
@Override
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
//执行命令"hget myLock mode"，尝试获取一个Hash值mode
"local mode = redis.call('hget', KEYS[1], 'mode'); " +
//获取不到，说明没有加读锁或者写锁
"if (mode == false) then " +
"redis.call('hset', KEYS[1], 'mode', 'write'); " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
//如果加过锁，那么就要看是不是写锁+写锁是不是自己加过的(即重入写锁)
"if (mode == 'write') then " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
//重入写锁
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"local currentExpire = redis.call('pttl', KEYS[1]); " +
"redis.call('pexpire', KEYS[1], currentExpire + ARGV[1]); " +
"return nil; " +
"end; " +
"end;" +
//执行命令"pttl myLock"，返回myLock的剩余过期时间
"return redis.call('pttl', KEYS[1]);",
Arrays.<Object>asList(getRawName()),//KEYS[1] = myLock
unit.toMillis(leaseTime),//ARGV[1] = 30000
getLockName(threadId)//ARGV[2] = UUID1:ThreadID1:write 或 ARGV[2] = UUID2:ThreadID2:write
);
}
...
}7.写锁RedissonWriteLock的可重入逻辑
(1)同一个客户端线程先加读锁再加读锁
(2)同一个客户端线程先加读锁再加写锁
(3)同一个客户端线程先加写锁再加读锁
(4)同一个客户端线程先加写锁再加写锁
前面分析了不同客户端线程的四种加锁情况：
情况一：先加读锁再加读锁，不互斥
情况二：先加读锁再加写锁，互斥
情况三：先加写锁再加读锁，互斥
情况四：先加写锁再加写锁，互斥
接下来分析同一个客户端线程的四种加锁情况：
情况一：先加读锁再加读锁，不互斥
情况二：先加读锁再加写锁，互斥
情况三：先加写锁再加读锁，不互斥
情况四：先加写锁再加写锁，不互斥
可以这样理解：写锁优先级高，读锁优先级低。同一个线程如果先加了优先级高的写锁，那就可以继续加优先级低的读锁。同一个线程如果先加了优先级低的读锁，那就不可以再加优先级高的写锁。一般锁可以降级，不可以升级。
(1)同一个客户端线程先加读锁再加读锁
客户端A(UUID1:ThreadID1)先加了一次读锁时：
//传入参数
KEYS[1] = myLock
KEYS[2] = {myLock}:UUID1:ThreadID1:rwlock_timeout
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1
ARGV[3] = UUID1:ThreadID1:write
//执行结果
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1客户端A(UUID1:ThreadID1)再加一次读锁时，判断通过可以加成功。
//执行命令
hget myLock mode，发现mode=read，表示已经加过读锁
hincrby myLock UUID1:ThreadID1 1
set {myLock}:UUID1:ThreadID1:rwlock_timeout:2 1
pexpire myLock 30000
pexpire {myLock}:UUID1:ThreadID1:rwlock_timeout:2 30000
//执行结果
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 2
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1
{myLock}:UUID1:ThreadID1:rwlock_timeout:2 ==> 1(2)同一个客户端线程先加读锁再加写锁
客户端A(UUID1:ThreadID1)先加了一次读锁时：
//传入参数
KEYS[1] = myLock
KEYS[2] = {myLock}:UUID1:ThreadID1:rwlock_timeout
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1
ARGV[3] = UUID1:ThreadID1:write
//执行结果
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1客户端A(UUID1:ThreadID1)再加一次写锁时，判断不通过，不可以加成功。
//传入参数
KEYS[1] = myLock
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1:write执行命令"hget myLock mode"，发现mode = read，不符合加写锁条件。所以同一个客户端线程，先加读锁再加写锁，是会互斥的。
(3)同一个客户端线程先加写锁再加读锁
客户端A(UUID1:ThreadID1)先加了一次写锁时：
//传入参数
KEYS[1] = myLock
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1:write
//执行结果
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1
}客户端A(UUID1:ThreadID1)再加一次读锁时，判断通过，可以加成功。
//传入参数
KEYS[1] = myLock
KEYS[2] = {myLock}:UUID1:ThreadID1:rwlock_timeout
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1
ARGV[3] = UUID1:ThreadID1:write
//执行命令
hget myLock mode，发现mode=write，表示已经加过写锁
hexists myLock UUID1:ThreadID1:write，判断写锁是自己加的，条件成立
hincrby myLock UUID1:ThreadID1 1，表示此时加了一个读锁
set {myLock}:UUID1:ThreadID1:rwlock_timeout:1 1
pexpire myLock 30000
pexpire {myLock}:UUID1:ThreadID11:rwlock_timeout:1 30000
//执行结果
//Hash结构
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1,
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1可见：如果是同一个客户端线程，先加写锁再加读锁，是可以加成功的。所以默认在线程持有写锁的期间，同样的线程可以多次加读锁。
(4)同一个客户端线程先加写锁再加写锁
客户端A(UUID1:ThreadID1)先加了一次写锁时：
//传入参数
KEYS[1] = myLock
ARGV[1] = 30000
ARGV[2] = UUID1:ThreadID1:write
//执行结果
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1
}客户端A(UUID1:ThreadID1)再加一次写锁时，判断通过，可以加成功。
//执行命令
hexists myLock UUID1:ThreadID1:write，判断是否是自己加的写锁
hincrby myLock UUID1:ThreadID1:write 1
pexpire myLock 50000
//执行结果
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 2
}可见：读写锁也是一种可重入锁。同一个客户端线程多次加写锁，是可以重入加锁的。先加的写锁是可以被读锁重入，先加的读锁则不可以被写锁重入。
8.读锁RedissonReadLock的释放读锁逻辑
(1)RedissonReadLock的释放读锁的流程
(2)释放读锁前主要三种情况
(3)RedissonReadLock的释放读锁的lua脚本
(4)对合并的情况一和情况二执行lua脚本
(5)对情况三执行lua脚本
(1)RedissonReadLock的释放读锁的流程
释放读锁调用的是RedissonLock的unlock()方法。
在RedissonLock的unlock()方法中，会执行get(unlockAsync())代码。也就是首先调用RedissonBaseLock的unlockAsync()方法，然后调用RedissonObject的get()方法。
其中unlockAsync()方法是异步化执行的方法，释放锁的操作就是异步执行的。而RedisObject的get()方法会通过RFuture同步等待获取异步执行的结果，可以将get(unlockAsync())理解为异步转同步。
在RedissonBaseLock的unlockAsync()方法中：可重入锁会调用RedissonLock.unlockInnerAsync()方法进行异步释放锁，读锁则会调用RedissonReadLock的unlockInnerAsync()方法进行异步释放锁，然后当完成释放锁的处理后，再通过异步去取消定时调度任务。
public class Application {
public static void main(String[] args) throws Exception {
Config config = new Config();
config.useClusterServers().addNodeAddress("redis://192.168.1.110:7001");
//读写锁
RedissonClient redisson = Redisson.create(config);
RReadWriteLock rwlock = redisson.getReadWriteLock("myLock");
rwlock.readLock().lock();//获取读锁
rwlock.readLock().unlock();//释放读锁
rwlock.writeLock().lock();//获取写锁
rwlock.writeLock().unlock();//释放写锁
...
}
}
public class RedissonLock extends RedissonBaseLock {
...
@Override
public void unlock() {
...
//异步转同步
//首先调用的是RedissonBaseLock的unlockAsync()方法
//然后调用的是RedissonObject的get()方法
get(unlockAsync(Thread.currentThread().getId()));
...
}
...
}
public abstract class RedissonBaseLock extends RedissonExpirable implements RLock {
...
@Override
public RFuture<Void> unlockAsync(long threadId) {
//异步执行释放锁的lua脚本
RFuture<Boolean> future = unlockInnerAsync(threadId);
CompletionStage<Void> f = future.handle((opStatus, e) -> {
//取消定时调度任务
cancelExpirationRenewal(threadId);
if (e != null) {
throw new CompletionException(e);
}
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + id + " thread-id: " + threadId);
throw new CompletionException(cause);
}
return null;
});
return new CompletableFutureWrapper<>(f);
}
protected abstract RFuture<Boolean> unlockInnerAsync(long threadId);
...
}
public class RedissonReadLock extends RedissonLock implements RLock {
...
@Override
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"...",
Arrays.<Object>asList(getRawName(), getChannelName(), timeoutPrefix, keyPrefix),
LockPubSub.UNLOCK_MESSAGE,
getLockName(threadId)
);
}
...
}(2)释放读锁前主要三种情况
情况一：不同客户端线程加了读锁
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 1,
"UUID2:ThreadID2": 1,
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1
{myLock}:UUID2:ThreadID2:rwlock_timeout:1 ==> 1情况二：同一个客户端线程多次重入加读锁
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 2
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1
{myLock}:UUID1:ThreadID1:rwlock_timeout:2 ==> 1情况一可以和情况二进行合并：
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 2,
"UUID2:ThreadID2": 1,
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1
{myLock}:UUID1:ThreadID1:rwlock_timeout:2 ==> 1
{myLock}:UUID2:ThreadID2:rwlock_timeout:1 ==> 1情况三：同一个客户端线程先加写锁再加读锁
//Hash结构
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1,
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1(3)RedissonReadLock的释放读锁的lua脚本
public class RedissonReadLock extends RedissonLock implements RLock {
...
@Override
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
//执行命令"hget myLock mode"
"local mode = redis.call('hget', KEYS[1], 'mode'); " +
//如果mode为false就发布一个消息
"if (mode == false) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end; " +
//执行命令"hexists myLock UUID1:ThreadIdD1"，判断当前线程对应的Hash值是否存在
"local lockExists = redis.call('hexists', KEYS[1], ARGV[2]); " +
"if (lockExists == 0) then " +
"return nil;" +
"end; " +
//执行命令"hincrby myLock UUID1:ThreadID1 -1"，递减当前线程对应的Hash值
"local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); " +
"if (counter == 0) then " +
"redis.call('hdel', KEYS[1], ARGV[2]); " +
"end;" +
//例如执行"del {myLock}:UUID1:ThreadId1:rwlock_timeout:2"
//删除当前客户端线程UUID1:ThreadId1的一个重入读锁；
"redis.call('del', KEYS[3] .. ':' .. (counter+1)); " +
//执行命令"hlen myLock > 1"，判断Hash里的元素是否超过1个
"if (redis.call('hlen', KEYS[1]) > 1) then " +
"local maxRemainTime = -3; " +
//获取key为锁名的Hash值的所有key
"local keys = redis.call('hkeys', KEYS[1]); " +
//遍历这些key，获取这些重入和非重入的读锁的最大剩余过期时间
"for n, key in ipairs(keys) do " +
"counter = tonumber(redis.call('hget', KEYS[1], key)); " +
//把key为mode的kv对排除
"if type(counter) == 'number' then " +
//通过递减拼接重入锁的key
"for i=counter, 1, -1 do " +
"local remainTime = redis.call('pttl', KEYS[4] .. ':' .. key .. ':rwlock_timeout:' .. i); " +
"maxRemainTime = math.max(remainTime, maxRemainTime);" +
"end; " +
"end; " +
"end; " +
//找出所有重入的和非重入的读锁的最大剩余过期时间后，就重置锁的过期时间为该时间
"if maxRemainTime > 0 then " +
"redis.call('pexpire', KEYS[1], maxRemainTime); " +
"return 0; " +
"end;" +
"if mode == 'write' then " +
"return 0;" +
"end; " +
"end; " +
//删除锁
"redis.call('del', KEYS[1]); " +
//发布一个事件
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; ",
//KEYS[1] = myLock，表示锁的名字
//KEYS[2] = redisson_rwlock:{myLock}，用于Redis的发布订阅用
//KEYS[3] = {myLock}:UUID1:ThreadID1:rwlock_timeout
//KEYS[4] = {myLock}
Arrays.<Object>asList(getRawName(), getChannelName(), timeoutPrefix, keyPrefix),
LockPubSub.UNLOCK_MESSAGE,//ARGV[1] = 0，表示发布事件类型
getLockName(threadId)//ARGV[2] = UUID1:ThreadID1，表示锁里面的该客户端线程代表的key
);
}
...
}参数说明：
KEYS[1] = myLock，表示锁的名字
KEYS[2] = redisson_rwlock:{myLock}，用于Redis的发布订阅用
KEYS[3] = {myLock}:UUID1:ThreadID1:rwlock_timeout
KEYS[4] = {myLock}
ARGV[1] = 0，表示发布事件类型
ARGV[2] = UUID1:ThreadID1，表示锁里面的该客户端线程代表的key(4)对合并的情况一和情况二执行lua脚本
一.客户端A(UUID1:ThreadID1)先释放一次读锁
二.客户端A(UUID1:ThreadID1)再释放一次读锁
三.客户端B(UUID2:ThreadID2)再释放一次读锁
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 2,
"UUID2:ThreadID2": 1,
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1
{myLock}:UUID1:ThreadID1:rwlock_timeout:2 ==> 1
{myLock}:UUID2:ThreadID2:rwlock_timeout:1 ==> 1一.客户端A(UUID1:ThreadID1)先释放一次读锁
首先执行命令"hget myLock mode"，发现mode = read。然后执行命令"hexists myLock UUID1:ThreadIdD1"，发现肯定是存在的，因为这个客户端线程UUID1:ThreadIdD1加过读锁。
接着执行命令"hincrby myLock UUID1:ThreadID1 -1"，将这个客户端线程对应的加读锁次数递减1，counter由2变成1。当counter大于1，说明还有线程持有着这个读锁。于是接着执行"del {myLock}:UUID1:ThreadId1:rwlock_timeout:2"，也就是删除用来记录当前客户端线程第2个重入锁过期时间的key。
此时myLock锁的数据变成如下：
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 1,
"UUID2:ThreadID2": 1,
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1
{myLock}:UUID2:ThreadID2:rwlock_timeout:1 ==> 1于是接着执行命令"hlen myLock"，判断Hash里的元素是否超过1个。如果超过1，那么就遍历已被线程获取的所有重入和非重入的读锁，即遍历所有类似"{myLock}:UUID2:ThreadID2:rwlock_timeout:1"的key。
然后接着执行命令"pttl {myLock}:UUID1:ThreadID1:rwlock_timeout:1"。即获取每一个重入读锁和非重入读锁的剩余过期时间，并找出其中最大的。执行"pexpire myLock"重置读锁的过期时间，为最大的剩余过期时间。
二.客户端A(UUID1:ThreadID1)再释放一次读锁
首先执行命令"hincrby myLock UUID1:ThreadID1 -1"，将这个客户端线程对应的加读锁次数递减1，counter由1变成0。当counter=0时，就执行命令"hdel myLock UUID1:ThreadID1"，即删除用来记录当前客户端线程重入锁次数的key。
然后接着执行命令"del {myLock}:UUID1:ThreadID1:rwlock_timeout:1"，即删除用来记录当前客户端线程第1个重入锁过期时间的key。最后获取每个重入读锁和非重入读锁的剩余过期时间，并找出其中最大的。执行"pexpire myLock"重置读锁的过期时间，为最大的剩余过期时间。
此时myLock锁的数据变成如下：
//Hash结构
myLock: {
"mode": "read",
"UUID2:ThreadID2": 1,
}
//String结构
{myLock}:UUID2:ThreadID2:rwlock_timeout:1 ==> 1三.客户端B(UUID2:ThreadID2)再释放一次读锁
首先执行命令"hincrby myLock UUID2:ThreadID2 -1"，将这个客户端线程对应的加读锁次数递减1，counter由1变成0。然后执行命令"hdel myLock UUID2:ThreadID2"，即删除用来记录当前客户端线程重入锁次数的key。接着执行命令"del {myLock}:UUID1:ThreadID1:rwlock_timeout:1"，即删除用来记录当前客户端线程第1个重入锁过期时间的key。
此时myLock锁的数据变成如下：
//Hash结构
myLock: {
"mode": "read"
}此时继续执行命令"hlen myLock"，发现为1，判断不通过，于是执行"del myLock"。也就是当没有线程再持有这个读锁时，就会彻底删除这个读锁，然后发布一个事件出去。
(5)对情况三执行lua脚本
这种情况是：同一个客户端线程先加写锁再加读锁。此时myLock锁的数据如下：
//Hash结构
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1,
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1首先执行命令"hincrby myLock UUID1:ThreadID1 -1"，将这个客户端线程对应的加读锁次数递减1，counter由1变成0。然后执行命令"hdel myLock UUID1:ThreadID1"，即删除用来记录当前客户端线程重入锁次数的key。接着执行"del {myLock}:UUID1:ThreadID1:rwlock_timeout:1"，即删除用来记录当前客户端线程第1个重入锁过期时间的key。
此时myLock锁的数据变成如下：
//Hash结构
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 1
}接着执行命令"hlen myLock > 1"，判断Hash里的元素是否超过1个。发现判断通过，但由于没有了读锁，所以最后会判断mode如果是write，就返回0。
9.写锁RedissonWriteLock的释放写锁逻辑
(1)释放写锁前主要有两种情况
(2)RedissonWriteLock的释放写锁的lua脚本
(3)执行释放写锁的lua脚本
(1)释放写锁前主要有两种情况
情况一：同一个客户端线程多次重入加写锁
情况二：同一个客户端线程先加写锁再加读锁
这两种情况的锁数据可以合并为如下：
//Hash结构
myLock: {
"mode": "write",
"UUID1:ThreadID1:write": 2,
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1接下来以这种锁数据为前提进行lua脚本分析。
(2)RedissonWriteLock的释放写锁的lua脚本
public class RedissonWriteLock extends RedissonLock implements RLock {
...
@Override
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
//首先执行命令"hget myLock mode"，发现mode=write
"local mode = redis.call('hget', KEYS[1], 'mode'); " +
"if (mode == false) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
"if (mode == 'write') then " +
//然后执行命令"hexists myLock UUID1:ThreadIdD1:write"，发现存在
"local lockExists = redis.call('hexists', KEYS[1], ARGV[3]); " +
"if (lockExists == 0) then " +
"return nil;" +
"else " +
//于是接着执行命令"hincrby myLock UUID1:ThreadID1:write -1"
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
//当counter大于0，说明还有线程持有写锁，那么就重置锁的过期时间
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
//当counter为0，就执行命令"hdel myLock UUID1:ThreadID1:write"
"redis.call('hdel', KEYS[1], ARGV[3]); " +
//判断key为锁名的Hash里元素是否超过1个
"if (redis.call('hlen', KEYS[1]) == 1) then " +
//如果只有1个，则说明没有线程持有锁了，此时可以删除掉锁对应的key
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"else " +
//如果有超过1个，则说明还有线程持有读锁，此时需要将写锁转读锁
"redis.call('hset', KEYS[1], 'mode', 'read'); " +
"end; " +
"return 1; "+
"end; " +
"end; " +
"end; " +
"return nil;",
//KEYS[1] = myLock，KEYS[2] = redisson_rwlock:{myLock}
Arrays.<Object>asList(getRawName(), getChannelName()),
LockPubSub.READ_UNLOCK_MESSAGE,//ARGV[1] = 0
internalLockLeaseTime,//ARGV[2] = 30000
getLockName(threadId)//ARGV[3] = UUID1:ThreadID1:write
);
}
...
}(3)执行释放写锁的lua脚本
一.参数说明
KEYS[1] = myLock
KEYS[2] = redisson_rwlock:{myLock}
ARGV[1] = 0
ARGV[2] = 30000
ARGV[3] = UUID1:ThreadID1:write二.lua脚本执行分析
首先执行命令"hget myLock mode"，发现mode = write。然后执行命令"hexists myLock UUID1:ThreadIdD1:write"，发现存在。于是接着执行命令"hincrby myLock UUID1:ThreadID1:write -1"，也就是将这个客户端线程对应的加写锁次数递减1，counter由2变成1。当counter大于0，说明还有线程持有写锁，那么就重置锁的过期时间。当counter为0，就执行命令"hdel myLock UUID1:ThreadID1:write"，即删除用来记录当前客户端线程重入写锁次数的key。
删除后，myLock的锁数据如下：
//Hash结构
myLock: {
"mode": "write",
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1接着执行命令"hlen myLock"，判断key为锁名的Hash里元素是否超过1个。如果只有1个，则说明没有线程持有锁了，此时可以删除掉锁对应的key。如果有超过1个，则说明还有线程持有读锁，此时需要将写锁转读锁。
因此，最后myLock的锁数据如下：
//Hash结构
myLock: {
"mode": "read",
"UUID1:ThreadID1": 1
}
//String结构
{myLock}:UUID1:ThreadID1:rwlock_timeout:1 ==> 1