之前秒杀项目中就用到了这个 Redisson 分布式锁 👇,这篇就一起来看看源码吧!
tryLock 加锁 流程
// 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, -1, null, threadId);
}private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {RFuture<Boolean> acquiredFuture;// 续租时间:锁的过期时间(没有设置的话就用默认的 internalLockLeaseTime 看门狗时间)if (leaseTime > 0) {acquiredFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);} else {acquiredFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);}CompletionStage<Boolean> f = acquiredFuture.thenApply(acquired -> {// lock acquiredif (acquired) {if (leaseTime > 0) {internalLockLeaseTime = unit.toMillis(leaseTime);} else {// 没配置过期时间就执行这里scheduleExpirationRenewal(threadId);}}return acquired;});return new CompletableFutureWrapper<>(f);
}
代码很长,主要看
tryLockInnerAsync
和scheduleExpirationRenewal
方法。
前置知识
// EVAL 命令,用于在 Redis 服务器端执行 Lua 脚本。
RedisStrictCommand<Boolean> EVAL_NULL_BOOLEAN = new RedisStrictCommand<Boolean>("EVAL", new BooleanNullReplayConvertor());// BooleanNullReplayConvertor 判断是不是 NULL。
public class BooleanNullReplayConvertor implements Convertor<Boolean> {@Overridepublic Boolean convert(Object obj) { return obj == null; }
}
tryLockInnerAsync
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {// getRawName 即 锁的名称return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,// 锁不存在,添加 hash 数据,可重入次数加一,毫秒级别过期时间,返回 null"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +// 锁存在,可重入次数加一,毫秒级别过期时间,返回 null"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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}
ARGV[1] 过期时间
ARGV[2] 即 getLockName(threadId) ,这里是 redisson 客户端id + 这个线程 ID , 如下 👇
scheduleExpirationRenewal (看门狗机制)
上面加锁完,就来到这段代码。
没有设置过期时间的话,默认给你设置 30 s 过期,并每隔 10s 自动续期,确保锁不会在使用过程中过期。
同时,防止客户端宕机,留下死锁。
// RedissonBaseLock.javaprotected void scheduleExpirationRenewal(long threadId) {ExpirationEntry entry = new ExpirationEntry();ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);if (oldEntry != null) {oldEntry.addThreadId(threadId);} else {entry.addThreadId(threadId);try {// 看这里 renewExpiration();} finally {if (Thread.currentThread().isInterrupted()) {cancelExpirationRenewal(threadId);}}}
}private void renewExpiration() {ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee == null) {return;}// 延时任务,10s 续期一次。Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent == null) {return;}Long threadId = ent.getFirstThreadId();if (threadId == null) {return;}// 续期操作CompletionStage<Boolean> future = renewExpirationAsync(threadId);future.whenComplete((res, e) -> {if (e != null) {log.error("Can't update lock " + getRawName() + " expiration", e);EXPIRATION_RENEWAL_MAP.remove(getEntryName());return;}if (res) {// reschedule itselfrenewExpiration();} else {cancelExpirationRenewal(null);}});}// 三分之一时间,30s /3= 10s}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}// 续期脚本
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {return evalWriteAsync(getRawName(), 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.singletonList(getRawName()),internalLockLeaseTime, getLockName(threadId));
}
get
上面的加锁操作,最终返回的是 return new CompletableFutureWrapper<>(f);
这个异步操作。
还记得上面的 BooleanNullReplayConvertor 吗,当 eval 执行加锁脚本时,成功会返回 null,并在这里转成 True 。
@Override
public <V> V get(RFuture<V> future) {if (Thread.currentThread().getName().startsWith("redisson-netty")) {throw new IllegalStateException("Sync methods can't be invoked from async/rx/reactive listeners");}try {return future.toCompletableFuture().get();} catch (InterruptedException e) {future.cancel(true);Thread.currentThread().interrupt();throw new RedisException(e);} catch (ExecutionException e) {throw convertException(e);}
}
那么,加锁的部分到这里就结束, 解锁 的就简单过一下 👇
unlock 解锁
// RedissonLock.java
protected RFuture<Boolean> unlockInnerAsync(long threadId) {return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,// 不存在,直接返回 null"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +"return nil;" +"end; " +// 减一"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +// 大于0,设置毫秒级过期时间,并返回0"if (counter > 0) then " +"redis.call('pexpire', KEYS[1], ARGV[2]); " +"return 0; " +// 删除锁,并向指定channel发布 0 这个消息,并返回1"else " +"redis.call('del', KEYS[1]); " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end; " +// 返回 null"return nil;",Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
KEYS[1] 为锁名,KEYS[2] channel 名 👇
ARGV[1] 为0 👇, ARGV[2] 过期时间,ARGV[3] 为 redisson 客户端id + 这个线程 ID
解锁后,取消续期任务。
结尾
通过源码,我们了解到上文提到的 redisson 框架的几个特点:自动续期,可重入锁, lua脚本。