秒杀相关问题解决

秒杀

超卖问题

如下,我们先来复现问题,抢购秒杀券的代码逻辑也是很简单,

先判断优惠券是否开始了,是的化,判断库存是否充足,如果是的化,扣减库存,最后创建订单

如下是代码

@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比较特殊所以要自己写一个方法在下边

这样字,这样的代码就十分健壮了