黑马点评项目

黑马点评项目"/>

黑马点评项目

一、异步秒杀思路

来看下之前的秒杀业务的整体流程：

前端发起请求到达 Nginx，Nginx 通过负载均衡，将请求转发至 Tomcat。在 Tomcat 中，程序的执行流程如上图所示，整个业务流程串行执行，所以，整个业务的耗时时间就是每一步的耗时之和。但是，在整个业务流程中，其中，查询优惠券、查询订单、减库存以及创建订单这四步都需要与数据库建立连接，执行相关的增删改查操作。由于数据库本身的并发能力是比较差的，再加上减库存和创建订单还是对数据库的写操作，另外为了避免线程安全问题，在执行减库存以及创建订单逻辑时间，还增加了分布式锁，这就导致了整体业务的耗时就会比较长，并发能力比较弱。

如何进行优化呢？
由于判断秒杀库存以及校验一人一单的逻辑执行时间较短，而减库存、创建订单是对数据库的写操作，耗时较久，可以将这两个部分拆分开来，由不同的线程进行执行。请求进来以后，主线程判断用户的购买资格，如果用户有购买资格，则开启独立线程来处理耗时较久的减库存以及下单操作，这样执行效率就会大大提高。为了进一步提高项目的性能，还应该进一步提高对于秒杀资格的判断的执行效率。由于判断秒杀资格依然需要查询数据库，为了提高效率，完全可以将优惠券信息以及订单信息缓存到 Redis 中，把对于秒杀资格的判断放到 Redis 中来执行。当秒杀资格判断执行结束后，程序可以直接将订单 id 返回给用户，用户则可以拿着订单 id 完成后续的付款等操作。对于减库存以及下单操作，如果用户有资格下单，就可以将优惠券 id、用户 id以及订单 id 等信息存储到阻塞队列中，然后由独立线程异步读取阻塞队列中的信息，完成操作。

不过这里有个难点，就是如何在 Redis 中完成秒杀库存的判断和一人一单的判断？
要想在 Redis 中判断库存是否充足以及一人一单，就需要将库存信息以及有关的订单信息缓存到 Redis 中，那我们应该选择什么样的数据结构来存储库存信息以及订单信息呢？优惠券的库存比较简单，库存是一个数据，可直接使用 String 类型进行存储，key 为优惠券的 id，value 为库存的值。要实现一人一单功能，就需要在 Redis 中记录当前优惠券被哪些用户购买过，后续再有用户购买时，只需要判断该用户是否在记录当中存在。那什么样的数据结构满足这样的需求呢？该数据结构首先应当满足在一个 key 中可以保存多个值，即一个订单对应多个用户，其次，由于一人一单，那么保存的用户 id 就不能重复。很明显，Set 类型的数据结构满足这样的需求。

再来复习下 Set 结构的特点：

• 无序
• 元素不可重复
• 查找快
• 支持交集、并集、差集等功能

分析下业务的执行流程：
由于秒杀库存以及校验一人一单对 Redis 的判断较多，业务流程较多，为了保证业务执行的原子性，需使用 Lua 脚本来完成。

二、改进秒杀业务，提高并发性能

需求：
① 新增秒杀优惠券的同时，将优惠券信息保存到 Redis 中
② 基于 Lua 脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
③ 如果抢购成功，将优惠券 id 和用户 id 封装后存入阻塞队列
④ 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

2.1 代码实现

VoucherServiceImpl，修改 addSeckillVoucher 方法，新增秒杀优惠券的同时，将优惠券信息保存到 Redis 中

@Service
public class VoucherServiceImpl extends ServiceImpl<VoucherMapper, Voucher> implements IVoucherService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Overridepublic Result queryVoucherOfShop(Long shopId) {// 查询优惠券信息List<Voucher> vouchers = getBaseMapper().queryVoucherOfShop(shopId);// 返回结果return Result.ok(vouchers);}@Override@Transactional(rollbackFor = Exception.class)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 脚本：seckill.lua

-- 优惠券id
local voucherId = ARGV[1]
-- 用户id
local userId = ARGV[2]-- 库存key
local stockKey = "seckill:stock:"..voucherId
-- 订单key
local orderKey = "seckill:order:"..voucherId-- 判断库存是否充足
if(tonumber(redis.call('get', stockKey)) <= 0) thenreturn 1
end-- 判断用户是否已经下过单
if(redis.call('sismember', orderKey, userId) == 1) thenreturn 2
end-- 扣减库存
redis.call('incrby', stockKey, -1)-- 将 userId 存入当前优惠券的 set 集合
redis.call('sadd', orderKey, userId)return 0

VoucherOrderServiceImpl，修改判断库存充足以及一人一单逻辑

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Autowiredprivate ISeckillVoucherService seckillVoucherService;@Autowiredprivate RedisIdWorker redisIdWorker;@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Autowiredprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}/**** 创建阻塞队列，并初始化阻塞队列的大小*/private static final BlockingQueue<VoucherOrder> orderTasks =new ArrayBlockingQueue<>(1024*1024);/**** 创建线程池*/private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();/**** 标有 @PostConstruct 注解的方法，容器在 bean 创建完成并且属性赋值完成后，会调用该初始化方法。* 容器启动时，便开始创建独立线程，从队列中读取数据，创建订单*/@PostConstructprivate void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {// 系统启动开始，便不断从阻塞队列中获取优惠券订单信息while(true){try {// 阻塞式获取订单信息VoucherOrder voucherOrder = orderTasks.take();createVoucherOrder(voucherOrder);} catch (InterruptedException e) {e.printStackTrace();}}}}private void createVoucherOrder(VoucherOrder voucherOrder) {// 判断当前优惠券用户是否已经下过单// 用户 idLong userId = UserHolder.getUser().getId();Long voucherId = voucherOrder.getVoucherId();RLock lock = redissonClient.getLock("lock:order:" + userId);// 获取互斥锁// 使用空参意味着不会进行重复尝试获取锁boolean isLock = lock.tryLock();if (!isLock) {// 获取锁失败，直接返回失败或者重试log.error("不允许重复下单！");return;}try {// 查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();if (count > 0) {log.error("不允许重复下单！");return;}// 扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).gt("stock", 0).update();// 扣减失败if (!success) {log.error("库存不足！");return;}// 创建订单save(voucherOrder);} finally {// 释放锁lock.unlock();}}@Overridepublic Result seckillVoucher(Long voucherId) {UserDTO user = UserHolder.getUser();// 执行 lua 脚本Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), user.getId().toString());int r = result.intValue();// 判断结果是否为 0if(r != 0){// 不为 0 ，代表没有购买资格Result.fail(r == 1 ? "库存不足！" : "不能重复下单!");}// 生成订单 idLong orderId = redisIdWorker.nextId("oder");// 为 0，有购买资格，把订单信息保存到阻塞队列VoucherOrder voucherOrder = new VoucherOrder();voucherOrder.setVoucherId(voucherId);voucherOrder.setUserId(user.getId());voucherOrder.setId(orderId);orderTasks.add(voucherOrder);// 返回订单 idreturn Result.ok(orderId);}
}

三、秒杀优化总结

秒杀业务的优化思路是什么？
① 先利用 Redis 完成库存余量、一人一单判断，完成抢单业务
② 再将下单业务放入阻塞队列，利用独立线程异步下单

基于阻塞队列的异步秒杀存在哪些问题？
① 内存限制问题。我们使用的 JDK 中的阻塞队列，使用的是 JVM 的内存，如果不加以限制，在高并发的情况下，就会有无数的订单对象需要去创建，并且存入阻塞队列中，可能会导致将来内存溢出，所以我们在创建阻塞队列的时候，设置了队列的长度。但是如果队列中订单信息存满了，后续新创建的订单就无法存入队列中。
② 数据安全问题。我们是基于内存保存的订单信息，如果服务突然宕机，那么内存中的订单信息也就丢失了。