Redisson 源码解析，如何利用Redis实现分布式可重入锁

正好最近研究了下Redisson的源码，和大家分享一下

前言

首先我们先回顾一下 Java 中的 ReentrantLock 是如何实现的？

这里我先简单介绍一下ReentrantLock 实现的思路

  • 锁标识：通过AQS的state变量作为锁标识，利用Java的CAS保证多线程竞争锁时的线程安全问题

  • 队列：未竞争到锁的线程进入AQS的队列并挂起，等待解锁时被唤醒（或者超时）

如何设计分布式可重入锁

首先锁标识，这个在Redis中很容易实现，可以用lock name 作为key，当前线程生成一个uuid，作为value，加上Redis 单线程模型，实现线程安全的锁竞争

这种方式在之前的博客里也提到过，可以参考下 Redis分布式锁的正确实现方式

但是如何基于Redis 做一个队列，像Java那样可以挂起唤醒线程呢？这点我在看源码之前一直没有想到...

那么Redisson 是如何做的呢？

答案：利用Redis的发布订阅，加上Java的Semaphore（信号量，不了解Semaphore的小伙伴可以Google一下）

Redisson 分布式锁实现思路

锁标识：Hash 数据结构，key 为锁的名字，filed 当前竞争锁成功线程的"唯一标识"，value 重入次数

队列：所有竞争锁失败的线程，会订阅当前锁的解锁事件，利用 Semaphore 实现线程的挂起和唤醒

源码分析

我们来看一下tryLock方法的源码

    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        // 尝试获取锁，返回null 代表获取锁成功，当获取锁失败时返回当前锁的释放时间
        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();
        // 订阅解锁事件
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // 等待订阅成功，成功后唤醒当前线程
        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

        try {
            // 再次判断一下是否超时
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
        
            while (true) {
                long currentTime = System.currentTimeMillis();
                // 尝试获取锁
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                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) {
                    // 等待解锁消息，此处利用Semaphore，锁未释放时，permits=0，线程处于挂起状态
                    // 当发布解锁消息时，当前的Semaphore对象的release() permits=1
                    // 所有的客户端都会有一个线程被唤醒，去尝试竞争锁
                    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));
    }

tryAcquire(leaseTime, unit, threadId); 这个方法我们下面会分析，现在我们只需要知道这个方法是用来获取锁就可以了

这个时候我们已经可以理清Redisson可重入锁的思路了

  1. 获取锁
  2. 如果获取锁失败，订阅解锁事件
  3. 之后是一个无限循环
while(true) {
  // 尝试获取锁

  // 判断是否超时

  // 等待解锁消息释放信号量 
  //（此时每个Java客户端都可能会有多个线程被挂起，但是只有一个线程会被唤醒）

  // 判断是否超时
}

利用信号量，合理控制线程对锁的竞争，合理利用系统资源，可以说做的灰常的奈斯了

需要注意：
!await(subscribeFuture, time, TimeUnit.MILLISECONDS) ，这里很多博客都解释错了，这里并不是等待发布解锁消息，只要订阅事件成功后，就会往下执行，真正等待解锁消息的是 getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);

这里你可能不信，为什么我说的就对啊，debug一下你就知道

tryLockInnerAsync

tryAcquire 内部依靠 tryLockInnerAsync 来实现获取锁的逻辑，我们来看下源码

    <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]); " +
                      // 竞争锁成功 返回null
                      "return nil; " +
                  "end; " +
                   // 如果锁已经被当前线程获取
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                       // 重入次数加1
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  // 锁被其他线程获取，返回锁的过期时间
                  "return redis.call('pttl', KEYS[1]);",

                    // 下面三个参数分别为 KEYS[1], ARGV[1], ARGV[2]
                    // 即锁的name，锁释放时间，当前线程唯一标识
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

tryLockInnerAsync 中利用lua脚本 和 Redis 单线程的特点来实现锁的竞争

这里可以看到锁的结构，和我们上文所说的一样，Hash 数据结构，key 为锁的name，filed 当前竞争锁成功线程的"唯一标识"，value 重入次数

unlockInnerAsync

接下来我们再来看解锁的核心代码

    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // 用锁的name和线程唯一标识去判断是否存在这样的键值对
                // 解铃还须系铃人，不存在则无权解锁，返回null
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                    "return nil;" +
                "end; " +
                // 解锁逻辑
                // 冲入次数-1
                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                // 如果大于0 代表当前线程重入锁多次无法解锁，更新锁的有效时间
                "if (counter > 0) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                "else " +
                    // 解锁，删除key
                    "redis.call('del', KEYS[1]); " +
                    // 发布解锁消息
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; "+
                "end; " +
                "return nil;",
                // KEYS[1]，KEYS[2]
                // 锁的name，发布订阅的Channel
                Arrays.<Object>asList(getName(), getChannelName()), 
                // ARGV[1] ~ ARGV[3]
                // 解锁消息，释放时间，当前线程唯一标识
                LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));

    }

发布解锁消息后，会调用到LockPubSub 的 onMessage，释放信号量，唤醒等待锁的线程

public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {

    public static final Long UNLOCK_MESSAGE = 0L;
    public static final Long READ_UNLOCK_MESSAGE = 1L;

    public LockPubSub(PublishSubscribeService service) {
        super(service);
    }
    
    @Override
    protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) {
        return new RedissonLockEntry(newPromise);
    }

    @Override
    protected void onMessage(RedissonLockEntry value, Long message) {
        if (message.equals(UNLOCK_MESSAGE)) {
            Runnable runnableToExecute = value.getListeners().poll();
            if (runnableToExecute != null) {
                runnableToExecute.run();
            }

            // 释放信号量
            value.getLatch().release();
        } else if (message.equals(READ_UNLOCK_MESSAGE)) {
            while (true) {
                Runnable runnableToExecute = value.getListeners().poll();
                if (runnableToExecute == null) {
                    break;
                }
                runnableToExecute.run();
            }

            value.getLatch().release(value.getLatch().getQueueLength());
        }
    }

}

参考

欢迎点赞、转发。你的支持就是对我最大的帮助

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 199,711评论 5赞 468
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 83,932评论 2赞 376
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 146,770评论 0赞 330
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 53,799评论 1赞 271
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 62,697评论 5赞 359
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,069评论 1赞 276
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,535评论 3赞 390
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,200评论 0赞 254
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,353评论 1赞 294
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,290评论 2赞 317
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,331评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,020评论 3赞 315
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,610评论 3赞 303
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,694评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,927评论 1赞 255
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,330评论 2赞 346
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 41,904评论 2赞 341