黑马点评-实现安全秒杀优惠券(使并发一人一单，防止并发超卖)

一.实现优惠券秒杀

1.最原始代码：

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Override@Transactionalpublic Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 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 success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).update();if (!success) {//扣减库存return Result.fail("库存不足！");}//6.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 6.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 6.2.用户idLong userId = UserHolder.getUser().getId();voucherOrder.setUserId(userId);// 6.3.代金券idvoucherOrder.setVoucherId(voucherId);save(voucherOrder); // 写入数据库return Result.ok(orderId);}
}

出现的问题：在高并法的情况下存在多买的情况，导致商品成为负数 

代码出现问题的原理图：

2.使用乐观锁解决超卖的问题

在修改时，查看之前查出来的数据和现在的数据库里面的优惠券的数量是否相同，相同就表示期间没有修改，则我们可以执行减库存的操作。

代码修改： 

// 把上面的sql语句改成这个
boolean success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).eq("stock",voucher.getStock()) //  添加乐观锁，但是成功率太低.update();

出现的问题：200个用户抢100个优惠券这100个优惠券还有剩余。 

3.优化乐观锁

在修改时，其实我们只用查看库存剩余的券的数量是否大于0就可以执行减库存了，不需要向上面一样写的这么严格。

 代码修改：

// 把上面的sql语句改成这个
boolean success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).gt("stock",0) //  优化乐观锁.update();

二.实现一人一单

1.有问题的代码

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Override@Transactionalpublic Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 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 success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).eq("stock",voucher.getStock()) //  添加乐观锁，但是成功率太低.update();if (!success) {//扣减库存return Result.fail("库存不足！");}// 6. 一人一单Long userId = UserHolder.getUser().getId();// 6.1. 查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 6.2. 判断是否存在if(count > 0){// 用户已经购买过了return Result.fail("用户已经购买过一次！");}//7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 7.2.用户idvoucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);save(voucherOrder); // 写入数据库return Result.ok(orderId);}
}

出现的问题：在高并发的情况下还是会出现一个人可以下多单

出现问题的原理图：

2.使用悲观锁：

更改步骤5之后的逻辑，添加synchronized

@Transactional
public synchronized Result createVoucherOrder(Long voucherId) {// 5. 一人一单Long userId = UserHolder.getUser().getId();// 5.1.查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("用户已经购买过一次！");}// 6.扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1") // set stock = stock - 1.eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0.update();if (!success) {// 扣减失败return Result.fail("库存不足！");}// 7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 7.2.用户idvoucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);
}

但是这样添加锁，锁的粒度太粗了，在使用锁过程中，控制 锁粒度 是一个非常重要的事情，因为如果锁的粒度太大，会导致每个线程进来都会锁住，所以我们需要去控制锁的粒度。

3.优化悲观锁的使用

思路：我们对用户的id进行上锁。但是，注意这里toString.intern，这里和tostring的底层原理有关使用了这个相当于创建了一个新的对象，所以同一id不会被锁住，加了intern就会去string的常量池里面去找，就不会创建新的了

@Transactional
public  Result createVoucherOrder(Long voucherId) {Long userId = UserHolder.getUser().getId();synchronized(userId.toString().intern()){// 5.1.查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("用户已经购买过一次！");}// 6.扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1") // set stock = stock - 1.eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0.update();if (!success) {// 扣减失败return Result.fail("库存不足！");}// 7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 7.2.用户idvoucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);}
}

4.再次优化 

但是这里我们还需要注意，这里是先释放锁在提交事务的，可能会出现问题，所以我们要把函数里面的锁去掉，在调用的地方加锁即可

Long userId = UserHolder.getUser().getId();
synchronized (userId.toString().intern()) {return this.createVoucherOrder(voucherId);
}

5.继续优化，前面的修改会导致事务失效

所以我们要获取代理对象并用代理对象调用 createVoucherOrder 方法

关于事务失效可以查看这篇博客 点击进入

如果还是不清楚可以查看一人一单视频的最后一点 

synchronized (userId.toString().intern()) {// 获取代理对象（事务）IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);
}

// 这里还需要在启动类上面添加这个注解，去暴露代理对象
@EnableAspectJAutoProxy(exposeProxy = true)

出现的问题：在多节点部署服务器的时候，还是会出现一人多单的问题，因为每个jvm的锁的监视器不同导致，无法锁住导致一人多单，这里就需要分布式锁

 出现问题的原理图：

6.使用分布式锁解决集群部署的并发问题

解决问题的原理图：

具体实现可以查看这篇博客  点击进入

7.使用异步的思路解决高并发

实现异步的思路的流程图：

① 新增秒杀优惠券的同时，将优惠券信息保存到Redis中

// 改造添加优惠券代码
@Transactional
public void addSeckillVoucher(Voucher voucher) {// 保存优惠券save(voucher);// 保存秒杀信息SeckillVoucher seckillVoucher = new SeckillVoucher();seckillVoucher.setVoucherId(voucher.getId());seckillVoucher.setStock(voucher.getStock());seckillVoucher.setBeginTime(voucher.getBeginTime());seckillVoucher.setEndTime(voucher.getEndTime());seckillVoucherService.save(seckillVoucher);// 保存秒杀库存到Redis中stringRedisTemplate.opsForValue().set(RedisConstants.SECKILL_STOCK_KEY + voucher.getId(),voucher.getStock().toString());
}

② 基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功

lua脚本实现思路的流程图：

-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]-- 2.数据key
-- 2.1.库存key
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2.订单key
local orderKey = 'seckill:order:' .. voucherId-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then-- 3.2.库存不足，返回1return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then-- 3.3.存在，说明是重复下单，返回2return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0

③ 如果抢购成功，将优惠券id和用户id封装后存入阻塞队列

public class VoucherOrderServiceImpl {private BlockingQueue<VoucherOrder> orderTasks =new ArrayBlockingQueue<>(1024 * 1024);private IVoucherOrderService proxy; // 保存这个类的代理对象public Result seckillVoucher(Long voucherId) {//获取用户Long userId = UserHolder.getUser().getId();// 1.执行lua脚本Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());int r = result.intValue();// 2.判断结果是否为0if (r != 0) {// 2.1.不为0 ，代表没有购买资格return Result.fail(r == 1 ? "库存不足" : "不能重复下单");}// 2.2. 为0，有购买资格，把下单信息保存到阻塞队列VoucherOrder voucherOrder = new VoucherOrder();// 2.3.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 2.4.用户idvoucherOrder.setUserId(userId);// 2.5.代金券idvoucherOrder.setVoucherId(voucherId);// 2.6.放入阻塞队列orderTasks.add(voucherOrder);// 3.获取代理对象，保证后面线程池中使用代理对象去完成消费阻塞队列里面的消息proxy = (IVoucherOrderService)AopContext.currentProxy();// 4.返回订单idreturn Result.ok(orderId);}
}

④ 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

@Slf4j
public class VoucherOrderServiceImpl {//异步处理线程池private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();private IVoucherOrderService proxy;@PostConstruct // 当前类初始化完毕就执行private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}private class VoucherOrderHandler implements Runnable {public void run() {while (true) {  // 一直去阻塞队列获取消息try {// 1.获取队列中的订单信息VoucherOrder voucherOrder = orderTasks.take();// 2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("处理订单异常", e);}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.获取用户Long userId = voucherOrder.getUserId();// 2.创建锁对象RLock redisLock = redissonClient.getLock("lock:order:" + userId);// 3.尝试获取锁boolean isLock = redisLock.lock();// 4.判断是否获得锁成功if (!isLock) {// 获取锁失败，直接返回失败或者重试log.error("不允许重复下单！");return;}try {//注意：由于是spring的事务是放在threadLocal中，此时的是多线程，事务会失效proxy.createVoucherOrder(voucherOrder);} finally {// 释放锁redisLock.unlock();}}@Transactionalpublic  void createVoucherOrder(VoucherOrder voucherOrder) {Long userId = voucherOrder.getUserId();// 5.1.查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了log.error("用户已经购买过了");return ;}// 6.扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1") // set stock = stock - 1.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0) // where id = ? and stock > 0.update();if (!success) {// 扣减失败log.error("库存不足");return ;}// 7.创建订单save(voucherOrder);}
}

 存在的问题：上面的阻塞队列在jvm中，会存在内存限制的问题，如果服务器宕机了，就会导致数据全部消失。

然后我们可以使用redis中的stream这个数据结构作为消息队列进行优化，后面优化的代码就不写了，因为要使用消息队列的话还是建议使用 rabbitmq 这种开源的消息队列组件就行了，这里我们只需要懂思想了。