时间会让我们更好。   

Redisson分布式锁原理

RedissonLock继承结构:

RLock接口中定义的方法:主要分析tryLock()实现。


(一) RedissonLock#tryLock:加锁逻辑

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long time = unit.toMillis(waitTime);// 等待时间
    long current = System.currentTimeMillis();
    final long threadId = Thread.currentThread().getId();// 使用当前线程ID,实现重入锁
    Long ttl = tryAcquire(leaseTime, unit, threadId);// 尝试获取锁
    // lock acquired
    if (ttl == null) {// 没有过期时间,未上锁,直接返回获取成功
        return true;
    }
    time -= (System.currentTimeMillis() - current);
    if (time <= 0) {// 等待时间已超时
        acquireFailed(threadId);
        return false;
    }
    current = System.currentTimeMillis();
    // 订阅锁的队列,等待锁被其余线程释放后通知
    // 通过Redis的Channel订阅监听队列,subscribe内部通过信号量semaphore,再通过await方法阻塞,内部其实是用CountDownLatch来实现阻塞,获取subscribe异步执行的结果,来保证订阅成功
    final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {// 阻塞等待subscribe的future的结果对象(即分布式解锁消息的结果)
        if (!subscribeFuture.cancel(false)) {
            subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
                @Override
                public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                    if (subscribeFuture.isSuccess()) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                }
            });
        }
        acquireFailed(threadId);
        return false;
    }

    try {
        time -= (System.currentTimeMillis() - current);
        if (time <= 0) {// subscribe方法调用超时,取消订阅,不再继续申请锁
            acquireFailed(threadId);
            return false;
        }

        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(leaseTime, unit, threadId);// 再次尝试申请锁
            if (ttl == null) {// 获得锁,返回
                return true;
            }

            time -= (System.currentTimeMillis() - currentTime);
            if (time <= 0) {// 超时
                acquireFailed(threadId);
                return false;
            }

            // waiting for message 等待订阅的队列消息
            currentTime = System.currentTimeMillis();
            // 通过信号量(共享锁)阻塞,等待解锁消息(这一点设计的非常精妙:减少了其他分布式节点的等待或者空转等无效锁申请的操作,整体提高了性能)
            if (ttl >= 0 && ttl < time) {// 在ttl内,从Entry的信号量获取一个许可(除非被中断或者一直没有可用的许可)
                getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {// 在等待剩余时间内...
                getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }

            time -= (System.currentTimeMillis() - currentTime);
            if (time <= 0) {// 超时
                acquireFailed(threadId);
                return false;
            }
        }
    } finally {
        unsubscribe(subscribeFuture, threadId);// 无论是否获得锁,都要取消订阅解锁消息
    }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
}
## RedissonLock#getEntry -> RedissonLock#getEntryName
protected String getEntryName() {
    return id + ":" + getName();// 客户端实例ID+锁名称来保证多个实例下的锁可重入
}

0x0001:RedissonLock#tryAcquire:尝试获取锁

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(leaseTime, unit, threadId));
}
## RedissonObject#get
protected final CommandAsyncExecutor commandExecutor;
protected final <V> V get(RFuture<V> future) {
    return commandExecutor.get(future);
}

0x0010:RedissonLock#tryAcquireAsync:异步尝试获取锁,异步调用,get方法阻塞

## RedissonLock#tryAcquireAsync
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) {// 起租时间不为空
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // lockWatchdogTimeout = 30 * 1000默认30秒
    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();
            if (ttlRemaining == null) {// 拿到锁
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}
## RedissonLock#scheduleExpirationRenewal Redisson避免死锁方案,避免锁未被释放。
## 若未设置过期时间的话,redission默认的过期时间是30s,同时未避免锁在业务未处理完成之前被提前释放,Redisson在获取到锁且默认过期时间的时候,会在当前客户端内部启动一个定时任务,每隔internalLockLeaseTime/3的时间去刷新key的过期时间,这样既避免了锁提前释放,同时如果客户端宕机的话,这个锁最多存活30s的时间就会自动释放
if (expirationRenewalMap.containsKey(getEntryName())) {
    return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
    @Override
    public void run(Timeout timeout) throws Exception {
        RFuture<Boolean> future = 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));

        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(), task) != null) {
    task.cancel();
}

0x0011:CommandAsyncService#get:包装了CountDownLatch是为了支持线程可中断操作

public <V> V get(RFuture<V> future) {
    if (!future.isDone()) {// Task还没执行完成
        final CountDownLatch l = new CountDownLatch(1);// 设置一个单线程的同步控制器
        future.addListener(new FutureListener<V>() {
            @Override
            public void operationComplete(Future<V> future) throws Exception {
                l.countDown();// 操作完成
            }
        });

        boolean interrupted = false;
        while (!future.isDone()) {
            try {
                l.await();
            } catch (InterruptedException e) {
                interrupted = true;
                break;
            }
        }

        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    }

    // commented out due to blocking issues up to 200 ms per minute for each thread
    // future.awaitUninterruptibly();
    if (future.isSuccess()) {
        return future.getNow();
    }
    throw convertException(future);
}

0x0100:RedissonLock#tryLockInnerAsync:真正执行的是一段具有原子性的Lua脚本,并且最终也是由CommandAsynExecutor去执行。使用Lua的好处:Redis会将整个脚本作为一个整体执行,中间不会被其他命令插入。因此在编写脚本的过程中无需担心会出现竞态条件,无需使用事务。

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);// 锁过期时间-毫秒
    // getName() - 逻辑锁名称:比如方法名+名称
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,"Lua脚本",Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
## RedissonLock#getLockName 锁对应的线程级别的名称,支持相同线程可重入,不同线程不可重入
protected String getLockName(long threadId) {
    return id + ":" + threadId;
}

    锁的实际存储类型是hash。KEY[1]->逻辑锁名称 ARGV[2]->线程级别的锁名称

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]);

    分析这段Lua脚本:

  • 检查锁名称是否存在,如果不存在,获取成功,同时设置线程级别锁名称,设置过期时间为internalLockLeaseTime
  • 如果检查存在KEYS[1], ARGV[2],获取成功,自增1,记录重入的次数,更新过期时间
  • 如果key不存在,直接返回key的剩余过期时间

0x0101:RedissonLock#acquireFailed:把该线程从获取锁操作的等待队列中直接删掉

private void acquireFailed(long threadId) {
    get(acquireFailedAsync(threadId));
}
## RedissonFairLock#acquireFailedAsync
@Override
protected RFuture<Void> acquireFailedAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_VOID,
                "local firstThreadId = redis.call('lindex', KEYS[1], 0); " + 
                "if firstThreadId == ARGV[1] then " + 
                    "local keys = redis.call('zrange', KEYS[2], 0, -1); " + 
                    "for i = 1, #keys, 1 do " + 
                        "redis.call('zincrby', KEYS[2], -tonumber(ARGV[2]), keys[i]);" + 
                    "end;" + 
                "end;" +
                "redis.call('zrem', KEYS[2], ARGV[1]); " +
                "redis.call('lrem', KEYS[1], 0, ARGV[1]); ",
                Arrays.<Object>asList(threadsQueueName, timeoutSetName), 
                getLockName(threadId), threadWaitTime);
}

0x0110:PublishSubscribe#subscribe:订阅消息

final AtomicReference<Runnable> listenerHolder = new AtomicReference<Runnable>();
// 根据channelName拿到信号量,channelName=UUID+":"+name,对应一个锁
final AsyncSemaphore semaphore = subscribeService.getSemaphore(new ChannelName(channelName));
final RPromise<E> newPromise = new RedissonPromise<E>() {
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return semaphore.remove(listenerHolder.get());
    }
};
Runnable listener = new Runnable() {
    @Override
    public void run() {
        E entry = entries.get(entryName);
        if (entry != null) {
            entry.aquire();
            semaphore.release();
            entry.getPromise().addListener(new TransferListener<E>(newPromise));
            return;
        }

        E value = createEntry(newPromise);
        value.aquire();

        E oldValue = entries.putIfAbsent(entryName, value);
        if (oldValue != null) {
            oldValue.aquire();
            semaphore.release();
            oldValue.getPromise().addListener(new TransferListener<E>(newPromise));
            return;
        }

        RedisPubSubListener<Object> listener = createListener(channelName, value);
        subscribeService.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);
    }
};
// 把生成的监听线程listenser加入到信号量的监听集合中去,后面发布解锁消息的时候,会唤醒
semaphore.acquire(listener);
listenerHolder.set(listener);
return newPromise;

(二) RedissonLock#unlock:解锁逻辑

@Override
public void unlock() {
    try {
        get(unlockAsync(Thread.currentThread().getId()));
    } catch (RedisException e) {
        ...
    }
}

0x0001:RedissonLock#unlockAsync:异步解锁

@Override
public RFuture<Void> unlockAsync(final long threadId) {
    final RPromise<Void> result = new RedissonPromise<Void>();
    RFuture<Boolean> future = unlockInnerAsync(threadId);// 此处Redis实现

    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) {// 解锁成功之后取消更新锁expire的时间任务,针对于没有锁过期时间的
                cancelExpirationRenewal(null);
            }
            result.trySuccess(null);
        }
    });
    return result;
}

0x0010:RedissonLock#unlockInnerAsync:当其他线程释放锁的时候,会同时根据锁的唯一通道publish一条分布式的解锁信息,接收到分布式消息后, 等待获取锁的Semaphore中的监听队列中的listenser线程可重新申请锁。

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"Lua脚本",Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
}

    拿出来Lua看看如何实现的:KEY[1]->逻辑锁名称 KEY[2]->getChannelName() ARGV[1]->0L ARGV[2]->过期时间 ARGV[3]->线程级别的锁名称

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;

    分析这段Lua脚本:

  • 如果键不存在,说明锁可以用了,发布锁释放消息后返回
  • 如果锁不是被当前线程锁定,则返回nil
  • Redisson支持重入,在解锁的时候,引用数减1
  • 如果重入数>0,重新设置过期时间
  • 如果重入数>=0,说明锁可以使用了,发布锁释放消息后返回

(三) LockPubSub#unlockMessage:处理解锁消息

Redisson在LockPubSub中处理解锁消息,首先看一下LockPubSub继承结构:

PublishSubscribe#createListener

// 模板方法,提供给子类实现
protected abstract void onMessage(E value, Long message);

private RedisPubSubListener<Object> createListener(final String channelName, final E value) {
    RedisPubSubListener<Object> listener = new BaseRedisPubSubListener() {
        @Override
        public void onMessage(CharSequence channel, Object message) {
            if (!channelName.equals(channel.toString())) {
                return;
            }
            PublishSubscribe.this.onMessage(value, (Long)message);// 此处会调用到LockPubSub.onMessage
        }
        ...
    };
    return listener;
}

LockPubSub:解锁消息

public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
    public static final Long unlockMessage = 0L;// 解锁消息,redis执行lua返回值
    public static final Long readUnlockMessage = 1L;
    @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();
            }
            // 释放一个,唤醒等待的entry.getLatch().tryAcquire去再次尝试获取锁
            value.getLatch().release();
        } else if (message.equals(readUnlockMessage)) {
            while (true) {// 如果还有其他Listeners回调,也唤醒执行
                Runnable runnableToExecute = value.getListeners().poll();
                if (runnableToExecute == null) {
                    break;
                }
                runnableToExecute.run();
            }
            value.getLatch().release(value.getLatch().getQueueLength());
        }
    }
}
发表新评论
选择表情