基于Redis和阻塞队列的异步秒杀业务

异步前

之前的秒杀业务的查询优惠券、查询订单、减库存、创建订单都要查询数据库，而且有分布式锁，使得整个业务耗时长，对此采用异步操作处理，异步操作类似于餐厅点餐，服务员负责点菜产生订单、厨师负责根据订单后厨做饭，整个流程由服务员和厨师两个线程完成，此为异步。

可以看到异步优化前，1000个请求的耗时均值497ms

异步优化方案

将判断秒杀库存和校验一人一单的操作放在redis进行，优惠券库存信息也放入redis以减少读取数据库的压力，采用set集合存储购买过优惠券的用户的id，set集合有元素不重复的特性，可以自动实现一人一单

整体业务逻辑如下：

Redis实现库存和秒杀资格判断（需求1和2）

优惠券信息保存到redis

修改添加秒杀券的代码，在添加秒杀券的同时把信息也保存到redis中

@Override@Transactionalpublic 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);//保存秒杀券信息到redisstringRedisTemplate.opsForValue().set("seckill:stock:"+voucher.getId(),voucher.getStock().toString());}

添加秒杀券，信息成功添加到redis中，秒杀券id是13，库存是100，如下图所示：

lua脚本查询redis中库存和一人一单购买资格

seckill.lua

---
--- Created by 懒大王Smile.
--- DateTime: 2024/7/6 10:47
---
-- 1.参数列表
-- 1.1优惠券id
local voucherId=ARGV[1]--1.2 用户id
local userId=ARGV[2]--2.数据key  ..是拼接符号
--2.1 库存key
local stockKey='seckill:stock:'..voucherId
--2.2 订单key
local orderKey='seckill:order:'..voucherId--3.脚本业务
--3.1 判断库存是否充足
if (tonumber(redis.call('get',stockKey))<=0) thenreturn 1
end
--3.2判断用户是否下单   若set集合中存在该用户id，则说明已下过单，返回1
if (tonumber(redis.call('sismember',orderKey,userId))==1) thenreturn 2
end--3.4扣库存
redis.call('incrby',stockKey,-1)
--3.5保存用户到set
redis.call('sadd',orderKey,userId)
return 0

VoucherOrderServiceImpl.java

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder>  implements IVoucherOrderService {@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate ISeckillVoucherService SeckillVoucherService;@Autowiredprivate RedissonClient redissonClient;@Resourceprivate RedisIdWorker redisIdWorker;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static  {SECKILL_SCRIPT=new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}@Overridepublic Result seckillVoucher(Long voucherId) {Long userId = UserHolder.getUser().getId();//执行lua脚本判断有无购买资格Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());int i = result.intValue();if (i!=0){return Result.fail(i==1?"优惠券库存不足":"不能重复下单");}long orderId = redisIdWorker.nextId("order:");//生成了orderId，把订单信息保存到阻塞队列，由另一个线程专门根据订单信息去数据库做增删改查，这就实现了异步//TODOreturn Result.ok(orderId);}
}

运行效果

同一用户两次下单id为13的秒杀券，第一次成功，第二次失败，如下图：

再次查看redis中voucherId=13的秒杀券，库存减1，且该对该秒杀券下单成功的用户已经存入set集合，userId=1010

优化后模拟大量用户抢购秒杀券的测试

优化后，1000个请求的耗时均值为178ms，相比最初的497ms减少很多

阻塞队列实现异步秒杀下单（需求3和4）

阻塞队列
当线程尝试从队列中获取元素时，若队列无元素，则线程会阻塞，直到队列中有元素才会被唤醒

前面实现了redis秒杀券资格判断，若该用户有资格，则其userId存入redis订单中，且redis中秒杀券库存自减

订单加入阻塞队列

//定义阻塞队列
private BlockingQueue<VoucherOrder> orderTasks=new ArrayBlockingQueue<>(1024*1024);

//订单加入阻塞队列//创建订单VoucherOrder order = new VoucherOrder();order.setVoucherId(voucherId);//TODO 生成了orderId，把订单信息保存到阻塞队列，由另一个线程专门根据订单信息去数据库做增删改查，这就实现了异步long orderId = redisIdWorker.nextId("order:");order.setId(orderId);order.setUserId(userId);//添加到阻塞队列orderTasks.add(order);

从阻塞队列中获取订单然后操作数据库

这里定义线程池，让线程去从阻塞队列中获取订单，实现异步操作数据库

//定义线程池，负责从阻塞队列中获取订单然后异步下单private static final ExecutorService SECKILL_ORDER_EXECUTOR= Executors.newSingleThreadExecutor();//定义线程  这是个内部类private class VoucherOrderHandler implements Runnable{@Overridepublic void run() {while (true){try {//获取队列中的订单VoucherOrder order = orderTasks.take();//创建订单handleVoucherOrder(order);} catch (InterruptedException e) {log.error("订单处理异常",e);}}}}//spring提供的注解  作用：类初始化后就执行VoucherOrderHandler方法
//向线程池提交一个线程@PostConstructprivate void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}private void handleVoucherOrder(VoucherOrder order) {Long userId = order.getUserId();RLock redisLock = redissonClient.getLock("lock:order:" + userId);boolean tryLock = redisLock.tryLock();//判断锁是否获取成功if (!tryLock){log.error("不允许重复下单");return ;}try {proxy.createVoucherOrder(order);//使用动态代理类的对象，事务可以生效} finally {redisLock.unlock();}}

完整代码

VoucherOrderServiceImpl.java

@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder>  implements IVoucherOrderService {@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate ISeckillVoucherService SeckillVoucherService;@Autowiredprivate RedissonClient redissonClient;@Resourceprivate RedisIdWorker redisIdWorker;//阻塞队列  当线程尝试从队列中获取元素时，若队列无元素，则线程会阻塞，直到队列中有元素才会被唤醒private BlockingQueue<VoucherOrder> orderTasks=new ArrayBlockingQueue<>(1024*1024);//线程池，负责从阻塞队列中获取订单然后异步下单private static final ExecutorService SECKILL_ORDER_EXECUTOR= Executors.newSingleThreadExecutor();//spring提供的注解  作用：类初始化后就执行VoucherOrderHandler方法//向线程池提交一个线程@PostConstructprivate void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}//线程  内部类private class VoucherOrderHandler implements Runnable{@Overridepublic void run() {while (true){try {//获取队列中的订单信息VoucherOrder order = orderTasks.take();//创建订单handleVoucherOrder(order);} catch (InterruptedException e) {log.error("订单处理异常",e);}}}}//代理对象//因为异步之后，子线程不能获取代理对象无法实现事务，所以要定义为全局变量，在主线程中就获取代理对象给子线程用IVoucherOrderService proxy;private void handleVoucherOrder(VoucherOrder order) {Long userId = order.getUserId();RLock redisLock = redissonClient.getLock("lock:order:" + userId);boolean tryLock = redisLock.tryLock();//判断锁是否获取成功if (!tryLock){log.error("不允许重复下单");return ;}try {//锁加到这里，事务提交后才释放锁proxy.createVoucherOrder(order);//使用动态代理类的对象，事务可以生效} finally {redisLock.unlock();}}private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static  {SECKILL_SCRIPT=new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}@Overridepublic Result seckillVoucher(Long voucherId) {Long userId = UserHolder.getUser().getId();//执行lua脚本判断有无购买资格Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());int i = result.intValue();if (i!=0){return Result.fail(i==1?"优惠券库存不足":"不能重复下单");}//创建订单VoucherOrder order = new VoucherOrder();order.setVoucherId(voucherId);//TODO 生成了orderId，把订单信息保存到阻塞队列，由另一个线程专门根据订单信息去数据库做增删改查，这就实现了异步long orderId = redisIdWorker.nextId("order:");order.setId(orderId);order.setUserId(userId);//添加到阻塞队列orderTasks.add(order);//获取事务的动态代理对象，需要在启动类加注解暴漏出对象proxy = (IVoucherOrderService)AopContext.currentProxy();//拿到动态代理对象return Result.ok(orderId);}//TODO spring对该类做了动态代理，用动态代理的对象提交的事务@Transactionalpublic void createVoucherOrder(VoucherOrder order) {//一人一单，根据优惠卷id和用户id去数据库查询是否已经存在该优惠卷Long id = order.getUserId();//为用户id加锁而不是对整个createVoucherOrder方法加锁，减小锁范围，提升性能，这样每个用户就有不同的锁//锁加在函数内部，锁内的代码执行完后就会释放锁，而事务的提交是在整个方法执行后提交的，也就是事务的提交在锁释放之后。//但是锁释放后其他线程就可以进来，此时事务可能还没有提交，可能出现并发问题，重复购买//所以要扩大锁的范围，把锁加到seckillVoucher方法后面，在事务提交后才能释放锁！int count = query().eq("user_id", id).eq("voucher_id", order).count();if (count >=1) {//count==1说明用户拥有了一个优惠券log.error("不能重复下单");return;
//                return Result.fail("不能重复购买优惠卷");}//4.扣减库存  防止超卖，加乐观锁，扣减库存前再查询一次库存判断
//        boolean b = SeckillVoucherService.update()
//                .setSql("stock=stock-1").
//                eq("voucher_id", voucherId).eq("stock",voucher.getStock()).update();//使用setSql方法设置了更新语句"stock=stock-1"，接着使用eq方法添加了两个条件："voucher_id"等于voucherId和"stock"等于voucher.getStock()//条件1：voucher_id=voucherId指当前操作的优惠卷的id=数据库中的优惠卷id，即通过优惠卷id指明了要修改哪个优惠卷的库存//条件2：stock=voucher.getStock,说明该线程修改库存期间没有其他线程来插队修改库存，那么数据是安全的//TODO ！！！注意！这种操作在并发情况下可能导致用户在优惠卷库存充足的情况下抢购优惠卷失败，也就是即使有库存也会抢购失败，此时可以判断库存是否充足，重新抢购//修改如下：最后库存判断，只要>0就可以修改boolean b = SeckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", order).gt("stock", 0).update();if (!b) {
//                return Result.fail("库存不足");log.error("库存不足");return;}save(order);}
}