Redisson分布式锁的源码解读

之前秒杀项目中就用到了这个 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 名 👇