灌云县建设局网站,外包公司做网站多少,wordpress plugin development,英语复试口语模板2022黑马Redis跟学笔记.实战篇 四4.3.秒杀优惠券功能4.3.1.秒杀优惠券的基本实现一、优惠卷秒杀1.1 全局唯一ID1.2 Redis实现全局唯一Id1.3 添加优惠卷1.4 实现秒杀下单4.3.2.超卖问题4.3.3.基于乐观锁解决超卖问题1. 悲观锁2. 乐观锁3. 乐观锁解决超卖问题4.4 秒杀的一人一单限…
2022黑马Redis跟学笔记.实战篇 四4.3.秒杀优惠券功能4.3.1.秒杀优惠券的基本实现一、优惠卷秒杀1.1 全局唯一ID1.2 Redis实现全局唯一Id1.3 添加优惠卷1.4 实现秒杀下单4.3.2.超卖问题4.3.3.基于乐观锁解决超卖问题1. 悲观锁2. 乐观锁3. 乐观锁解决超卖问题4.4 秒杀的一人一单限制功能4.4.1 实现秒杀的一人一单限制优惠券秒杀一人一单4.4.2.单机模式下的线程安全问题4.4.3 集群模式下的线程安全问题4.4.4 分布式锁4.4.4.1 分布式锁原理4.4.4.2 Redis的String结构实现分布式锁4.4.4.2.1 实现分布式锁版本一4.4.4.3 锁误删问题1.Redis分布式锁误删情况说明2. 解决Redis分布式锁误删问题4.4.4.4 分布式锁的原子性操作问题4.4.4.5 Lua脚本解决原子性问题1. 利用Java代码调用Lua脚本改造分布式锁4.4.4.6 Redission分布式锁1.分布式锁 - redission功能介绍2. 分布式锁-Redission快速入门4.4.4.7 Hash结构解决锁的可重入问题分布式锁-redission可重入锁原理4.4.4.8 watchDog解决锁超时释放问题1.分布式锁-redission锁重试和WatchDog机制2. 分布式锁-redission锁的MutiLock原理4.3.秒杀优惠券功能
4.3.1.秒杀优惠券的基本实现
一、优惠卷秒杀
1.1 全局唯一ID
每个店铺都可以发布优惠券
当用户抢购时就会生成订单并保存到tb_voucher_order这张表中而订单表如果使用数据库自增ID就存在一些问题
id的规律性太明显受单表数据量的限制 场景分析如果我们的id具有太明显的规则用户或者说商业对手很容易猜测出来我们的一些敏感信息比如商城在一天时间内卖出了多少单这明显不合适。
场景分析二随着我们商城规模越来越大mysql的单表的容量不宜超过500W数据量过大之后我们要进行拆库拆表但拆分表了之后他们从逻辑上讲他们是同一张表所以他们的id是不能一样的 于是乎我们需要保证id的唯一性。
全局ID生成器是一种在分布式系统下用来生成全局唯一ID的工具一般要满足下列特性 Redis都能实现以上5点 唯一性想到了关键字incrby 高可用集群方案、主从方案、哨兵方案 高性能内存存储性能好 递增采用递增 安全性自增然后再拼接一些其它信息让规律不要那么明显
为了增加ID的安全性我们可以不直接使用Redis自增的数值而是拼接一些其它信息
ID的组成部分符号位1bit永远为0
时间戳31bit以秒为单位可以使用69年
序列号32bit秒内的计数器支持每秒产生232个不同ID
1.2 Redis实现全局唯一Id
新建类RedisWorker.java 编写测试的主方法查看设置的时间2023年1月1日0点0分距离现在的时间偏移量
public static void main(String[] args) {// 设置初始时间2023年1月1日 0点0分0秒LocalDateTime time LocalDateTime.of(2023, 1, 1, 0, 0, 0);// 该方法将此本地时间与作为参数传递的指定日期和偏移量相结合以计算epoch-second值long second time.toEpochSecond(ZoneOffset.UTC);System.out.println(second: second);}运行主方法 得到这个second之后就作为常量当作初始时间戳。 修改RedisWorker.java
Component
public class RedisWorker {/*** 把距离当前时间的偏移量作为时间戳*/private static final long BEGIN_TIMESTAMP 1672531200L;/*** 序列号的长度(位数)*/private static final int COUNT_BITS 32;Resourceprivate StringRedisTemplate stringRedisTemplate;public long nextID(String keyPrefix) {// 1.生成时间戳LocalDateTime now LocalDateTime.now();long nowSecond now.toEpochSecond(ZoneOffset.UTC);long timestamp nowSecond - BEGIN_TIMESTAMP;// 2.生成序列号// 2.1 获取当前日期精确到天,保证一天生成一个key// 2.2 自增长String date now.format(DateTimeFormatter.ofPattern(yyyyMMdd));Long sequence stringRedisTemplate.opsForValue().increment(icr: keyPrefix : date);//3.拼接返回long id (timestamp COUNT_BITS) | sequence;return id;}}测试类
知识小贴士关于countdownlatch
countdownlatch名为信号枪主要的作用是同步协调在多线程的等待于唤醒问题
我们如果没有CountDownLatch 那么由于程序是异步的当异步程序没有执行完时主线程就已经执行完了然后我们期望的是分线程全部走完之后主线程再走所以我们此时需要使用到CountDownLatch。
CountDownLatch 中有两个最重要的方法
1、countDown2、await
await 方法 是阻塞方法我们担心分线程没有执行完时main线程就先执行所以使用await可以让main线程阻塞那么什么时候main线程不再阻塞呢当CountDownLatch 内部维护的 变量变为0时就不再阻塞直接放行那么什么时候CountDownLatch 维护的变量变为0 呢我们只需要调用一次countDown 内部变量就减少1我们让分线程和变量绑定 执行完一个分线程就减少一个变量当分线程全部走完CountDownLatch 维护的变量就是0此时await就不再阻塞统计出来的时间也就是所有分线程执行完后的时间。
修改CommentApplicationTests.java
Testpublic void getId() throws InterruptedException {ExecutorService pools CacheClient.newFixedThreadPool(500);// 程序计数器 设置的数量和循环数量一致CountDownLatch latch new CountDownLatch(300);Runnable runnable new Runnable() {Overridepublic void run() {for (int i 0; i 100; i) {long id redisWorker.nextID(order);System.out.println(id: id);}// 每一个线程跑完就剪掉一次计数(倒计时)latch.countDown();}};runnable.run();long begin System.currentTimeMillis();for (int i 0; i 300; i) {pools.submit(runnable);}latch.await();long end System.currentTimeMillis();System.out.println(总时间是: (end - begin));}运行测试类查看结果没有重复的ID 这些是10进制的我们粘贴到科学计算器中用二进制看一下是64位的 看一下Redis,自增到30100
1.3 添加优惠卷
每个店铺都可以发布优惠券分为平价券和特价券。平价券可以任意购买而特价券需要秒杀抢购 tb_voucher优惠券的基本信息优惠金额、使用规则等 tb_seckill_voucher优惠券的库存、开始抢购时间结束抢购时间。特价优惠券才需要填写这些信息
平价卷由于优惠力度并不是很大所以是可以任意领取
而代金券由于优惠力度大所以像第二种卷就得限制数量从表结构上也能看出特价卷除了具有优惠卷的基本信息以外还具有库存抢购时间结束时间等等字段
新增普通卷代码: 在PostMan中加入如下语句
{shopId:2,title:50元代金券,subTitle:周一至周日均可使用,rules:每日特惠\\n无需预约\\n可以无限叠加\\不兑现、不找零\\n仅限堂食,payValue:4700,actualValue:5000,type:0,stock:200,beginTime:2023-01-01T09:00:00,endTime:2023-03-01T12:00:00
}添加成功 看数据库
新增秒杀卷代码 用PostMan添加数据 json如下
{shopId:2,title:100元代金券,subTitle:每天都可以使用,rules:兔年特惠\\n无需预约\\n可以无限叠加\\不兑现、不找零\\n仅限堂食,payValue:8000,actualValue:10000,type:1,stock:100,beginTime:2023-01-01T09:00:00,endTime:2023-03-01T12:00:00
}点击send后 数据库中 登录后点击抢购优惠券 打开开发者工具可以发现优惠券ID拼接在最后POST请求
地址
http://localhost:8080/api/voucher-order/seckill/21.4 实现秒杀下单
下单核心思路当我们点击抢购时会触发右侧的请求我们只需要编写对应的controller即可。 秒杀下单应该思考的内容
下单时需要判断两点
秒杀是否开始或结束如果尚未开始或已经结束则无法下单库存是否充足不足则无法下单
下单核心逻辑分析
当用户开始进行下单我们应当去查询优惠卷信息查询到优惠卷信息判断是否满足秒杀条件。
比如时间是否充足如果时间充足则进一步判断库存是否足够如果两者都满足则扣减库存创建订单然后返回订单id如果有一个条件不满足则直接结束。
修改VoucherOrderController.java
RestController
RequestMapping(/voucher-order)
public class VoucherOrderController {Autowiredprivate VoucherOrderServiceImpl voucherOrderService;PostMapping(seckill/{id})public Result seckillVoucher(PathVariable(id) Long voucherId) {return voucherOrderService.seckillVoucher(voucherId);}
}修改接口IVoucherOrderService.java
public interface IVoucherOrderService extends IServiceVoucherOrder {Result seckillVoucher(Long voucherId);
}修改VoucherOrderServiceImpl.java
Service
public class VoucherOrderServiceImpl extends ServiceImplVoucherOrderMapper, VoucherOrder implements IVoucherOrderService {Autowiredprivate ISeckillVoucherService iSeckillVoucherService;Autowiredprivate RedisWorker redisWorker;OverrideTransactionalpublic Result seckillVoucher(Long voucherId) {// 1.查询秒杀优惠券信息// select * from tb_seckill_voucher where voucher_id ?SeckillVoucher seckillVoucher iSeckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始LocalDateTime beginTime seckillVoucher.getBeginTime();LocalDateTime endTime seckillVoucher.getEndTime();LocalDateTime now LocalDateTime.now();if (now.isBefore(beginTime)) {// 当前时间早于秒杀开始时间说明秒杀没有开始return Result.fail(秒杀尚未开始,请耐心等待秒杀开始时间: beginTime.format(DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss)));}// 3.判断秒杀是否已经结束if (now.isAfter(endTime)) {// 当前时间晚于秒杀结束时间说明秒杀结束了return Result.fail(秒杀已经结束,感谢支持!);}// 4.判断库存是否充足Integer stock seckillVoucher.getStock();if (stock 0) {return Result.fail(商品已经售罄!);}// 5.扣减库存boolean result iSeckillVoucherService.update().setSql(stock stock -1).eq(voucher_id, voucherId).update();if (!result) {return Result.fail(商品已经售罄!);}// 6.创建订单/*** 获取订单id*/long orderId redisWorker.nextID(order);VoucherOrder voucherOrder new VoucherOrder();voucherOrder.setId(orderId);voucherOrder.setVoucherId(voucherId);voucherOrder.setUserId(UserHolder.getUser().getId());// 将订单信息保存到数据库// insert into tb_voucher_order values ()save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);}
}
重启应用为了防止后面一直登录用户费时间我们把token设置成永久 设置TTL 点击限时抢购 显示抢购成功然后我们去数据库看一下库存扣减成功 订单增加了1条
4.3.2.超卖问题
为了方便观察更改数据库的库存
UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;清空订单数据
DELETE FROM tb_voucher_order;打开JMeter,配置相关操作模拟多线程操作 配置Http请求 配置HTTP信息头管理器 配置断言
配置完成后点击启动 查看结果树 看聚合报告
看数据库发现产生了109个订单 看秒杀优惠券库存出现超卖问题
有关超卖问题分析在我们原有代码中是这么写的
假设线程1过来查询库存判断出来库存大于0正准备去扣减库存但是还没有来得及去扣减此时线程2过来线程2也去查询库存发现这个数量一定也大于0那么这两个线程都会去扣减库存最终多个线程相当于一起去扣减库存此时就会出现库存的超卖问题。 超卖问题是典型的多线程安全问题针对这一问题的常见解决方案就是加锁而对于加锁我们通常有两种解决方案见下图 4.3.3.基于乐观锁解决超卖问题
1. 悲观锁
悲观锁可以实现对于数据的串行化执行比如syn和lock都是悲观锁的代表同时悲观锁中又可以再细分为公平锁非公平锁可重入锁等等。
2. 乐观锁
乐观锁会有一个版本号每次操作数据会对版本号1再提交回数据时会去校验是否比之前的版本大1 如果大1 则进行操作成功这套机制的核心逻辑在于如果在操作过程中版本号只比原来大1 那么就意味着操作过程中没有人对他进行过修改他的操作就是安全的如果不大1则数据被修改过当然乐观锁还有一些变种的处理方式比如cas
乐观锁的典型代表就是cas利用cas进行无锁化机制加锁var5 是操作前读取的内存值while中的var1var2 是预估值如果预估值 内存值则代表中间没有被人修改过此时就将新值去替换内存值。
其中do while 是为了在操作失败时再次进行自旋操作即把之前的逻辑再操作一次。 Unsafe.class
public final long getAndAddLong(Object var1, long var2, long var4) {long var5;do {var6 this.getLongVolatile(var1, var2);} while(!this.compareAndSwapLong(var1, var2, var5, var5 var4));return var5;}课程中的使用方式
课程中的使用方式是没有像cas一样带自旋的操作也没有对version的版本号1 他的操作逻辑是在操作时对版本号进行1 操作然后要求version 如果是1 的情况下才能操作那么第一个线程在操作后数据库中的version变成了2但是他自己满足version1 所以没有问题此时线程2执行线程2 最后也需要加上条件version 1 但是现在由于线程1已经操作过了所以线程2操作时就不满足version1 的条件了所以线程2执行失败。 这里分析其实版本号和stock是异曲同工之妙看stock库存剩余量即可无需添加version的版本信息简化表的修改。
3. 乐观锁解决超卖问题
修改代码方案一、
VoucherOrderServiceImpl 在扣减库存时改为 代码如下VoucherOrderServiceImpl.java // 5.2扣减库存(针对超卖问题用乐观锁CAS解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock ?
boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).eq(stock, stock).update();然后重置数据库数据
UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;重庆程序打开JMeter再测试一下 查看结果树很多不成功 查看报告异常比例也很高 看一下数据库只有21个订单 看一下库存并没有超卖
以上逻辑的核心含义是只要我扣减库存时的库存和之前我查询到的库存是一样的就意味着没有人在中间修改过库存那么此时就是安全的但是以上这种方式通过测试发现会有很多失败的情况失败的原因在于在使用乐观锁过程中假设100个线程同时都拿到了100的库存然后大家一起去进行扣减但是100个人中只有1个人能扣减成功其他的人在处理时他们在扣减时库存已经被修改过了所以此时其他线程都会失败。
修改代码方案二、
之前的方式要修改前后都保持一致但是这样我们分析过成功的概率太低所以我们的乐观锁需要变一下改成stock大于0 即可。
修改VoucherOrderServiceImpl.java // 5.3扣减库存(针对使用乐观锁CAS没卖完解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock 0boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).gt(stock, 0).update();恢复数据库的库存和订单
UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;再启动JMeter,查看吞吐量如下 查看请求 再看下数据库 订单100单无误 库存0无误 知识小扩展
针对cas中的自旋压力过大我们可以使用Longaddr这个类去解决 Java8 提供的一个对AtomicLong改进后的一个类LongAdder 大量线程并发更新一个原子性的时候天然的问题就是自旋会导致并发性问题当然这也比我们直接使用syn来的好 所以利用这么一个类LongAdder来进行优化 如果获取某个值则会对cell和base的值进行递增最后返回一个完整的值。 4.4 秒杀的一人一单限制功能
4.4.1 实现秒杀的一人一单限制
优惠券秒杀一人一单
目前的模式是1个用户可以买多单这样不利于店家的推广。
需求修改秒杀业务要求同一个优惠券一个用户只能下一单。
现在的问题在于
优惠卷是为了引流但是目前的情况是一个人可以无限制的抢这个优惠卷所以我们应当增加一层逻辑让一个用户只能下一个单而不是让一个用户下多个单
具体操作逻辑如下比如时间是否充足如果时间充足则进一步判断库存是否足够然后再根据优惠卷id和用户id查询是否已经下过这个订单如果下过这个订单则不再下单否则进行下单。
初步代码增加一人一单逻辑
修改VoucherOrderServiceImpl 添加逻辑 Long userID UserHolder.getUser().getId();// 5.实现1人1单加入逻辑根据优惠券id和用户id查询订单// 5.1查询订单,并不用查询出具体的值而是查询出数量即可Integer count query().eq(user_id, userID).eq(voucher_id, voucherId).count();// 5.2判断订单是否存在if (count 0) {// 5.2.1 存在就返回异常结果return Result.fail(秒杀优惠券每人限购1张,感谢配合,本优惠券最终解释权归ty公司所有!);}清理数据库 UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;重启应用测试一下 配置一下JMeter,token只配置了1个按理来说可以控制1个用户只下1单 查看结果 结果如下发现同1人还是可以下8单
4.4.2.单机模式下的线程安全问题
存在问题现在的问题还是和之前一样并发过来查询数据库都不存在订单所以我们还是需要加锁但是乐观锁比较适合更新数据而现在是插入数据所以我们需要使用悲观锁操作。
**注意**在这里提到了非常多的问题我们需要慢慢的来思考首先我们的初始方案是封装了一个createVoucherOrder方法同时为了确保他线程安全在方法上添加了一把synchronized 锁。
修改VoucherOrderServiceImpl.java Transactionalpublic synchronized Result createVoucherOrder(Long voucherId) {Long userID UserHolder.getUser().getId();// 5.实现1人1单加入逻辑根据优惠券id和用户id查询订单// 5.1查询订单,并不用查询出具体的值而是查询出数量即可Integer count query().eq(user_id, userID).eq(voucher_id, voucherId).count();// 5.2判断订单是否存在if (count 0) {// 5.2.1 存在就返回异常结果return Result.fail(秒杀优惠券每人限购1张,感谢配合,本优惠券最终解释权归ty公司所有!);}// 5.2.2 不存在再减少库存// 6.1扣减库存(会出现超卖问题)// update tb_seckill_voucher set stock stock -1 where voucher_id ?/*boolean result iSeckillVoucherService.update().setSql(stock stock -1).eq(voucher_id, voucherId).update();*/// 6.2扣减库存(针对超卖问题用乐观锁CAS解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock ?/*boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).eq(stock, stock).update();*/// 6.3扣减库存(针对使用乐观锁CAS没卖完解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock 0boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).gt(stock, 0).update();if (!result) {return Result.fail(商品已经售罄!);}// 7.创建订单/*** 获取订单id*/long orderId redisWorker.nextID(order);VoucherOrder voucherOrder new VoucherOrder();voucherOrder.setId(orderId);voucherOrder.setVoucherId(voucherId);voucherOrder.setUserId(userID);// 将订单信息保存到数据库// insert into tb_voucher_order values ()save(voucherOrder);//8.返回订单idreturn Result.ok(orderId);}但是像这样在方法上添加锁相当于是this锁任何对象进来都会获取到锁锁的粒度太粗了在使用锁过程中控制锁粒度 是一个非常重要的事情因为如果锁的粒度太大会导致每个线程进来都会锁住所以我们需要去控制锁的粒度以下这段代码需要修改为 intern() 这个方法是从常量池中拿到数据如果我们直接使用userId.toString() 他拿到的对象实际上是不同的对象new出来的对象我们使用锁必须保证锁必须是同一把所以我们需要使用intern()方法。 安装Translation插件看中文解释 配置鼠标悬浮
修改VoucherOrderServiceImpl.java Transactionalpublic Result createVoucherOrder(Long voucherId) {Long userID UserHolder.getUser().getId();// 通过悲观锁锁住用户实现一人一单synchronized (userID.toString().intern()) {// 5.实现1人1单加入逻辑根据优惠券id和用户id查询订单// 5.1查询订单,并不用查询出具体的值而是查询出数量即可Integer count query().eq(user_id, userID).eq(voucher_id, voucherId).count();// 5.2判断订单是否存在if (count 0) {// 5.2.1 存在就返回异常结果return Result.fail(秒杀优惠券每人限购1张,感谢配合,本优惠券最终解释权归ty公司所有!);}// 5.2.2 不存在再减少库存// 6.1扣减库存(会出现超卖问题)// update tb_seckill_voucher set stock stock -1 where voucher_id ?/*boolean result iSeckillVoucherService.update().setSql(stock stock -1).eq(voucher_id, voucherId).update();*/// 6.2扣减库存(针对超卖问题用乐观锁CAS解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock ?/*boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).eq(stock, stock).update();*/// 6.3扣减库存(针对使用乐观锁CAS没卖完解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock 0boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).gt(stock, 0).update();if (!result) {return Result.fail(商品已经售罄!);}// 7.创建订单/*** 获取订单id*/long orderId redisWorker.nextID(order);VoucherOrder voucherOrder new VoucherOrder();voucherOrder.setId(orderId);voucherOrder.setVoucherId(voucherId);voucherOrder.setUserId(userID);// 将订单信息保存到数据库// insert into tb_voucher_order values ()save(voucherOrder);//8.返回订单idreturn Result.ok(orderId);}}但是以上代码还是存在问题问题的原因在于当前方法被spring的事务控制如果你在方法内部加锁可能会导致当前方法事务还没有提交但是锁已经释放也会导致问题所以我们选择将当前方法整体包裹起来确保事务不会出现问题如下
在seckillVoucher 方法中添加以下逻辑这样就能保证事务的特性同时也控制了锁的粒度。让对象锁锁住整个方法因为Transactional是方法结束后才提交的。防止事务还没提交前有线程进入方法体内查询。不把synchronized放在方法上是因为放在方法上相当于是this锁任何对象都可以来获取。 // 实现一人一单锁住对象Long userID UserHolder.getUser().getId();synchronized (userID.toString().intern()) {return createVoucherOrder(voucherId);}但是以上做法依然有问题因为你调用的方法其实是this.的方式调用的事务想要生效还得利用代理来生效所以这个地方我们需要获得原始的事务对象 来操作事务。 修改VoucherOrderServiceImpl.java // 实现一人一单获取user对象锁Long userID UserHolder.getUser().getId();synchronized (userID.toString().intern()) {// 调用本类方法的时候Spring事务是失效的解决方案二调用AopContext APIObject o AopContext.currentProxy();IVoucherOrderService proxy (IVoucherOrderService) o;return proxy.createVoucherOrder(voucherId);}修改IVoucherOrderService.java
public interface IVoucherOrderService extends IServiceVoucherOrder {Result seckillVoucher(Long voucherId);Result createVoucherOrder(Long voucherId);
}pom.xml加入依赖 !-- Spring事务失效采用AopContext API来处理 --dependencygroupIdorg.aspectj/groupIdartifactIdaspectjweaver/artifactId/dependency修改HmDianPingApplication.java添加注解 HmDianPingApplication.java
EnableAspectJAutoProxy(exposeProxy true)
清理数据库 UPDATE tb_seckill_voucher set stock 100 where voucher_id 2;delete from tb_voucher_order;重启应用运行JMeter
查看数据库
4.4.3 集群模式下的线程安全问题
通过加锁可以解决在单机情况下的一人一单安全问题但是在集群模式下就不行了。
1、我们将服务启动两份端口分别为8081和8082 给新服务重新命名重新指定一个端口 -Dserver.port8082 点选2个应用点击启动注意用dubug启动
2、然后修改nginx的conf目录下的nginx.conf文件配置反向代理和负载均衡
3.配置完后启动nginx
nginx.exe -s reload4.访问
http://localhost:8080/api/voucher/list/1刷新2-3次8082有查询sql日志输出 8081有查询sql日志输出 这样就模拟多集群负载均衡完毕。
打开数据库恢复数据 UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;delete FROM tb_voucher_order;打开IDEA打上断点 打开PostMan配置配置2个Http请求2个配置完全一样访问路径和header的authorization都一样 路径http://ocalhost:8080/api/voucher-order/seckill/2 post请求 header:authorization 之后点击send测试发现2个端口的服务都进入了断点这明显是有问题的因为二者配置的是同一个用户不应该都获取到锁。与我们的设想有出入
继续跑断点发现2个端口计算的count都是0 全部跑完后发现订单出现了2个库存少了2个又一次出现了1人多卖现象。
具体操作(略)
有关锁失效原因分析
由于现在我们部署了多个tomcat每个tomcat都有一个属于自己的jvm那么假设在服务器A的tomcat内部有两个线程这两个线程由于使用的是同一份代码那么他们的锁对象是同一个是可以实现互斥的但是如果现在是服务器B的tomcat内部又有两个线程但是他们的锁对象写的虽然和服务器A一样但是锁对象却不是同一个所以线程3和线程4可以实现互斥但是却无法和线程1和线程2实现互斥这就是 集群环境下syn锁失效的原因在这种情况下我们就需要使用分布式锁来解决这个问题。
4.4.4 分布式锁
4.4.4.1 分布式锁原理
基本原理和实现方式对比
分布式锁满足分布式系统或集群模式下多进程可见并且互斥的锁。
分布式锁的核心思想就是让大家都使用同一把锁只要大家使用的是同一把锁那么我们就能锁住线程不让线程进行让程序串行执行这就是分布式锁的核心思路。 那么分布式锁他应该满足一些什么样的条件呢
可见性多个线程都能看到相同的结果注意这个地方说的可见性并不是并发编程中指的内存可见性只是说多个进程之间都能感知到变化的意思
互斥互斥是分布式锁的最基本的条件使得程序串行执行
高可用程序不易崩溃时时刻刻都保证较高的可用性
高性能由于加锁本身就让性能降低所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能
安全性安全也是程序中必不可少的一环 常见的分布式锁有三种
Mysqlmysql本身就带有锁机制但是由于mysql性能本身一般所以采用分布式锁的情况下其实使用mysql作为分布式锁比较少见
Redisredis作为分布式锁是非常常见的一种使用方式现在企业级开发中基本都使用redis或者zookeeper作为分布式锁利用setnx这个方法如果插入key成功则表示获得到了锁如果有人插入成功其他人插入失败则表示无法获得到锁利用这套逻辑来实现分布式锁
Zookeeperzookeeper也是企业级开发中较好的一个实现分布式锁的方案由于本套视频并不讲解zookeeper的原理和分布式锁的实现所以不过多阐述。
4.4.4.2 Redis的String结构实现分布式锁
实现分布式锁时需要实现的两个基本方法 获取锁 互斥确保只能有一个线程获取锁 非阻塞尝试一次成功返回true失败返回false 释放锁 手动释放超时释放获取锁时添加一个超时时间 我们发现互斥和设置有效期是2个指令
setnx lock thread1
expire thread1 52个指令有可能只执行了第一条指令第二条执行还没执行的时候redis宕机了。这就需要找一条指令把2件事都干了。
set lock thread2 ex 5 nx 核心思路
我们利用redis 的setNx 方法当有多个线程进入时我们就利用该方法第一个线程进入时redis 中就有这个key 了返回了1如果结果是1则表示他抢到了锁那么他去执行业务然后再删除锁退出锁逻辑没有抢到锁的哥们等待一定时间后重试即可。
4.4.4.2.1 实现分布式锁版本一
加锁逻辑
锁的基本接口 新增接口ILock.java ILock.java
package com.hmdp.lock;/*** InterfaceName: ILock* Description:* Author: wty* Date: 2023/2/16*/public interface ILock {/*** param* return boolean true获取锁成功false获取锁失败* description //获取尝试锁* param: expireTime 锁持有的超时时间过期后自动释放* date 2023/2/16 12:07* author wty**/boolean tryLock(long expireTime);/*** param* return void* description //释放锁* date 2023/2/16 12:08* author wty**/void unLock();
}
SimpleRedisLock
利用setnx方法进行加锁同时增加过期时间防止死锁此方法可以保证加锁和增加过期时间具有原子性。新建类SimpleRedisLock SimpleRedisLock.java
public class SimpleRedisLock implements ILock {/*** 锁前缀*/private static final String KEY_PREFIX lock:;/*** 定义锁的名称*/private String lockName;StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String lockName, StringRedisTemplate stringRedisTemplate) {this.lockName lockName;this.stringRedisTemplate stringRedisTemplate;}Overridepublic boolean tryLock(long expireTime) {// 获取当前线程的标识long threadId Thread.currentThread().getId();// set key 1 ex expireTime nxBoolean flag stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX lockName, threadId , expireTime, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);}Overridepublic void unLock() {stringRedisTemplate.delete(KEY_PREFIX lockName);}
}
释放锁逻辑
SimpleRedisLock
释放锁防止删除别人的锁 Overridepublic void unLock() {stringRedisTemplate.delete(KEY_PREFIX lockName);}修改业务代码VoucherOrderServiceImpl.java
Service
public class VoucherOrderServiceImpl extends ServiceImplVoucherOrderMapper, VoucherOrder implements IVoucherOrderService {Autowiredprivate ISeckillVoucherService iSeckillVoucherService;Autowiredprivate RedisWorker redisWorker;ResourceStringRedisTemplate stringRedisTemplate;/*** param* return com.hmdp.dto.Result* description //秒杀优惠券* param: voucherId* date 2023/2/15 22:09* author wty**/Overridepublic Result seckillVoucher(Long voucherId) {// 1.查询秒杀优惠券信息// select * from tb_seckill_voucher where voucher_id ?SeckillVoucher seckillVoucher iSeckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始LocalDateTime beginTime seckillVoucher.getBeginTime();LocalDateTime endTime seckillVoucher.getEndTime();LocalDateTime now LocalDateTime.now();if (now.isBefore(beginTime)) {// 当前时间早于秒杀开始时间说明秒杀没有开始return Result.fail(秒杀尚未开始,请耐心等待秒杀开始时间: beginTime.format(DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss)));}// 3.判断秒杀是否已经结束if (now.isAfter(endTime)) {// 当前时间晚于秒杀结束时间说明秒杀结束了return Result.fail(秒杀已经结束,感谢支持!);}// 4.判断库存是否充足Integer stock seckillVoucher.getStock();if (stock 0) {return Result.fail(商品已经售罄!);}// 实现一人一单获取user对象锁Long userID UserHolder.getUser().getId();/*// 使用JDK提供的锁监视器synchronized来实现synchronized (userID.toString().intern()) {// 调用本类方法的时候Spring事务是失效的解决方案二调用AopContext APIObject o AopContext.currentProxy();IVoucherOrderService proxy (IVoucherOrderService) o;return proxy.createVoucherOrder(voucherId);}*/// 尝试自定义锁监视器SimpleRedisLock simpleRedisLock new SimpleRedisLock(order: userID.toString().intern(), stringRedisTemplate);boolean flag simpleRedisLock.tryLock(RedisConstants.LOCK_VOUVHER_ORDER_TTL);if (!flag) {// 获取锁失败,就直接返回错误信息即可return Result.fail([秒杀优惠券]不允许重复下单!本秒杀业务一切解释器归ty公司所有);}Result result;try {Object o AopContext.currentProxy();IVoucherOrderService proxy (IVoucherOrderService) o;return proxy.createVoucherOrder(voucherId);} finally {simpleRedisLock.unLock();}}/*** param* return com.hmdp.dto.Result* description //根据优惠券id和用户id查询订单 减少库存生成订单* param: voucherId* date 2023/2/15 22:12* author wty**/OverrideTransactionalpublic Result createVoucherOrder(Long voucherId) {Long userID UserHolder.getUser().getId();// 5.实现1人1单加入逻辑根据优惠券id和用户id查询订单// 5.1查询订单,并不用查询出具体的值而是查询出数量即可Integer count query().eq(user_id, userID).eq(voucher_id, voucherId).count();// 5.2判断订单是否存在if (count 0) {// 5.2.1 存在就返回异常结果return Result.fail(秒杀优惠券每人限购1张,感谢配合,本优惠券最终解释权归ty公司所有!);}// 5.2.2 不存在再减少库存// 6.1扣减库存(会出现超卖问题)// update tb_seckill_voucher set stock stock -1 where voucher_id ?/*boolean result iSeckillVoucherService.update().setSql(stock stock -1).eq(voucher_id, voucherId).update();*/// 6.2扣减库存(针对超卖问题用乐观锁CAS解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock ?/*boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).eq(stock, stock).update();*/// 6.3扣减库存(针对使用乐观锁CAS没卖完解决)// update tb_seckill_voucher set stock stock -1 where voucher_id ? and stock 0boolean result iSeckillVoucherService.update().setSql(stock stock - 1).eq(voucher_id, voucherId).gt(stock, 0).update();if (!result) {return Result.fail(商品已经售罄!);}// 7.创建订单/*** 获取订单id*/long orderId redisWorker.nextID(order);VoucherOrder voucherOrder new VoucherOrder();voucherOrder.setId(orderId);voucherOrder.setVoucherId(voucherId);voucherOrder.setUserId(userID);// 将订单信息保存到数据库// insert into tb_voucher_order values ()save(voucherOrder);//8.返回订单idreturn Result.ok(orderId);}
}修改RedisConstants.java
public static final Long LOCK_VOUVHER_ORDER_TTL 1200L; // TODO 后续改成5L恢复数据库数据 UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;delete FROM tb_voucher_order;Debug模式重启2个应用 发送PostMan请求 看一下IDEA中的断点发现一个是true获取到锁另一个是false未获取到锁。 查看数据库发现库存减少1符合 发现订单产生1条记录符合
4.4.4.3 锁误删问题
1.Redis分布式锁误删情况说明
逻辑说明
持有锁的线程在锁的内部出现了阻塞导致他的锁自动释放这时其他线程线程2来尝试获得锁就拿到了这把锁然后线程2在持有锁执行过程中线程1反应过来继续执行而线程1执行过程中走到了删除锁逻辑此时就会把本应该属于线程2的锁进行删除这就是误删别人锁的情况说明。
解决方案解决方案就是在每个线程释放锁的时候去判断一下当前这把锁是否属于自己如果属于自己则不进行锁的删除假设还是上边的情况线程1卡顿锁自动释放线程2进入到锁的内部执行逻辑此时线程1反应过来然后删除锁但是线程1一看当前这把锁不是属于自己于是不进行删除锁逻辑当线程2走到删除锁逻辑时如果没有卡过自动释放锁的时间点则判断当前这把锁是属于自己的于是删除这把锁。
2. 解决Redis分布式锁误删问题
需求修改之前的分布式锁实现满足
在获取锁时存入线程标示可以用UUID表示在释放锁时先获取锁中的线程标示判断是否与当前线程标示一致
如果一致则释放锁如果不一致则不释放锁
核心逻辑在存入锁时放入自己线程的标识在删除锁时判断当前这把锁的标识是不是自己存入的如果是则进行删除如果不是则不进行删除。
具体代码如下加锁和释放锁 通过UUID和线程id区分每个线程。 修改SimpleRedisLock.java
public class SimpleRedisLock implements ILock {/*** 锁前缀*/private static final String KEY_PREFIX lock:;/*** 线程ID前缀* true是可以把UUID中的横线去掉*/private static final String THREAD_ID_PREFIX UUID.randomUUID().toString(true) -;/*** 定义锁的名称*/private String lockName;StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String lockName, StringRedisTemplate stringRedisTemplate) {this.lockName lockName;this.stringRedisTemplate stringRedisTemplate;}/*** param* return boolean* description // 获取锁* param: expireTime* date 2023/2/16 12:24* author wty**/Overridepublic boolean tryLock(long expireTime) {// 获取当前线程的标识//long threadId Thread.currentThread().getId();String threadId THREAD_ID_PREFIX Thread.currentThread().getId();// set key 1 ex expireTime nxBoolean flag stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX lockName, threadId , expireTime, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);}/*** param* return void* description //释放锁* date 2023/2/16 12:24* author wty**/Overridepublic void unLock() {// 获取当前线程的IDString threadId THREAD_ID_PREFIX Thread.currentThread().getId();String threadId_Redis stringRedisTemplate.opsForValue().get(KEY_PREFIX lockName);// 只有redis中的线程和当前的线程是同一个才允许释放锁if (String.valueOf(threadId).equals(threadId_Redis)) {stringRedisTemplate.delete(KEY_PREFIX lockName);}}
}
修改数据库sql UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;重启应用用PostMan发送请求 之后允许断点通过让第一个端口能成功获取锁 此时模拟业务阻塞数据有效期过期但是业务还是没有办理完。我们可以人为手动删除掉redis中的锁。 可以看到value的形式满足UUID-线程id的形式 删除后紧接着我们让端口2来获取锁 然后我们跟端口1看看能否释放锁发现是不能的 我们跟端口2发现它可以释放锁 查看数据库
解决了误删的操作。
有关代码实操说明
在我们修改完此处代码后我们重启工程然后启动两个线程第一个线程持有锁后手动释放锁第二个线程 此时进入到锁内部再放行第一个线程此时第一个线程由于锁的value值并非是自己所以不能释放锁也就无法删除别人的锁此时第二个线程能够正确释放锁通过这个案例初步说明我们解决了锁误删的问题。
4.4.4.4 分布式锁的原子性操作问题
更为极端的误删逻辑说明
线程1现在持有锁之后在执行业务逻辑过程中他正准备删除锁而且已经走到了条件判断的过程中比如他已经拿到了当前这把锁确实是属于他自己的正准备删除锁但是此时他的锁到期了那么此时线程2进来但是线程1他会接着往后执行当他卡顿结束后他直接就会执行删除锁那行代码相当于条件判断并没有起到作用这就是删锁时的原子性问题之所以有这个问题是因为线程1的拿锁比锁删锁实际上并不是原子性的我们要防止刚才的情况发生。
4.4.4.5 Lua脚本解决原子性问题
Redis提供了Lua脚本功能在一个脚本中编写多条Redis命令确保多条命令执行时的原子性。Lua是一种编程语言它的基本语法大家可以参考网站Lua官网。 这里重点介绍Redis提供的调用函数我们可以使用lua去操作redis又能保证他的原子性这样就可以实现拿锁比锁删锁是一个原子性动作了作为Java程序员这一块并不作一个简单要求并不需要大家过于精通只需要知道他有什么作用即可。
这里重点介绍Redis提供的调用函数语法如下
redis.call(命令名称, key, 其它参数, ...)例如我们要执行set name jack则脚本是这样
# 执行 set name jack
redis.call(set, name, jack)例如我们要先执行set name Rose再执行get name则脚本如下
# 先执行 set name jack
redis.call(set, name, Rose)
# 再执行 get name
local name redis.call(get, name)
# 返回
return name写好脚本以后需要用Redis命令来调用脚本调用脚本的常见命令如下 例如我们要执行 redis.call(‘set’, ‘name’, ‘jack’) 这个脚本语法如下 这里最后的0是指key类型的参数比如这个脚本里没有参数都是常量设置好的就是0
EVAL return redis.call(set,name,jack) 0自己试一下
如果脚本中的key、value不想写死可以作为参数传递。key类型参数会放入KEYS数组其它参数会放入ARGV数组在脚本中可以从KEYS和ARGV数组获取这些参数 Lua脚本中数组是从1开始的 我们自己玩一下指令
接下来我们来回顾一下我们释放锁的逻辑
释放锁的业务流程是这样的
1、获取锁中的线程标示
2、判断是否与指定的标示当前线程标示一致
3、如果一致则释放锁删除
4、如果不一致则什么都不做
如果用Lua脚本来表示则是这样的
最终我们操作redis的拿锁比锁删锁的lua脚本就会变成这样 Lua脚本如下
-- 锁的key
local key KEYS[1]-- 当前线程的标识
local threadID ARGV[1]-- 获取锁中的线程标识
local threadID_Redis redis.call(get,KEYS[1])-- 比较线程的标识与锁中的标识是否一致
if(ARGV[1] threadID_Redis) then-- 一致就释放锁redis.call(del,KEYS[1])
end
return 01. 利用Java代码调用Lua脚本改造分布式锁
lua脚本本身并不需要大家花费太多时间去研究只需要知道如何调用大致是什么意思即可所以在笔记中并不会详细的去解释这些lua表达式的含义。
我们的RedisTemplate中可以利用execute方法去执行lua脚本参数对应关系就如下图股 下载插件EmmyLua 下载完插件之后新建Lua脚本 把上面的脚本拷贝进去即可
Java代码
修改SimpleRedisLock.java增加
public class SimpleRedisLock implements ILock {/*** 锁前缀*/private static final String KEY_PREFIX lock:;/*** 线程ID前缀* true是可以把UUID中的横线去掉*/private static final String THREAD_ID_PREFIX UUID.randomUUID().toString(true) -;/*** 定义锁的名称*/private String lockName;StringRedisTemplate stringRedisTemplate;private static final DefaultRedisScriptLong UNLOCK_SCRIPT;static {UNLOCK_SCRIPT new DefaultRedisScript();UNLOCK_SCRIPT.setLocation(new ClassPathResource(unlock.lua));UNLOCK_SCRIPT.setResultType(Long.class);}public SimpleRedisLock(String lockName, StringRedisTemplate stringRedisTemplate) {this.lockName lockName;this.stringRedisTemplate stringRedisTemplate;}/*** param* return void* description //释放锁(基于Lua脚本)* date 2023/2/16 15:37* author wty**/Overridepublic void unLock() {// 调用Lua脚本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX lockName),THREAD_ID_PREFIX Thread.currentThread().getId());}
}
经过以上代码改造后我们就能够实现 拿锁、锁删锁的原子性动作了~ 下面测试一下。 恢复数据库 UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;重启应用Debug模式 打开PostMan2个Http请求分别点击send 第一个端口跑到获取锁的断点 模拟业务阻塞有效期超时去redis中删除lock 让端口2跑完获取锁逻辑重新生成新的锁
此时切换让端口1跑到释放锁的位置模拟减少库存生成订单但是还没有释放锁。此时看一下Redis数据库说明还没释放锁 再让端口2跑完所有断点redis中lock的那一条删除了这样保证了释放锁的原子性。
小总结
基于Redis的分布式锁实现思路
利用set nx ex获取锁并设置过期时间保存线程标示释放锁时先判断线程标示是否与自己一致一致则删除锁 特性 利用set nx满足互斥性利用set ex保证故障时锁依然能释放避免死锁提高安全性利用Redis集群保证高可用和高并发特性
笔者总结我们一路走来利用添加过期时间防止死锁问题的发生但是有了过期时间之后可能出现误删别人锁的问题这个问题我们开始是利用删之前 通过拿锁比锁删锁这个逻辑来解决的也就是删之前判断一下当前这把锁是否是属于自己的但是现在还有原子性问题也就是我们没法保证拿锁比锁删锁是一个原子性的动作最后通过lua表达式来解决这个问题
但是目前还剩下一个问题锁不住什么是锁不住呢你想一想如果当过期时间到了之后我们可以给他续期一下比如续个30s就好像是网吧上网 网费到了之后然后说来网管再给我来10块的是不是后边的问题都不会发生了那么续期问题怎么解决呢可以依赖于我们接下来要学习redission啦
测试逻辑
第一个线程进来得到了锁手动删除锁模拟锁超时了其他线程会执行lua来抢锁当第一天线程利用lua删除锁时lua能保证他不能删除他的锁第二个线程删除锁时利用lua同样可以保证不会删除别人的锁同时还能保证原子性。
4.4.4.6 Redission分布式锁
1.分布式锁 - redission功能介绍
基于setnx实现的分布式锁存在下面的问题
重入问题重入问题是指 获得锁的线程可以再次进入到相同的锁的代码块中可重入锁的意义在于防止死锁比如HashTable这样的代码中他的方法都是使用synchronized修饰的假如他在一个方法内调用另一个方法那么此时如果是不可重入的不就死锁了吗所以可重入锁他的主要意义是防止死锁我们的synchronized和Lock锁都是可重入的。
不可重试是指目前的分布式只能尝试一次我们认为合理的情况是当线程在获得锁失败后他应该能再次尝试获得锁。
**超时释放**我们在加锁时增加了过期时间这样的我们可以防止死锁但是如果卡顿的时间超长虽然我们采用了lua表达式防止删锁的时候误删别人的锁但是毕竟没有锁住有安全隐患
主从一致性 如果Redis提供了主从集群当我们向集群写数据时主机需要异步的将数据同步给从机而万一在同步过去之前主机宕机了就会出现死锁问题。 那么什么是Redission呢?
Redisson是一个在Redis的基础上实现的Java驻内存数据网格In-Memory Data Grid。它不仅提供了一系列的分布式的Java常用对象还提供了许多分布式服务其中就包含了各种分布式锁的实现。
Redission提供了分布式锁的多种多样的功能 官网网站Redisson官网 github地址
2. 分布式锁-Redission快速入门
pom.xml中引入依赖
dependencygroupIdorg.redisson/groupIdartifactIdredisson/artifactIdversion3.13.6/version
/dependency配置Redisson客户端 RedissonConfig代码如下
Configuration
public class RedissonConfig {Beanpublic RedissonClient redissonClient(){// 配置类Config config new Config();// 添加redis地址这里添加了单点的地址(虚拟机地址)也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress(redis://192.168.150.101:6379).setPassword(123321);// 创建RedissonClient对象return Redisson.create(config);}
}
如何使用Redission的分布式锁
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 修改VoucherOrderServiceImpl.java
Service
public class VoucherOrderServiceImpl extends ServiceImplVoucherOrderMapper, VoucherOrder implements IVoucherOrderService {Autowiredprivate ISeckillVoucherService iSeckillVoucherService;Autowiredprivate RedisWorker redisWorker;ResourceStringRedisTemplate stringRedisTemplate;Resourceprivate RedissonClient redissonClient;Overridepublic Result seckillVoucher(Long voucherId) {// 1.查询秒杀优惠券信息// select * from tb_seckill_voucher where voucher_id ?SeckillVoucher seckillVoucher iSeckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始LocalDateTime beginTime seckillVoucher.getBeginTime();LocalDateTime endTime seckillVoucher.getEndTime();LocalDateTime now LocalDateTime.now();if (now.isBefore(beginTime)) {// 当前时间早于秒杀开始时间说明秒杀没有开始return Result.fail(秒杀尚未开始,请耐心等待秒杀开始时间: beginTime.format(DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss)));}// 3.判断秒杀是否已经结束if (now.isAfter(endTime)) {// 当前时间晚于秒杀结束时间说明秒杀结束了return Result.fail(秒杀已经结束,感谢支持!);}// 4.判断库存是否充足Integer stock seckillVoucher.getStock();if (stock 0) {return Result.fail(商品已经售罄!);}// 实现一人一单获取user对象锁Long userID UserHolder.getUser().getId();// 用Redisson提供的可重入锁RLock lock redissonClient.getLock(lock:order: userID.toString().intern());boolean flag lock.tryLock();if (!flag) {// 获取锁失败,就直接返回错误信息即可return Result.fail([秒杀优惠券]不允许重复下单!本秒杀业务一切解释器归ty公司所有);}try {Object o AopContext.currentProxy();IVoucherOrderService proxy (IVoucherOrderService) o;return proxy.createVoucherOrder(voucherId);} finally {lock.unlock();}}}
恢复数据库 UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;依然重启2个端口用其中1个PostMan发送请求即可。 查看数据库 再恢复数据库测试并发。 UPDATE tb_seckill_voucher SET stock 100 WHERE voucher_id 2;DELETE FROM tb_voucher_order;用JMeter测试一下发现只允许成功1个 数据库结果
4.4.4.7 Hash结构解决锁的可重入问题
分布式锁-redission可重入锁原理
在Lock锁中他是借助于底层的一个voaltile的一个state变量来记录重入的状态的比如当前没有人持有这把锁那么state0假如有人持有这把锁那么state1如果持有这把锁的人再次持有这把锁那么state就会1 如果是对于synchronized而言他在c语言代码中会有一个count原理和state类似也是重入一次就加一释放一次就-1 直到减少成0 时表示当前这把锁没有被人持有。
在redission中我们的也支持支持可重入锁。
我们来模拟一下这个过程 新建测试类RedissonTest.java
Slf4j
SpringBootTest
public class RedissonTest {Resourceprivate RedissonClient redissonClient;RLock lock;BeforeEachvoid setUp() {lock redissonClient.getLock(lock);}Testvoid method1() {boolean isLock lock.tryLock();if (!isLock) {log.error(获取锁失败1);return;}try {log.info(获取锁成功,1);method2();} finally {log.info(释放锁,1);lock.unlock();}}Testvoid method2() {boolean isLock lock.tryLock();if (!isLock) {log.error(获取锁失败2);return;}try {log.info(获取锁成功,2);} finally {log.info(释放锁,2);lock.unlock();}}
}
看Redis的图形界面method2获取锁的时候 method2释放锁的时候 method1释放锁后lock移除。
在分布式锁中他采用hash结构用来存储锁其中大key表示表示这把锁是否存在用小key表示当前这把锁被哪个线程持有所以接下来我们一起分析一下当前的这个lua表达式
这个地方一共有3个参数
KEYS[1] 锁名称
ARGV[1] 锁失效时间
ARGV[2] id “:” threadId; 锁的小key
exists: 判断数据是否存在 name是lock是否存在,如果0就表示当前这把锁不存在
redis.call(‘hset’, KEYS[1], ARGV[2], 1);此时他就开始往redis里边去写数据 写成一个hash结构
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]);我们跟一下源码看一下 以上是tryLock的源码。接下来看unLock的 4.4.4.8 watchDog解决锁超时释放问题
1.分布式锁-redission锁重试和WatchDog机制
说明由于课程中已经说明了有关tryLock的源码解析以及其看门狗原理所以笔者在这里给大家分析lock()方法的源码解析希望大家在学习过程中能够掌握更多的知识
抢锁过程中获得当前线程通过tryAcquire进行抢锁该抢锁逻辑和之前逻辑相同
1、先判断当前这把锁是否存在如果不存在插入一把锁返回null
2、判断当前这把锁是否是属于当前线程如果是则返回null
所以如果返回是null则代表着当前这哥们已经抢锁完毕或者可重入完毕但是如果以上两个条件都不满足则进入到第三个条件返回的是锁的失效时间同学们可以自行往下翻一点点你能发现有个while( true) 再次进行tryAcquire进行抢锁
long threadId Thread.currentThread().getId();
Long ttl tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl null) {return;
}接下来会有一个条件分支因为lock方法有重载方法一个是带参数一个是不带参数如果带带参数传入的值是-1如果传入参数则leaseTime是他本身所以如果传入了参数此时leaseTime ! -1 则会进去抢锁抢锁的逻辑就是之前说的那三个逻辑
if (leaseTime ! -1) {return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}如果是没有传入时间则此时也会进行抢锁 而且抢锁时间是默认看门狗时间 commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()
ttlRemainingFuture.onComplete((ttlRemaining, e) 这句话相当于对以上抢锁进行了监听也就是说当上边抢锁完毕后此方法会被调用具体调用的逻辑就是去后台开启一个线程进行续约逻辑也就是看门狗线程
RFutureLong ttlRemainingFuture tryLockInnerAsync(waitTime,commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) - {if (e ! null) {return;}// lock acquiredif (ttlRemaining null) {scheduleExpirationRenewal(threadId);}
});
return ttlRemainingFuture;此逻辑就是续约逻辑注意看commandExecutor.getConnectionManager().newTimeout 此方法
Method( new TimerTask() {},参数2 参数3 )
指的是通过参数2参数3 去描述什么时候去做参数1的事情现在的情况是10s之后去做参数一的事情
因为锁的失效时间是30s当10s之后此时这个timeTask 就触发了他就去进行续约把当前这把锁续约成30s如果操作成功那么此时就会递归调用自己再重新设置一个timeTask()于是再过10s后又再设置一个timerTask完成不停的续约
那么大家可以想一想假设我们的线程出现了宕机他还会续约吗当然不会因为没有人再去调用renewExpiration这个方法所以等到时间之后自然就释放了。
private void renewExpiration() {ExpirationEntry ee EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee null) {return;}Timeout task commandExecutor.getConnectionManager().newTimeout(new TimerTask() {Overridepublic 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;}RFutureBoolean future renewExpirationAsync(threadId);future.onComplete((res, e) - {if (e ! null) {log.error(Cant update lock getName() expiration, e);return;}if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}2. 分布式锁-redission锁的MutiLock原理
Redisson分布式锁主从一致性问题 为了提高redis的可用性我们会搭建集群或者主从现在以主从为例 此时我们去写命令写在主机上 主机会将数据同步给从机但是假设在主机还没有来得及把数据写入到从机去的时候此时主机宕机哨兵会发现主机宕机并且选举一个slave变成master而此时新的master中实际上并没有锁信息此时锁信息就已经丢掉了。 为了解决这个问题redission提出来了MutiLock锁使用这把锁咱们就不使用主从了每个节点的地位都是一样的 这把锁加锁的逻辑需要写入到每一个主丛节点上只有所有的服务器都写入成功此时才是加锁成功假设现在某个节点挂了那么他去获得锁的时候只要有一个节点拿不到都不能算是加锁成功就保证了加锁的可靠性。
那么MutiLock 加锁原理是什么呢笔者画了一幅图来说明
当我们去设置了多个锁时redission会将多个锁添加到一个集合中然后用while循环去不停去尝试拿锁但是会有一个总共的加锁时间这个时间是用需要加锁的个数 * 1500ms 假设有3个锁那么时间就是4500ms假设在这4500ms内所有的锁都加锁成功 那么此时才算是加锁成功如果在4500ms有线程加锁失败则会再次去进行重试. 实现连锁机制 修改RedissonConfig.java
Configuration
public class RedissonConfig {Beanpublic RedissonClient redissonClient() {// 配置类Config config new Config();// 添加redis地址这里添加了单点的地址也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress(redis://192.168.183.145:6379).setPassword(112453);// 创建RedissonClient对象return Redisson.create(config);}Beanpublic RedissonClient redissonClient2() {// 配置类Config config new Config();// 添加redis地址这里添加了单点的地址也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress(redis://192.168.193.175:6380).setPassword(557724);// 创建RedissonClient对象return Redisson.create(config);}Beanpublic RedissonClient redissonClient3() {// 配置类Config config new Config();// 添加redis地址这里添加了单点的地址也可以使用config.useClusterServers()添加集群地址config.useSingleServer().setAddress(redis://192.168.177.145:6381).setPassword(5896);// 创建RedissonClient对象return Redisson.create(config);}
}
修改RedissonTest.java
Slf4j
SpringBootTest
public class RedissonTest {Resourceprivate RedissonClient redissonClient;Resourceprivate RedissonClient redissonClient2;Resourceprivate RedissonClient redissonClient3;RLock lock;BeforeEachvoid setUp() {RLock lock1 redissonClient.getLock(lock);RLock lock2 redissonClient.getLock(lock);RLock lock3 redissonClient.getLock(lock);// 创建连锁lock redissonClient.getMultiLock(lock1, lock2, lock3);}Testvoid method1() {boolean isLock lock.tryLock();if (!isLock) {log.error(获取锁失败1);return;}try {log.info(获取锁成功,1);method2();} finally {log.info(释放锁,1);lock.unlock();}}Testvoid method2() {boolean isLock lock.tryLock();if (!isLock) {log.error(获取锁失败2);return;}try {log.info(获取锁成功,2);} finally {log.info(释放锁,2);lock.unlock();}}}
