秒杀
超卖问题
如下,我们先来复现问题,抢购秒杀券的代码逻辑也是很简单,
先判断优惠券是否开始了,是的化,判断库存是否充足,如果是的化,扣减库存,最后创建订单
如下是代码
@Override
@Transactional
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券
SeckillVoucher voucher = seckillVoucher.getById(voucherId);
if(voucher == null) {
return Result.fail("优惠券不存在");
}
//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("库存不足!");
}
//5.扣减库存
boolean isSuccess = seckillVoucher.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).update();
//6.判断是否成功
if(!isSuccess) {
return Result.fail("扣减库存失败!");
}
//7.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//7.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//7.2用户id
Long userId = UserHolder.getUser().getId();
voucherOrder.setUserId(userId);
//7.3代金券id
voucherOrder.setVoucherId(voucherId);
//7.4保存到voucher_order表中
save(voucherOrder);
//8.返回订单id
return Result.ok(orderId);
}
问题出现如下代码
我们判断是否充足的时候,有可能很多线程进来刚好都通过了,就会有问题
测试
设置jmeter
这里是加上token头,因为我的系统写了token头,才能通过,你要是没有的化,就不用
测试结果
如此就是超卖了
乐观锁 & 悲观锁
理解乐观锁 & 悲观锁
乐观锁有可能你还不是很懂,但是你一定要知道这个,乐观锁,实际上没有加锁,主打的就是一个乐观,如果发现没有问题,就不上锁,如果有问题,就通过特殊的手段保证线程的安全,这里的特殊的手段一般来说,就是类似于cas这样的,使用一个标识来判断是否有线程安全问题
悲观锁就很好理解了,就是平常我们加的粒度很大的锁,例如synchronized
乐观锁的思想
乐观锁的实操都是一个统一的思想,就是cas,比较 + 交换
比较的是什么,得到的旧值 和 我们再一次得到的值(理解为新值) 判断是否是一致的,如果不是一致的,那么就代表着有线程安全问题
我们再来理解一下这里的比较的意思,为什么要比较我们得到的值,举个例子
一开始我们拿到 stock = 100
然后过了几s,我们再去获取stock,发现stock = 98
是不是就说明这里的stock被人用过了,那么就有线程安全问题!此时我们就退出,或者人为再去加锁,都是可以的,一般来说,乐观锁不会直接加锁
我们再来想一个问题,为什么要有乐观锁???
我直接加锁不好吗?? 为的是两个字 性能!!!
我们一旦加了大粒度的锁,就会消耗性能,在那等吗,当然消耗了,所以就有了乐观锁的存在,它实际上是没有锁的,所以性能当然高!!!
乐观锁的缺点
那难道说,乐观锁,就那么好,没什么缺点? 肯定是有的, 会有完成率的问题
完成率不高,甚至于说,本来200 人抢100张优惠券的问题,但是由于设置的乐观锁, 再高并发下,很容易很多的线程都没有抢到,这种问题,在我这里也出现了, 解决办法就是改变比较条件就行,实例请看下面
乐观锁解决超卖
想我这里就是,简单的cas,判断是否是刚刚的库存
乐观锁完成率不高问题
我们这里更改了条件,只要库存 > 0的化,就可以成功!
这里为什么可以保证原子性,我觉得需要特别说明一下
我们请求打到数据库的时候那个时间点 有条件 stock > 0
因为有事务的原因,mysql这里的写操作是线程安全的,所以这里不会有问题
一人一单问题
一人一单问题,也是可能会有线程安全问题
我们先来看流程图
再超卖问题解决之下,去判断是否已经下过一单了,是的化,就不去下单
代码如下
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券
SeckillVoucher voucher = seckillVoucher.getById(voucherId);
if (voucher == null) {
return Result.fail("优惠券不存在");
}
//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();
//5.一人一单
int count = query().eq("user_id", userId)
.eq("voucher_id", voucherId)
.count();
if(count > 0) {
return Result.fail("你已经买过了");
}
//6.扣减库存
boolean isSuccess = seckillVoucher
.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock",0)
.update();
//7.判断是否成功
if (!isSuccess) {
return Result.fail("扣减库存失败!");
}
//8.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//8.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//8.2用户id
voucherOrder.setUserId(userId);
//8.3代金券id
voucherOrder.setVoucherId(voucherId);
//8.4保存到voucher_order表中
save(voucherOrder);
//9.返回订单id
return Result.ok(orderId);
}
问题处在这
如果高并发的情况下,就有可能会有问题
复现线程安全问题
jmeter设置: 和超卖问题的复现jmeter设置是一致的
原先订单数100
抢购17号优惠券,正常来说,一个用户只能抢1张
测试结果
下了10单
这里就是一人一单出了线程安全问题!
加锁解决
@Autowired
private SeckillVoucherServiceImpl seckillVoucher;
@Resource
private RedisIdWorker redisIdWorker;
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券
SeckillVoucher voucher = seckillVoucher.getById(voucherId);
if (voucher == null) {
return Result.fail("优惠券不存在");
}
//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("库存不足!");
}
return createVoucherOrder(voucherId);
}
@Transactional
public synchronized Result createVoucherOrder(Long voucherId) {
Long userId = UserHolder.getUser().getId();
//5.一人一单
int count = query()
.eq("user_id", userId)
.eq("voucher_id", voucherId)
.count();
System.out.println("此时count为" + count);
if (count > 0) {
return Result.fail("用户已经购买过一次");
}
//6.扣减库存
boolean isSuccess = seckillVoucher
.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock", 0)
.update();
//7.判断是否成功
if (!isSuccess) {
return Result.fail("扣减库存失败!");
}
//8.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//8.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//8.2用户id
voucherOrder.setUserId(userId);
//8.3代金券id
voucherOrder.setVoucherId(voucherId);
//8.4保存到voucher_order表中
save(voucherOrder);
//9.返回订单id
return Result.ok(orderId);
}
再整个方法上加锁,这样确实是万无一失
测试
结果是正确的
优化加锁
如果直接再方法上加锁的化,那么锁的是类对象,也就是这里的service类对象,那么单用户情况下就没问题,但是在多用户情况下就会有问题,因为这里的锁是service类,那么相当于锁的是全部人,也就是说,别的用户还得等你抢完了才能枪,所以这里的锁的粒度有问题,应该锁的是对应的用户而不是所有用户!!!
@Transactional
public Result createVoucherOrder(Long voucherId) {
//只锁住相同用户,所以这里用userId
Long userId = UserHolder.getUser().getId();
//这里是更细粒度的锁,这里不能直接用Long userId来锁,因为有可能是同一个对象,jvm知识
//所以这里用字符串对象,但是Long的toString()里边也是new String(),所以这里要intern()
//避免相同的用户却有着不同的锁,再字符串池里边找到我们那个唯一的用户string
synchronized (userId.toString().intern()) {
//5.一人一单
int count = query()
.eq("user_id", userId)
.eq("voucher_id", voucherId)
.count();
System.out.println("此时count为" + count);
if (count > 0) {
return Result.fail("用户已经购买过一次");
}
//6.扣减库存
boolean isSuccess = seckillVoucher
.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock", 0)
.update();
//7.判断是否成功
if (!isSuccess) {
return Result.fail("扣减库存失败!");
}
//8.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//8.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//8.2用户id
voucherOrder.setUserId(userId);
//8.3代金券id
voucherOrder.setVoucherId(voucherId);
//8.4保存到voucher_order表中
save(voucherOrder);
//9.返回订单id
return Result.ok(orderId);
}
}
这样子锁的才是用户,多人抢的化,就不会相互干涉
事务失效
提到这个我不得不说,这个问题比较难理解,这里的问题相关springboot中的事务
我们来看这里的代码
在这个方法上我们加上了事务 @Transactional 也就是springboot事务处理
,这个注解默认什么都不写的情况下,事务的隔离级别是数据库的隔离级别,而我这里的数据库是mysql,也就是读已提交
什么是读已提交,也就是说,只能读到已经提交的事务,那些没有提交的事务,别的事务是看不到的
这个隔离级别就是为了解决脏读 + 脏写的问题来着,但是反而在这里会出现问题
我门来看这里的流程
- 事务开始
- 上锁
- 业务代码
- 释放锁
- 事务结束
因为这里的锁是嵌套在这个方法里边的,并不是方法上的,所以说,我们释放锁的时候,事务不一定结束!! 换种方法说,就是事务没有提交!
这个问题很关键! 你事务没有提交,意思是别人根本读不到你这里的已经下了单的order,并且你还已经释放锁了,所以别的线程进来,就可以又来下单
所以总的来说,你看这个代码,这个问题的出现就是那么一瞬间的事,但是还是有可能会出现问题的
我们总结一下,为什么会出现这个问题,就是因为释放锁 和 事务的提交不同步,先释放锁了,才去提交事务,这样别人就有可乘之机,所以我们的解决方法就是先去提交事务,再去释放锁
那么我门的锁,就应该锁的是这整个方法了
代码
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券
SeckillVoucher voucher = seckillVoucher.getById(voucherId);
if (voucher == null) {
return Result.fail("优惠券不存在");
}
//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();
synchronized (userId.toString().intern()) {
return createVoucherOrder(voucherId);
}
}
@Transactional
public Result createVoucherOrder(Long voucherId) {
Long userId = UserHolder.getUser().getId();
//5.一人一单
int count = query()
.eq("user_id", userId)
.eq("voucher_id", voucherId)
.count();
System.out.println("此时count为" + count);
if (count > 0) {
return Result.fail("用户已经购买过一次");
}
//6.扣减库存
boolean isSuccess = seckillVoucher
.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock", 0)
.update();
//7.判断是否成功
if (!isSuccess) {
return Result.fail("扣减库存失败!");
}
//8.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//8.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//8.2用户id
voucherOrder.setUserId(userId);
//8.3代金券id
voucherOrder.setVoucherId(voucherId);
//8.4保存到voucher_order表中
save(voucherOrder);
//9.返回订单id
return Result.ok(orderId);
}
这里还有一个问题,就是这里没有调用事务
这里的 return createVoucherOrder(voucherId);
实际上的写法是这样
return this.createVoucherOrder(voucherId);
是用这个类的对象来调用的,但是由于spring底层是通过aop来实现事务管理的,我们要用代理对象才能发起一个事务,不然还是会有问题!!!
代码
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券
SeckillVoucher voucher = seckillVoucher.getById(voucherId);
if (voucher == null) {
return Result.fail("优惠券不存在");
}
//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();
synchronized (userId.toString().intern()) {
//获取代理对象
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
}
}
@Transactional
public Result createVoucherOrder(Long voucherId) {
Long userId = UserHolder.getUser().getId();
//5.一人一单
int count = query()
.eq("user_id", userId)
.eq("voucher_id", voucherId)
.count();
System.out.println("此时count为" + count);
if (count > 0) {
return Result.fail("用户已经购买过一次");
}
//6.扣减库存
boolean isSuccess = seckillVoucher
.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock", 0)
.update();
//7.判断是否成功
if (!isSuccess) {
return Result.fail("扣减库存失败!");
}
//8.创建订单
VoucherOrder voucherOrder = new VoucherOrder();
//8.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//8.2用户id
voucherOrder.setUserId(userId);
//8.3代金券id
voucherOrder.setVoucherId(voucherId);
//8.4保存到voucher_order表中
save(voucherOrder);
//9.返回订单id
return Result.ok(orderId);
}
要设置这个还得加一个依赖
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
</dependency>
在主启动类上
@EnableAspectJAutoProxy(exposeProxy = true)
@MapperScan("com.hmdp.mapper")
@SpringBootApplication
public class HmDianPingApplication {
public static void main(String[] args) {
SpringApplication.run(HmDianPingApplication.class, args);
}
}
一人一单(集群)
搭建集群
我这里搭建的集群是jvm的集群,在idea中的jvm集群
idea复用一个8082接口的应用程序
按alt + 8 可以跳出service
然后复制一份应用程序
更改端口
nginx配置
打开nginx conf文件下的nginx.conf
这里需要修改就是,下面的把注释打开,并且把上面的8081固定的关闭
这里的意思就是,请求8080,转到http://backend
然后nginx自动会轮询这两个server,一个是8081,一个是8082
worker_processes 1;
events {
worker_connections 1024;
}
http {
include mime.types;
default_type application/json;
sendfile on;
keepalive_timeout 65;
server {
listen 8080;
server_name localhost;
# 指定前端项目所在的位置
location / {
root html/hmdp;
index index.html index.htm;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
location /api {
default_type application/json;
#internal;
keepalive_timeout 30s;
keepalive_requests 1000;
#支持keep-alive
proxy_http_version 1.1;
rewrite /api(/.*) $1 break;
proxy_pass_request_headers on;
#more_clear_input_headers Accept-Encoding;
proxy_next_upstream error timeout;
# proxy_pass http://127.0.0.1:8081;
proxy_pass http://backend;
}
}
upstream backend {
server 127.0.0.1:8081 max_fails=5 fail_timeout=10s weight=1;
server 127.0.0.1:8082 max_fails=5 fail_timeout=10s weight=1;
}
}
更改完之后,要在cmd上重新启动一下
nginx.exe -s reload
问题出现
将两个应用程序以调试的模式打开
在这里打个断点
apifox的设置
第一个用户的配置
第二个用户是一样的
也就是说是同一用户,不过是不同的集群
测试
原先数据库的数据
首先是优惠券表
然后是优惠券订单表
是空的
2个接口都发起请求
发现两个应用程序都进去了
测试发现两个请求都进来了,这和我们的一人一单有问题,他这里会生成两个订单
如下
正常来说,我这里设了锁,应该是只能下一单,但是这里再集群情况下,下了两单,所以发生了线程安全问题!
发生问题的有原因
我门要先搞清楚集群的问题,如果是两个集群的化,那么代表的是两个jvm,相当于两个不同的进程,而我们之前那样子加锁,它的范围是jvm的内部,所以这里加锁无效,从这,就引申出分布式锁的概念
简单总结一下,就是没锁上,需要更大范围的锁!
分布式锁
分布式锁有三种实现
-
对于mysql来说,它的互斥锁的实现就是通过事务来实现的,我们再写的时候,会再写上加锁,但我认为这个还是很难的,如果要实现的哈
用redis来实现,比较好实现,就是用setnx,来实现互斥锁
zookeeper 我还不懂,掠过
我这里的获取锁 + 释放锁,已经写好了
代码如下
/**
* 尝试获取锁
* @param pattern key
* @param value 值
* @param <T>
* @return
*/
public <T> boolean tryLock(String pattern,T value)
{
Boolean flag = redisTemplate.opsForValue().setIfAbsent(pattern, value, 2, TimeUnit.MINUTES);
return BooleanUtil.isTrue(flag);
}
/**
* 解锁
* @param pattern
*/
public void unlock(String pattern) {
//删除锁
redisTemplate.delete(pattern);
}
给我封装到了我的redis 操作的工具类里边了
问题解决
我们先来看,解决问题的流程,做好一个心里预期
这个流程还算简洁的
我们直接看解决的代码
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Autowired
private ISeckillVoucherService seckillVoucherService;
@Autowired
private RedisIdWorker redisIdWorker;
@Autowired
private RedisCache redisCache;
/**
* 抢购秒杀券
*
* @param voucherId
* @return
*/
@Override
// @Transactional
public Long seckillVoucher(Long voucherId) {
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
log.info("当前库存为 : {}", voucher.getStock());
if (Objects.isNull(voucher)) {
throw new BaseException("优惠券不存在!");
}
LocalDateTime nowTime = LocalDateTime.now();
//优惠券时间是否开始了
if (voucher.getBeginTime().isAfter(nowTime)) {
throw new BaseException("优惠券时间还没开始!");
}
//是否结束了
if (voucher.getEndTime().isBefore(nowTime)) {
throw new BaseException("优惠券时间已经结束了");
}
//判断库存是否充足
if (voucher.getStock() < 1) {
throw new BaseException("库存不足!");
}
Long userId = UserHolder.getUser().getId();
//锁的value是当前线程id
long threadId = Thread.currentThread().getId();
boolean isSuccess = redisCache.tryLock(RedisConstants.LOCK_SECKILL_VOUCHER_KEY, threadId + "", RedisConstants.LOCK_SECKILL_VOUCHER_TTL, TimeUnit.SECONDS);
if (!isSuccess) {
throw new BaseException("用户已经买过了!");
}
try {
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
} finally {
String lockId = redisCache.getObject(RedisConstants.LOCK_SECKILL_VOUCHER_KEY);
//判断是否是一样的锁
if (StrUtil.isNotBlank(lockId) && lockId.equals(Thread.currentThread().getId() + "")) {
redisCache.unlock(RedisConstants.LOCK_SECKILL_VOUCHER_KEY);
}
}
}
@Transactional
public synchronized Long createVoucherOrder(Long voucherId) {
Long userId = UserHolder.getUser().getId();
//一人一单问题
LambdaQueryWrapper<VoucherOrder> orderWrapper = new LambdaQueryWrapper<>();
orderWrapper.eq(VoucherOrder::getUserId, UserHolder.getUser().getId())
.eq(VoucherOrder::getVoucherId, voucherId);
int count = count(orderWrapper);
if (count > 0) {
throw new BaseException("你已经买过了!");
}
//扣减库存
boolean isSuccess = seckillVoucherService
.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId)
.gt("stock", 0)
.update();
if (!isSuccess) {
throw new BaseException("扣减库存失败!");
}
long orderId = redisIdWorker.nextId("order");
VoucherOrder voucherOrder = VoucherOrder.builder()
.id(orderId)
.userId(userId)
.voucherId(voucherId)
.build();
save(voucherOrder);
return orderId;
}
}
改动的代码如下
代码很简洁,就是加一个锁,只不过这个锁是再redis中,如果获取失败,那就说明,有人再抢,有人再抢的化,就直接爆错退出,这样才符合一人一单
测试
这里的测试我就不写了,因为没什么意思,最后结果就是一人一单
小问题
这里的小问题,就是如果按照我门上面这种写法,锁住是所有的用户,而我们加锁是对各个用户加锁,做到一人一单,用户之间应该是隔离的才对,所以这里应该再锁上加上用户的标记,这样别的用户就可以进来了,去枪单
这里实现还是很简单的,就是写rediskey的时候加上 userId
分布式锁误删问题
因为我们加了redis分布式锁,并且这里的分布式锁,是有过期时间的,所以就会延申出这个问题
为什么要设置过期时间
我们首先先声明一点,为什么必须要加过期时间,咱们这个问题的出现就是由于这个过期时间的问题,为什么我们不能做成永久key呢?
这个问题的答案,就是如果我们做成永久key,一个线程拿到了锁,然后突然发生异常了,或者业务阻塞了, 那么就相当于说,锁释放不了了,那么程序的性能就会大大降低,甚至于我们有可能得人为去干预这个问题
虽然说,我们这个例子,理论上来说,是可以用永久key的,但是大部分的业务是不行的,所以这里要设置过期时间
问题的出现
这个问题得想一想才行
这个情况是比较极端的,但是不代表没有可能会出现
极端情况下,我们线程1占有了锁,然后突然业务阻塞了,但是业务阻塞的时间比锁的过期时间还要长,这就会导致业务还没结束,锁已经被释放了!
那么在高并发的情况下,另外一个线程2,乘虚而入,拿到了锁,并且开始执行业务
而在这个时候,在线程2拿到了锁,线程1突然醒了过来,执行了业务代码,然后去执行释放锁的代码
那么这里就会出问题了,线程1不知道这个锁已经被换了主人了,他直接就把锁释放掉了
那么会导致什么结果呢???
在高并发的情况下,线程3一看没有锁了,就乘虚而入
线程2还没执行完
线程3就拿到了锁
这样下去,当线程2完成了业务,他就释放了锁,那么此时的线程3本来持有锁的,锁被人删了,后面线程4就乘虚而入
这样就像线程1删了线程2的锁
线程2删了线程3的锁
线程3删了线程4的锁
这样子迭代下去,不出问题才怪
这个问题属于是线程安全问题
解决办法
解决办法的思想也很简单,既然你删错锁了,是因为你不知道此时的锁的主人是谁,你以为是自己的,那么我门只要在锁上写上一个标记,代表着此时的锁是谁的.我们去释放锁的时候,就去判断是不是自己的锁,这样就没什么问题了
代码
按道理来说这里不应该用线程id来当作标识,因为还是有可能会重复,所以应该用uuid来当标识才对
修改如下
这样就不会有可能是有重复的问题了
原子性问题的出现
我们看上面的解决办法,好像已经很不错了,但是还是有一个漏洞,那就是这里的流程 判断锁是不是自己 和 释放锁不是一个原子操作
我们来看这个图,就能看明白,这里是一个很极端的情况
首先线程1拿到锁,然后执行完业务,想要释放锁,按照我们的解决方法,我们获取锁,是不是自己,发现是, 就在这个瞬间,突然线程1发生了阻塞
这里的阻塞,有可能是jvm的垃圾回收所导致,或者其他
当我们阻塞的时间超过了锁的过期时间,就会超时释放锁
那么线程2也会乘虚而入,拿到锁
当线程1醒过来的时候,因为前面已经判断过了,所以就会去删线程2的锁
还是会出现误删问题!!!
当然了,出现这个问题的条件是很苛刻的,就是线程1在判断完锁是自己的时候,突然发生阻塞,并且阻塞的时间超过redis锁的过期时间
解决办法
所以我们要想解决这个棘手的问题,我们就要让判断锁是不是自己 和 释放锁变成一个原子操作
这里就引出了redis 的lua脚本,它可以做到原子性!
LUA脚本
简单的介绍lua脚本
它是一个脚本语言,有点类似于js
在redis中,执行lua脚本
这里的key 和 value可以不用写死,可以作为参数传递
特别要注意这里的KEYS,和ARGV数组,需要注意的是,这里的数组是从1开始的
解决上面的问题
先写一个lua脚本
脚本的意思就是判断锁 + 释放锁
原先java代码的改变
/**
* 释放锁
*/
private void unlock(String lockKey,String uuid) {
DefaultRedisScript<Long> longDefaultRedisScript = new DefaultRedisScript<>();
longDefaultRedisScript.setLocation(new ClassPathResource("unlock.lua"));
longDefaultRedisScript.setResultType(Long.class);
redisCache.execute(
longDefaultRedisScript,
Arrays.asList(lockKey),
uuid
);
}
这里不再我的工具类里边写unlock了,这里的unlock比较特殊所以要自己写一个方法在下边
这样字,这样的代码就十分健壮了
