本站消息

站长简介/公众号


站长简介:逗比程序员,理工宅男,前每日优鲜python全栈开发工程师,利用周末时间开发出本站,欢迎关注我的微信公众号:程序员总部,程序员的家,探索程序员的人生之路!分享IT最新技术,关注行业最新动向,让你永不落伍。了解同行们的工资,生活工作中的酸甜苦辣,谋求程序员的最终出路!

  价值13000svip视频教程,java大神匠心打造,零基础java开发工程师视频教程全套,基础+进阶+项目实战,包含课件和源码

  出租广告位,需要合作请联系站长


+关注
已关注

分类  

暂无分类

标签  

暂无标签

日期归档  

2020-11(403)

2020-12(497)

2021-01(423)

2021-02(280)

2021-03(351)

Redisson分布式锁以及其底层原理

发布于2020-11-19 20:54     阅读(68)     评论(0)     点赞(27)     收藏(4)


介绍与配置

Redisson官方文档:https://github.com/redisson/redisson/wiki/Redisson%E9%A1%B9%E7%9B%AE%E4%BB%8B%E7%BB%8D

Springboot 自动配置类: RedissonAutoConfiguration 

pom配置:

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
   <groupId>org.apache.commons</groupId>
   <artifactId>commons-pool2</artifactId>
</dependency>
<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson-spring-boot-starter</artifactId>
   <version>3.13.6</version>
</dependency>

如果什么都不配置的话,会默认使用单Redis节点模式,代码中直接就可以使用  RedissonClient 

 

具体配置可参考官方文档:https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95

分布式锁测试

@Slf4j
@SpringBootTest(classes = DemoWebApplication.class)
public class RedissonTest {

  @Resource
    private RedissonClient redissonClient;

  @Test
    public void redissonTest() throws InterruptedException {
        log.info("===redissonTest====start===============");
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                lockTest();
            }).start();
        }
        Thread.sleep(30000);
        log.info("===redissonTest====end===============");
    }
  private void tryLockTest() {
    // 见下文
  }
  private void lockTest() {
    // 见下文
  }
}
 
  private void tryLockTest() {
        String threadName = Thread.currentThread().getName();
        log.info("===Thread=={}===start===", threadName);
        RLock lock = redisson.getLock("DistributedRedisLockTest");

        // 尝试加锁,最多等待10秒,上锁以后30秒自动解锁
        boolean lockFlag = false;
        try {
            // 尝试去加锁,10秒没获取到锁,则返回false
            // res = lock.tryLock(10, TimeUnit.SECONDS);
            // 尝试去加锁,10秒没获取到锁,则返回false,获取到则返回true,获取到锁后30秒自动释放
            // 当waitTime设置为0时,就相当于setNx,获取不到锁直接退出
            lockFlag = lock.tryLock(5, 1, TimeUnit.SECONDS);
            if (!lockFlag) {
                log.info("===Thread=={}==res={}==没有获取到锁,退出===", threadName, lockFlag);
                return;
            }

            log.info("===Thread=={}============getLock===", threadName);
            // 模拟业务逻辑
            Thread.sleep(2000);
        } catch (Exception e) {
            log.error("执行异常,e:{}", ExceptionUtils.getStackTrace(e));
        } finally {
            log.info("===Thread=={}==========isHeldByCurrentThread={}", threadName, lock.isHeldByCurrentThread());
            // 释放锁也可能出现异常,比如业务代码没执行完,锁就过期,此时进行释放会抛异常,加个当前线程是否持有所的判断
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        log.info("===Thread=={}==lockFlag={}=end===", threadName, lockFlag);
    }

 

    private void lockTest() {
        String threadName = Thread.currentThread().getName();
        log.info("===Thread=={}===start===", threadName);
        RLock lock = redisson.getLock("DistributedRedisLockTest");

        // lock表示去加锁,加锁成功,没有返回值,继续执行下面代码;加锁失败,它会一直阻塞,直到锁被释放,再继续往下执行
        // lock.lock();
        // 1秒自动释放时间,但是后续执行unlock操作时会报错(自己只能解锁自己的,第一个线程释放之后执行到unlock方法,但是此时锁已经是第二个线程的了)
        lock.lock(1, TimeUnit.SECONDS);

        log.info("===Thread=={}============getLock===", threadName);
        try {
            Thread.sleep(2000);
        } catch (Exception e) {
            log.error("执行异常,e:{}", ExceptionUtils.getStackTrace(e));
        } finally {
            log.info("===Thread=={}==========isHeldByCurrentThread={}", threadName, lock.isHeldByCurrentThread());
            // 释放锁也可能出现异常,比如业务代码没执行完,锁就过期,此时进行释放会抛异常,加个当前线程是否持有所的判断
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        log.info("===Thread=={}===end===", threadName);
    }

原理

本文中Redisson版本为 redisson-spring-boot-starter 3.13.6

先看下接口方法:

public interface RRLock extends Lock, RLockAsync{
    //----------------------Lock接口方法-----------------------

    /**
     * 加锁 锁的有效期默认30秒
     */
    void lock();
    /**
     * tryLock()方法是有返回值的,它表示用来尝试获取锁,如果获取成功,则返回true,如果获取失败(即锁已被其他线程获取),则返回false .
     */
    boolean tryLock();
    /**
     * tryLock(long time, TimeUnit unit)方法和tryLock()方法是类似的,只不过区别在于这个方法在拿不到锁时会等待一定的时间,
     * 在时间期限之内如果还拿不到锁,就返回false。如果如果一开始拿到锁或者在等待期间内拿到了锁,则返回true。
     *
     * @param time 等待时间
     * @param unit 时间单位 小时、分、秒、毫秒等
     */
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    /**
     * 解锁
     */
    void unlock();
    /**
     * 中断锁 表示该锁可以被中断 假如A和B同时调这个方法,A获取锁,B为获取锁,那么B线程可以通过
     * Thread.currentThread().interrupt(); 方法真正中断该线程
     */
    void lockInterruptibly();

    //----------------------RLock接口方法-----------------------
    /**
     * 加锁 上面是默认30秒这里可以手动设置锁的有效时间
     *
     * @param leaseTime 锁有效时间
     * @param unit      时间单位 小时、分、秒、毫秒等
     */
    void lock(long leaseTime, TimeUnit unit);
    /**
     * 这里比上面多一个参数,多添加一个锁的有效时间
     *
     * @param waitTime  等待时间
     * @param leaseTime 锁有效时间
     * @param unit      时间单位 小时、分、秒、毫秒等
     */
    boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    /**
     * 检验该锁是否被线程使用,如果被使用返回True
     */
    boolean isLocked();
    /**
     * 检查当前线程是否获得此锁(这个和上面的区别就是该方法可以判断是否当前线程获得此锁,而不是此锁是否被线程占有)
     * 这个比上面那个实用
     */
    boolean isHeldByCurrentThread();
    /**
     * 中断锁 和上面中断锁差不多,只是这里如果获得锁成功,添加锁的有效时间
     * @param leaseTime  锁有效时间
     * @param unit       时间单位 小时、分、秒、毫秒等
     */
    void lockInterruptibly(long leaseTime, TimeUnit unit);  
}
  • tryLock方法
@Override
    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();
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // 获取到锁,则直接返回true
        if (ttl == null) {
            return true;
        }
        // 没获取到锁,校验是否超过等待时长,超过则返回false
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
     // 订阅监听redis消息,并且创建RedissonLockEntry,其中RedissonLockEntry中比较关键的是一个 Semaphore属性对象,用来控制本地的锁请求的信号量同步,返回的是netty框架的Future实现。 RFuture
<RedissonLockEntry> subscribeFuture = subscribe(threadId); // 阻塞等待subscribe的future的结果对象,如果subscribe方法调用超过了time,说明已经超过了客户端设置的最大wait time,则直接返回false,取消订阅,不再继续申请锁了。
     // 只有await返回true,才进入循环尝试获取锁

     if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) { if (!subscribeFuture.cancel(false)) { subscribeFuture.onComplete((res, e) -> { if (e == null) { unsubscribe(subscribeFuture, threadId); } }); } acquireFailed(waitTime, unit, threadId); return false; } try { time -= System.currentTimeMillis() - current; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; }   // 如果没有超过尝试获取锁的等待时间,那么通过While一直获取锁。最终只会有两种结果
       // 1) 在等待时间内获取锁成功 返回true 2)等待时间结束了还没有获取到锁那么返回false。
while (true) { long currentTime = System.currentTimeMillis(); ttl = tryAcquire(waitTime, leaseTime, unit, threadId); // 获取锁成功 if (ttl == null) { return true; }           // 获取锁失败 time -= System.currentTimeMillis() - currentTime; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; } // waiting for message currentTime = System.currentTimeMillis(); if (ttl >= 0 && ttl < time) { subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); }            time -= System.currentTimeMillis() - currentTime; if (time <= 0) { acquireFailed(waitTime, unit, threadId); return false; } } } finally { unsubscribe(subscribeFuture, threadId); } // return get(tryLockAsync(waitTime, leaseTime, unit)); }

 

    private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
     // 如果指定了失效时间,就按指定的失效时间执行,然后返回
if (leaseTime != -1) { return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG); }
     // 如果没有指定失效时间(leaseTime=-1),则默认配置30秒 (getLockWatchdogTimeOut()=30) RFuture
<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime, commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); // 加锁完毕之后,启动看门狗线程,定时的延期失效时间(定时任务为 internalLockLeaseTime / 3 毫秒之后执行)
     ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
       if (e != null) {
                return;
            }
            // lock acquired
            if (ttlRemaining == null) {
          // 启动看门狗任务 scheduleExpirationRenewal(threadId); } });
return ttlRemainingFuture; }
    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);
     // 通过lua脚本访问Redis,保证操作的原子性, 以及达到批量操作的效果,提升性能
     
// KEYS[1] :需要加锁的key,这里需要是字符串类型。
     // ARGV[1] :锁的超时时间,防止死锁
     // ARGV[2] :锁的唯一标识,id(UUID.randomUUID()) + “:” + threadId
     return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              // 检查是否key已经被占用,如果没有则设置超时时间和唯一标识,初始化value=1
     "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; " +
               // 如果锁重入,需要判断锁的key field 都一直情况下 value 加一
               "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(getName()), internalLockLeaseTime, getLockName(threadId)); }
  • unlock方法
  @Override
    public RFuture<Void> unlockAsync(long threadId) {
        RPromise<Void> result = new RedissonPromise<Void>();
        RFuture<Boolean> future = unlockInnerAsync(threadId);

        future.onComplete((opStatus, e) -> {
       // 释放锁后取消刷新锁失效时间的调度任务 cancelExpirationRenewal(threadId);
if (e != null) { result.tryFailure(e); return; }        // 非锁的持有者释放锁时抛出异常 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; } result.trySuccess(null); }); return result; }   // 通过 Lua 脚本执行 Redis 命令释放锁

   // KEYS[1] :需要加锁的key,这里需要是字符串类型。

   // KEYS[2] :redis消息的ChannelName,一个分布式锁对应唯一的一个channelName:“redisson_lock__channel__{” + getName() + “}”

   // ARGV[1] :reids消息体,这里只需要一个字节的标记就可以,主要标记redis的key已经解锁,再结合redis的Subscribe,能唤醒其他订阅解锁消息的客户端线程申请锁。=

   // ARGV[2] :锁的超时时间,防止死锁

   // ARGV[3] :锁的唯一标识,也就是刚才介绍的 id(UUID.randomUUID()) + “:” + threadId

  protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
               // key和field不匹配,说明当前客户端线程没有持有锁,不能主动解锁。      
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
               // 如果counter>0说明锁在重入,不能删除key "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "else " +
               // 删除key并且publish 解锁消息 "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; " + "end; " + "return nil;"
, Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)); }

使用 EVAL 命令执行 Lua 脚本来释放锁:

  1. key 不存在,说明锁已释放,直接执行 publish 命令发布释放锁消息并返回 1
  2. key 存在,但是 field 在 Hash 中不存在,说明自己不是锁持有者,无权释放锁,返回 nil
  3. 因为锁可重入,所以释放锁时不能把所有已获取的锁全都释放掉,一次只能释放一把锁,因此执行 hincrby 对锁的值减一
  4. 释放一把锁后,如果还有剩余的锁,则刷新锁的失效时间并返回 0;如果刚才释放的已经是最后一把锁,则执行 del 命令删除锁的 key,并发布锁释放消息,返回 1

注意这里有个实际开发过程中,容易出现很容易出现上面第二步异常,非锁的持有者释放锁时抛出异常。

大体流程图:

 

 

特点

  • 分布式
  • 可以自动释放锁,防止死锁
  • 可重入锁 (相同线程不需要在等待锁,而是可以直接进行相应操作)
  • 防误删,当前线程只能删除当前线程的锁 (业务执行时间过长,超过锁失效时间,锁被释放,第二个线程获取锁,此时第一个线程执行到释放锁代码时,不能删除第二个线程的锁)
  • 可阻塞等待
  • 看门狗机制,延长过期时间(没有设置过期时间的情况,leaseTime=-1,默认失效时间为30秒,启动看门狗线程,定时检查是否需要延长时间scheduleExpirationRenewal)
  • 锁种类多样:可重入锁、公平锁、联锁、红锁、读写锁

存在的问题

  分布式架构中的CAP理论,分布式系统只能同时满足两个

    • 一致性(Consistency)
    • 可用性(Availability)
    • 分区容错性(Partition tolerance)
  • Redisson分布式锁是AP模式,当锁存在的redis节点宕机,可能会被误判为锁失效,或者没有加锁。(Zookeeper实现的分布式锁,是CP理论)

    -- 解决办法:

  1. tradoff,分布式锁的redis采用单机部署,分布式锁专用
  2. 如果对锁比较关注,一致性要求比较高,可以使用ZK实现的分布式锁

与Zookeeper实现的分布式锁比较

 TODO

总结

  • 如果考虑各种网络、宕机等原因,很多问题需要考虑,问题会变的复杂,其实分布式锁的应用场景不多,很多情况可以绕开分布式锁,使用其他方式解决,比如 队列,异步,响应式
  • 个人经验:分布式锁的场景,更多的应用是一个操作不能同时多处进行,不能短时间内重复执行,需要幂等操作等场景,比如:防止快速的重复提交,mq与定时任务双线更改状态,防止消息重复消费 等等。这些情况一般使用setNx即可解决
  • 所谓的减库存其实也用不到分布式锁

 

 

参考:

Redisson实现分布式锁(1)---原理

Redisson实现分布式锁(2)—RedissonLock

利用Redisson实现分布式锁及其底层原理解析

分布式锁在面试中会被问到哪一些知识点?[视频]

原文链接:https://www.cnblogs.com/dong320/p/13993553.html



所属网站分类: 技术文章 > 博客

作者:java大师

链接:http://www.javaheidong.com/blog/article/1131/dbe58db13ab1fbc85418/

来源:java黑洞网

任何形式的转载都请注明出处,如有侵权 一经发现 必将追究其法律责任

27 0
收藏该文
已收藏

评论内容:(最多支持255个字符)