Redisson分布式锁的源码解读

原创

Java4ye

发布于 2024-07-02 06:16:55

1210

发布于 2024-07-02 06:16:55

之前秒杀项目中就用到了这个 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 acquired
        if (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> {
    @Override
    public 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));
}

ARGV1 过期时间

ARGV2 即 getLockName(threadId) ，这里是 redisson 客户端id + 这个线程 ID ，如下 👇

scheduleExpirationRenewal （看门狗机制）

上面加锁完，就来到这段代码。

没有设置过期时间的话，默认给你设置 30 s 过期，并每隔 10s 自动续期，确保锁不会在使用过程中过期。

同时，防止客户端宕机，留下死锁。

// RedissonBaseLock.java

protected 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() {
        @Override
        public 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 itself
                    renewExpiration();
                } 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));
}

KEYS1 为锁名，KEYS2 channel 名 👇