序
本文主要研究一下redisson的分布式鎖
maven
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.8.1</version>
</dependency>
實(shí)例
@Test
public void testDistributedLock(){
Config config = new Config();
// config.setTransportMode(TransportMode.EPOLL);
config.useSingleServer()
.setAddress("redis://192.168.99.100:6379");
RedissonClient redisson = Redisson.create(config);
IntStream.rangeClosed(1,5)
.parallel()
.forEach(i -> {
executeLock(redisson);
});
executeLock(redisson);
}
public void executeLock(RedissonClient redisson){
RLock lock = redisson.getLock("myLock");
boolean locked = false;
try{
LOGGER.info("try lock");
locked = lock.tryLock();
// locked = lock.tryLock(1,2,TimeUnit.MINUTES);
LOGGER.info("get lock result:{}",locked);
if(locked){
TimeUnit.HOURS.sleep(1);
LOGGER.info("get lock and finish");
}
}catch (Exception e){
e.printStackTrace();
}finally {
LOGGER.info("enter unlock");
if(locked){
lock.unlock();
}
}
}
源碼解析
RedissonLock.tryLock
redisson-3.8.1-sources.jar!/org/redisson/RedissonLock.java
@Override
public boolean tryLock() {
return get(tryLockAsync());
}
@Override
public RFuture<Boolean> tryLockAsync() {
return tryLockAsync(Thread.currentThread().getId());
}
@Override
public RFuture<Boolean> tryLockAsync(long threadId) {
return tryAcquireOnceAsync(-1, null, threadId);
}
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(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, 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;
}
<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.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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
protected String getLockName(long threadId) {
return id + ":" + threadId;
}
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 = renewExpirationAsync(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(), new ExpirationEntry(threadId, task)) != null) {
task.cancel();
}
}
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return 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));
}
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
if (leaseTime != -1) {
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.addListener(new FutureListener<Long>() {
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
return;
}
Long ttlRemaining = future.getNow();
// lock acquired
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
- 這里leaseTime沒有設(shè)置的話，默認(rèn)是-1，使用的是commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()，默認(rèn)為30秒
- tryLockInnerAsync使用的是一段lua腳本，該腳本有3個(gè)參數(shù)，第一個(gè)參數(shù)為KEYS數(shù)組，后面幾個(gè)參數(shù)為ARGV數(shù)組的元素
- 這里key的值為調(diào)用方指定的這個(gè)redissonLock的名稱，兩個(gè)變量，第一個(gè)為leaseTime，第二個(gè)為鎖的名稱，使用redissonLock的id+線程id
- lua腳本第一個(gè)方法判斷redissonLock的hashmap是否存在，如果不存在則創(chuàng)建，該hashmap有一個(gè)entry的key為鎖名稱，valude為1，之后設(shè)置該hashmap失效時(shí)間為leaseTime
- lua腳本第二個(gè)方法是在redissonLock的hashmap存在的情況下，將該鎖名的value增1，同時(shí)設(shè)置失效時(shí)間為leaseTime
- 最后返回該redissonLock名稱的key的ttl
- 執(zhí)行成功之后判斷ttl是否還有值，有的話則調(diào)用scheduleExpirationRenewal，防止lock未執(zhí)行完就失效
- scheduleExpirationRenewal是注冊(cè)一個(gè)延時(shí)任務(wù)，在internalLockLeaseTime / 3的時(shí)候觸發(fā)，執(zhí)行的方法是renewExpirationAsync，將該鎖失效時(shí)間重置回internalLockLeaseTime
- scheduleExpirationRenewal里頭給scheduleExpirationRenewal任務(wù)增加listener，如果設(shè)置成功之后還會(huì)再次遞歸調(diào)用scheduleExpirationRenewal重新注冊(cè)延時(shí)任務(wù)
- tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId)方法是指定自動(dòng)解鎖時(shí)間時(shí)調(diào)用的方法，它與tryAcquireOnceAsync的區(qū)別在于，它對(duì)ttl的方回值采用long值來判斷，如果是null，才執(zhí)行延長(zhǎng)失效時(shí)間的定時(shí)任務(wù)，而tryAcquireOnceAsync方法采用的是BooleanNullReplayConvertor，只要返回不是null，則返回true
RedissonLock.unlock
redisson-3.8.1-sources.jar!/org/redisson/RedissonLock.java
@Override
public void unlock() {
try {
get(unlockAsync(Thread.currentThread().getId()));
} catch (RedisException e) {
if (e.getCause() instanceof IllegalMonitorStateException) {
throw (IllegalMonitorStateException)e.getCause();
} else {
throw e;
}
}
// Future<Void> future = unlockAsync();
// future.awaitUninterruptibly();
// if (future.isSuccess()) {
// return;
// }
// if (future.cause() instanceof IllegalMonitorStateException) {
// throw (IllegalMonitorStateException)future.cause();
// }
// throw commandExecutor.convertException(future);
}
@Override
public RFuture<Void> unlockAsync(final long threadId) {
final RPromise<Void> result = new RedissonPromise<Void>();
RFuture<Boolean> future = unlockInnerAsync(threadId);
future.addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<Boolean> future) throws Exception {
if (!future.isSuccess()) {
cancelExpirationRenewal(threadId);
result.tryFailure(future.cause());
return;
}
Boolean opStatus = future.getNow();
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + threadId);
result.tryFailure(cause);
return;
}
if (opStatus) {
cancelExpirationRenewal(null);
}
result.trySuccess(null);
}
});
return result;
}
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
}
String getChannelName() {
return prefixName("redisson_lock__channel", getName());
}
void cancelExpirationRenewal(Long threadId) {
ExpirationEntry task = expirationRenewalMap.get(getEntryName());
if (task != null && (threadId == null || task.getThreadId() == threadId)) {
expirationRenewalMap.remove(getEntryName());
task.getTimeout().cancel();
}
}
- unlockInnerAsync通過lua腳本來釋放鎖，該lua使用兩個(gè)key，一個(gè)是redissonLock名稱，一個(gè)是channelName
- 該lua使用的變量有三個(gè)，一個(gè)是pubSub的unlockMessage，默認(rèn)為0，一個(gè)是internalLockLeaseTime，默認(rèn)為commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()，一個(gè)是鎖名稱
- 如果該redissonLock不存在，則直接發(fā)布unlock消息返回1；如果該鎖不存在則返回nil；
- 如果該鎖存在則將其計(jì)數(shù)-1，如果counter大于0，則重置下失效時(shí)間，返回0；如果counter不大于0，則刪除該redissonLock鎖，發(fā)布unlockMessage，返回1；如果上面條件都沒有命中返回nil
- unlockAsync里頭對(duì)unlockInnerAsync注冊(cè)了FutureListener，主要是調(diào)用cancelExpirationRenewal，取消掉scheduleExpirationRenewal任務(wù)
LockPubSub
redisson-3.8.1-sources.jar!/org/redisson/pubsub/LockPubSub.java
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)) {
Runnable runnableToExecute = value.getListeners().poll();
if (runnableToExecute != null) {
runnableToExecute.run();
}
value.getLatch().release();
}
}
}
- 接收到unlockMessage的時(shí)候，會(huì)調(diào)用RedissonLockEntry的listener，然后觸發(fā)latch的release
- tryAcquireOnceAsync這個(gè)方法默認(rèn)沒有創(chuàng)建LockPubSub，而且沒有指定自動(dòng)解鎖時(shí)間，則定時(shí)任務(wù)會(huì)一直延長(zhǎng)失效時(shí)間，這個(gè)可能存在鎖一直沒釋放的風(fēng)險(xiǎn)
小結(jié)
加鎖有如下注意事項(xiàng)：
- 加鎖需要設(shè)置超時(shí)時(shí)間，防止出現(xiàn)死鎖
- 加鎖以及設(shè)置超時(shí)時(shí)間的時(shí)候，需要保證兩個(gè)操作的原子性，因而最好使用lua腳本或者使用支持NX以及EX的set方法
- 加鎖的時(shí)候需要把加鎖的調(diào)用方信息，比如線程id給記錄下來，這個(gè)在解鎖的時(shí)候需要使用
- 對(duì)于加鎖時(shí)長(zhǎng)不確定的任務(wù)，為防止任務(wù)未執(zhí)行完導(dǎo)致超時(shí)被釋放，需要對(duì)尚未運(yùn)行完的任務(wù)延長(zhǎng)失效時(shí)間
解鎖有如下注意事項(xiàng)：
- 解鎖一系列操作(
判斷key是否存在，存在的話刪除key等)需要保證原子性，因而最好使用lua腳本 - 解鎖需要判斷調(diào)用方是否與加鎖時(shí)記錄的是否一致，防止鎖被誤刪
- 如果有延續(xù)失效時(shí)間的延時(shí)任務(wù)，在解鎖的時(shí)候，需要終止掉該任務(wù)
