前言
在上一节内容中，我们已经实现了使用redis分布式锁解决商品“超卖”的问题，本节内容是对redis分布式锁的优化。在上一节的redis分布式锁中，我们的锁有俩个可以优化的问题。第一，锁需要实现可重入，同一个线程不用重复去获取锁；第二，锁没有续期功能，导致业务没有执行完成就已经释放了锁，存在一定的并发访问问题。本案例中通过使用redis的hash数据结构实现可重入锁，使用Timer实现锁的续期功能，完成redis分布式锁的优化。最后，我们通过集成第三方redisson工具包，完成分布式锁以上俩点的优化内容。Redisson提供了简单易用的API，使得开发人员可以轻松地在分布式环境中使用Redis。
正文
- 加锁的lua脚本：使用exists和hexists指令判断是否存在锁，如果不存在或者存在锁并且该锁下面的field有值，就使用hincrby指令使锁的值加1，实现可重入，否则直接返回0，加锁失败。
if redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
"then " +
" redis.call('hincrby', KEYS[1], ARGV[1], 1) " +
" redis.call('expire', KEYS[1], ARGV[2]) " +
" return 1 " +
"else " +
" return 0 " +
"end"- 解锁的lua脚本: 使用hexists指令判断是否存在锁，如果为0，代表没有对应field字段的锁，直接返回nil；如果使用hincrby指令使锁field字段锁的值减少1之后值为0，代表锁已经不在占用，可以删除该锁；否则直接返回0，代表是可重入锁，锁还没有释放。
if redis.call('hexists', KEYS[1], ARGV[1]) == 0 " +
"then " +
" return nil " +
"elseif redis.call('hincrby', KEYS[1], ARGV[1], -1) == 0 " +
"then " +
" return redis.call('del', KEYS[1]) " +
"else " +
" return 0 " +
"end"- 实现续期的lua脚本：使用hexists指令判断锁的field值是否存在，如果值为1存在，则将该锁的过期时间更新，否则直接返回0，代表没有找到该锁，续期失败。
if redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
"then " +
" return redis.call('expire', KEYS[1], ARGV[2]) " +
"else " +
" return 0 " +
"end";- 创建一个自定义的锁工具类MyRedisDistributeLock，实现加锁、解锁、续期功能
- MyRedisDistributeLock实现
package com.ht.atp.plat.util; import org.jetbrains.annotations.NotNull; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import java.util.Arrays; import java.util.Timer; import java.util.TimerTask; import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; public class MyRedisDistributeLock implements Lock { public MyRedisDistributeLock(StringRedisTemplate redisTemplate, String lockName, long expire) { this.redisTemplate = redisTemplate; this.lockName = lockName; this.expire = expire; this.uuid = getId(); } /** * redis工具类 */ private StringRedisTemplate redisTemplate; /** * 锁名称 */ private String lockName; /** * 过期时间 */ private Long expire; /** * 锁的值 */ private String uuid; @Override public void lock() { this.tryLock(); } @Override public void lockInterruptibly() { } @Override public boolean tryLock() { try { return this.tryLock(-1L, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } return false; } @Override public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException { if (time != -1) { this.expire = unit.toSeconds(time); } String script = "if redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1 " + "then " + " redis.call('hincrby', KEYS[1], ARGV[1], 1) " + " redis.call('expire', KEYS[1], ARGV[2]) " + " return 1 " + "else " + " return 0 " + "end"; while (!this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuid, String.valueOf(expire))) { Thread.sleep(50); } // //加锁成功后，自动续期 this.renewExpire(); return true; } @Override public void unlock() { String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 0 " + "then " + " return nil " + "elseif redis.call('hincrby', KEYS[1], ARGV[1], -1) == 0 " + "then " + " return redis.call('del', KEYS[1]) " + "else " + " return 0 " + "end"; Long flag = this.redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuid); if (flag == null) { throw new IllegalMonitorStateException("this lock doesn't belong to you!"); } } @NotNull @Override public Condition newCondition() { return null; } /** * 给线程拼接唯一标识 * * @return */ private String getId() { return UUID.randomUUID() + "-" + Thread.currentThread().getId(); } private void renewExpire() { String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 1 " + "then " + " return redis.call('expire', KEYS[1], ARGV[2]) " + "else " + " return 0 " + "end"; new Timer().schedule(new TimerTask() { @Override public void run() { System.out.println("-------------------"); Boolean flag = redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuid, String.valueOf(expire)); if (flag) { renewExpire(); } } }, this.expire * 1000 / 3); } }- 实现加锁功能
- 实现解锁功能
- 使用Timer实现锁的续期功能
- 使用MyRedisDistributeLock实现库存的加锁业务
- 使用自定义MyRedisDistributeLock工具类实现加锁业务
public void checkAndReduceStock() { //1.获取锁 MyRedisDistributeLock myRedisDistributeLock = new MyRedisDistributeLock(stringRedisTemplate, "stock", 10); myRedisDistributeLock.lock(); try { // 2. 查询库存数量 String stockQuantity = stringRedisTemplate.opsForValue().get("P0001"); // 3. 判断库存是否充足 if (stockQuantity != null && stockQuantity.length() != 0) { Integer quantity = Integer.valueOf(stockQuantity); if (quantity > 0) { // 4.扣减库存 stringRedisTemplate.opsForValue().set("P0001", String.valueOf(--quantity)); } } else { System.out.println("该库存不存在！"); } } finally { myRedisDistributeLock.unlock(); } }- 启动服务7000、7001、7002，压测优化后的自定义分布式锁：平均访问时间362ms,吞吐量每秒246，库存扣减为0，表明优化后的分布式锁是可用的。
- 集成redisson工具包，使用第三方工具包实现分布式锁，完成并发访问“超卖”问题案例演示
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.11.6</version>
</dependency>- 创建一个redisson配置类，引入redisson客户端工具
package com.ht.atp.plat.config;
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;
@Configuration
public class MyRedissonConfig {
@Bean
RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer()
.setAddress("redis://192.168.110.88:6379");
//配置看门狗的默认超时时间为30s,供续期使用
config.setLockWatchdogTimeout(30000);
return Redisson.create(config);
}
}
- 使用Redisson锁实现“超卖”业务方法
//可重入锁
@Override
public void checkAndReduceStock() {
// 1.加锁，获取锁失败重试
RLock lock = this.redissonClient.getLock("lock");
lock.lock();
try {
// 2. 查询库存数量
String stockQuantity = stringRedisTemplate.opsForValue().get("P0001");
// 3. 判断库存是否充足
if (stockQuantity != null && stockQuantity.length() != 0) {
Integer quantity = Integer.valueOf(stockQuantity);
if (quantity > 0) {
// 4.扣减库存
stringRedisTemplate.opsForValue().set("P0001", String.valueOf(--quantity));
}
} else {
System.out.println("该库存不存在！");
}
} finally {
// 4.释放锁
lock.unlock();
}
}- 开启7000、7001、7002服务，压测扣减库存接口
- 压测结果：平均访问时间222ms，吞吐量为384每秒
- 库存扣减结果为0
结语
综上所述，无论是自定义分布式锁还是使用redisson工具类，都能实现分布式锁解决并发访问的“超卖问题”，redisson工具使用集成更加方便简洁，推荐使用redisson工具包。本节内容到这里就结束了，我们下期见。。。。。。
