Python微信订餐小程序课程视频

https://edu.csdn.net/course/detail/36074

Python实战量化交易理财系统

https://edu.csdn.net/course/detail/35475

一、前言

我们在实现使用Redis实现分布式锁，最开始一般使用SET resource-name anystring NX EX max-lock-time进行加锁，使用Lua脚本保证原子性进行实现释放锁。这样手动实现比较麻烦，对此Redis官网也明确说Java版使用Redisson来实现。小编也是看了官网慢慢的摸索清楚，特写此记录一下。从官网到整合Springboot到源码解读，以单节点为例，小编的理解都在注释里，希望可以帮助到大家！！

二、为什么使用Redisson

1. 我们打开官网

redis中文官网

2. 我们可以看到官方让我们去使用其他

3. 打开官方推荐

4. 找到文档

Redisson地址

5. Redisson结构

三、Springboot整合Redisson

1. 导入依赖

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-data-redisartifactId>
dependency>
<dependency>
    <groupId>redis.clientsgroupId>
    <artifactId>jedisartifactId>
dependency>

<dependency>
    <groupId>org.redissongroupId>
    <artifactId>redissonartifactId>
    <version>3.12.0version>
dependency>

2. 以官网为例查看如何配置

3. 编写配置类

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author wangzhenjun
 * @date 2022/2/9 9:57
 */
@Configuration
public class MyRedissonConfig {

    /**
 * 所有对redisson的使用都是通过RedissonClient来操作的
 * @return
 */
    @Bean(destroyMethod="shutdown")
    public RedissonClient redisson(){
        // 1. 创建配置
        Config config = new Config();
        // 一定要加redis://
        config.useSingleServer().setAddress("redis://192.168.17.130:6379");
        // 2. 根据config创建出redissonClient实例
        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }
}

4. 官网测试加锁例子

5. 根据官网简单Controller接口编写

@ResponseBody
@GetMapping("/hello")
public String hello(){
    // 1.获取一把锁，只要锁名字一样，就是同一把锁
    RLock lock = redisson.getLock("my-lock");
    // 2. 加锁
    lock.lock();// 阻塞试等待 默认加的都是30s
    // 带参数情况
    // lock.lock(10, TimeUnit.SECONDS);// 10s自动解锁，自动解锁时间一定要大于业务的执行时间。
    try {
        System.out.println("加锁成功" + Thread.currentThread().getId());
        Thread.sleep(30000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        // 3. 解锁
        System.out.println("解锁成功：" + Thread.currentThread().getId());
        lock.unlock();
    }
    return "hello";
}

6. 测试

四、lock.lock()源码分析

1. 打开RedissonLock实现类

2. 找到实现方法

@Override
public void lock() {
    try {
        // 我们发现不穿过期时间源码默认过期时间为-1
        lock(-1, null, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

3. 按住Ctrl进去lock方法

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    // 获取线程的id，占有锁的时候field的值为UUID:线程号id
    long threadId = Thread.currentThread().getId();
    // 尝试获得锁
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired 获得锁，返回
    if (ttl == null) {
        return;
    }
    // 这里说明获取锁失败，就通过线程id订阅这个锁
    RFuture future = subscribe(threadId);
 if (interruptibly) {
 commandExecutor.syncSubscriptionInterrupted(future);
 } else {
 commandExecutor.syncSubscription(future);
 }

 try {
 // 这里进行自旋，不断尝试获取锁
 while (true) {
 // 继续尝试获取锁
 ttl = tryAcquire(leaseTime, unit, threadId);
 // lock acquired 获取成功
 if (ttl == null) {
 // 直接返回，挑出自旋
 break;
 }

 // waiting for message 继续等待获得锁
 if (ttl >= 0) {
 try {
 future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
 } catch (InterruptedException e) {
 if (interruptibly) {
 throw e;
 }
 future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
 }
 } else {
 if (interruptibly) {
 future.getNow().getLatch().acquire();
 } else {
 future.getNow().getLatch().acquireUninterruptibly();
 }
 }
 }
 } finally {
 // 取消订阅
 unsubscribe(future, threadId);
 }
// get(lockAsync(leaseTime, unit));
}

4. 进去尝试获取锁方法

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    // 直接进入异步方法
    return get(tryAcquireAsync(leaseTime, unit, threadId));
}

private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
 // 这里进行判断如果没有设置参数leaseTime = -1
 if (leaseTime != -1) {
 return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL\_LONG);
 }
 // 此方法进行获得锁，过期时间为看门狗的默认时间
 // private long lockWatchdogTimeout = 30 * 1000;看门狗默认过期时间为30s
 // 加锁和过期时间要保证原子性，这个方法后面肯定调用执行了Lua脚本，我们下面在看
 RFuture ttlRemainingFuture = tryLockInnerAsync(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;
}

5. 查看tryLockInnerAsync()方法

 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand 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]); " +
 "return nil; " +
 "end; " +
 // hexists查看哈希表的指定字段是否存在，存在锁并且是当前线程持有锁
 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
 // hincrby自增一
 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
 // 锁的值大于1，说明是可重入锁，重置过期时间
 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
 "return nil; " +
 "end; " +
 // 锁已存在，且不是本线程，则返回过期时间ttl
 "return redis.call('pttl', KEYS[1]);",
 Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

6. 进入4留下的定时任务scheduleExpirationRenewal()方法

一步步往下找源码：scheduleExpirationRenewal --->renewExpiration

根据下面源码，定时任务刷新时间为：internalLockLeaseTime / 3，是看门狗的1/3，即为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 future = renewExpirationAsync(threadId);
 future.onComplete((res, e) -> {
 if (e != null) {
 log.error("Can't update lock " + getName() + " expiration", e);
 return;
 }

 if (res) {
 // reschedule itself
 renewExpiration();
 }
 });
 }
 }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

 ee.setTimeout(task);
}

五、lock.lock(10, TimeUnit.SECONDS)源码分析

1. 打开实现类

@Override
public void lock(long leaseTime, TimeUnit unit) {
    try {
        // 这里的过期时间为我们输入的10
        lock(leaseTime, unit, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

2. 方法lock()实现展示，同三.3源码

3. 直接来到尝试获得锁tryAcquireAsync()方法

private  RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
 // 这里进行判断如果没有设置参数leaseTime = -1,此时我们为10
 if (leaseTime != -1) {
 // 来到此方法
 return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL\_LONG);
 }
 // 此处省略后面内容，前面以详细说明。。。。
}

4. 打开tryLockInnerAsync()方法

我们不难发现和没有传过期时间的方法一样，只不过leaseTime的值变了。

 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand 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]); " +
 "return nil; " +
 "end; " +
 // hexists查看哈希表的指定字段是否存在，存在锁并且是当前线程持有锁
 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
 // hincrby自增一
 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
 // 锁的值大于1，说明是可重入锁，重置过期时间
 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
 "return nil; " +
 "end; " +
 // 锁已存在，且不是本线程，则返回过期时间ttl
 "return redis.call('pttl', KEYS[1]);",
 Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

六、lock.unlock()源码分析

1. 打开方法实现

@Override
public void unlock() {
    try {
        // 点击进入释放锁方法
        get(unlockAsync(Thread.currentThread().getId()));
    } catch (RedisException e) {
        if (e.getCause() instanceof IllegalMonitorStateException) {
            throw (IllegalMonitorStateException) e.getCause();
        } else {
            throw e;
        }
    }

// Future future = unlockAsync();
// future.awaitUninterruptibly();
// if (future.isSuccess()) {
// return;
// }
// if (future.cause() instanceof IllegalMonitorStateException) {
// throw (IllegalMonitorStateException)future.cause();
// }
// throw commandExecutor.convertException(future);
}

2. 打开unlockAsync()方法

@Override
public RFuture unlockAsync(long threadId) {
 RPromise result = new RedissonPromise();
 // 解锁方法，后面展开说
 RFuture future = unlockInnerAsync(threadId);
 // 完成
 future.onComplete((opStatus, e) -> {
 if (e != null) {
 // 取消到期续订
 cancelExpirationRenewal(threadId);
 // 将这个未来标记为失败并通知所有人
 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;
 }

 cancelExpirationRenewal(threadId);
 result.trySuccess(null);
 });

 return result;
}

3. 打开unlockInnerAsync()方法

protected RFuture unlockInnerAsync(long threadId) {
 return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL\_BOOLEAN,
 // 判断释放锁的线程和已存在锁的线程是不是同一个线程，不是返回空
 "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
 "return nil;" +
 "end; " +
 // 释放锁后，加锁次数减一
 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
 // 判断剩余数量是否大于0
 "if (counter > 0) then " +
 // 大于0 ，则刷新过期时间
 "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;",
 Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK\_MESSAGE, internalLockLeaseTime, getLockName(threadId));

}

七、总结

这样大家就跟着小编走完了一遍底层源码，是不是感觉自己又行了，哈哈哈。小编走下来一遍觉得收货还是蛮大的，以前不敢点进去源码，进去就懵逼了，所以人要大胆的向前迈出第一步。一起加油吧，看到这里不一键三连，有点对不起小编了哦！！

顺便推广一下自己的网站！！！
点击访问！欢迎访问，里面也是有很多好的文章哦！

转载请注明：xuhss » Springboot基于Redisson实现Redis分布式可重入锁【案例到源码分析】