分布式锁 Redission 介绍及使用其可重入锁 和 WatchDog 机制 和 MutiLock原理

Posted 2022-09-20 Perceus

(目录)

---

上一篇博文部分：

---

分布式锁-redission

1、 redission功能介绍

基于setnx实现的分布式锁存在下面的问题：

---

那么什么是Redission呢 ?

Redission提供了分布式锁的多种多样的功能

---

2、Redission快速入门

<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson</artifactId>
	<version>3.13.6</version>
</dependency>

@Configuration
public class RedissonConfig 

    @Bean
    public RedissonClient redissonClient()
        // 配置类
        Config config = new Config();
        // 添加Redis地址，这里添加单节点的地址，也可以使用 config.userClusterServers() 添加集群
        config.useSingleServer().setAddress("redis://192.168.150.101:6379")
                .setPassword("123321");
        // 创建RedissonClient对象
        return Redisson.create(config);
    

---

@Resource
private RedissionClient redissonClient;

@Test
void testRedisson() throws Exception
    //获取锁(可重入)，指定锁的名称
    RLock lock = redissonClient.getLock("anyLock");
    //尝试获取锁，参数分别是：获取锁的最大等待时间(期间会重试)，锁自动释放时间，时间单位
    boolean isLock = lock.tryLock(1,10,TimeUnit.SECONDS);
    //判断获取锁成功
    if(isLock)
        try
            System.out.println("执行业务");          
        finally
            //释放锁
            lock.unlock();
        
        
    
    

---

在 VoucherOrderServiceImpl，注入RedissonClient

@Resource
private RedissonClient redissonClient;

@Override
public Result seckillVoucher(Long voucherId) 
        // 1.查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 2.判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) 
            // 尚未开始
            return Result.fail("秒杀尚未开始！");
        
        // 3.判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) 
            // 尚未开始
            return Result.fail("秒杀已经结束！");
        
        // 4.判断库存是否充足
        if (voucher.getStock() < 1) 
            // 库存不足
            return Result.fail("库存不足！");
        
        Long userId = UserHolder.getUser().getId();
        //创建锁对象 这个代码不用了，因为我们现在要使用分布式锁
        //SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //获取锁对象
        boolean isLock = lock.tryLock();
       
		//加锁失败
        if (!isLock) 
            return Result.fail("不允许重复下单");
        
        try 
            //获取代理对象(事务)
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
         finally 
            //释放锁
            lock.unlock();
        
 

JMeter并发测试：

---

3、 redission可重入锁原理

---

在Lock锁中，他是借助于底层的一个voaltile的一个state变量来记录重入的状态。

---

redission中，我们的也支持支持可重入锁

---

所以接下来我们一起分析一下当前的这个lua表达式

这个地方一共有3个参数

exists: 判断数据是否存在 name：是lock是否存在,如果==0，就表示当前这把锁不存在

--此时他就开始往redis里边去写数据 ，写成一个hash结构
redis.call(hset, KEYS[1], ARGV[2], 1); 

Lock
 id + **":"** + threadId :  1

如果当前这把锁存在，则第一个条件不满足，再判断

redis.call(hexists, KEYS[1], ARGV[2]) == 1

此时需要通过大key+小key判断当前这把锁是否是属于自己的，如果是自己的，则进行

redis.call(hincrby, KEYS[1], ARGV[2], 1)

将当前这个锁的value进行+1 ，redis.call(pexpire, KEYS[1], ARGV[1]); 然后再对其设置过期时间，如果以上两个条件都不满足，则表示当前这把锁抢锁失败，最后返回pttl，即为当前这把锁的失效时间

如果小伙帮们看了前边的源码， 你会发现他会去判断当前这个方法的返回值是否为null，如果是null，则对应则前两个if对应的条件，退出抢锁逻辑，如果返回的不是null，即走了第三个分支，在源码处会进行while(true)的自旋抢锁。

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

---

4、 redission锁重试和WatchDog机制

说明：

抢锁过程中，获得当前线程，通过tryAcquire进行抢锁，该抢锁逻辑和之前逻辑相同

---

所以如果返回是null，则代表着当前这哥们已经抢锁完毕，或者可重入完毕，但是如果以上两个条件都不满足，则进入到第三个条件，返回的是锁的失效时间，可以自行往下翻一点点，你能发现有个while( true) 再次进行tryAcquire进行抢锁

long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) 
    return;

if (leaseTime != -1) 
    return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);

commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout() // 30s

RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                        commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                        TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> 
    if (e != null) 
        return;
    

    // lock acquired
    if (ttlRemaining == null) 
        scheduleExpirationRenewal(threadId);
    
);
return ttlRemainingFuture;

此逻辑就是续约逻辑，注意看 commandExecutor.getConnectionManager().newTimeout（） 此方法

Method(  new TimerTask() ,参数2 ，参数3  )

指的是：通过参数2，参数3 去描述什么时候去做参数1的事情，现在的情况是：10s之后去做参数一的事情

那么大家可以想一想，假设我们的线程出现了宕机他还会续约吗？

private void renewExpiration() 
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) 
        return;
    
    
    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;
            
            
            RFuture<Boolean> future = renewExpirationAsync(threadId);
            future.onComplete((res, e) -> 
                if (e != null) 
                    log.error("Cant update lock " + getName() + " expiration", e);
                    return;
                
                
                if (res) 
                    // reschedule itself
                    renewExpiration();
                
            );
        
    , internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    
    ee.setTimeout(task);

---

流程图解：

Redisson 分布式锁原理：

---

5、 redission锁的MutiLock原理

为了提高redis的可用性，我们会搭建集群或者主从

以主从为例：

MutiLock 加锁原理是什么呢？

---