Redis实战之Redis如何实现异步秒杀优化详解

网络
秒杀优化-异步秒杀思路 未优化的思路 当用户发起请求，此时会请求nginx，nginx会访问到tomcat，而tomcat中的程序，会进行串行操作，分成如下几个

秒杀优化-异步秒杀思路

未优化的思路

当用户发起请求，此时会请求nginx，nginx会访问到tomcat，而tomcat中的程序，会进行串行操作，分成如下几个步骤

1、查询优惠卷

2、判断秒杀库存是否足够

3、查询订单

4、校验是否是一人一单

5、扣减库存

6、创建订单

在这六步操作中，又有很多操作是要去操作数据库的，而且还是一个线程串行执行， 这样就会导致我们的程序执行的很慢

优化方案

我们将耗时比较短的逻辑判断放入到redis中，比如是否库存足够，比如是否一人一单，这样的操作，只要这种逻辑可以完成，就意味着我们是一定可以下单完成的，我们只需要进行快速的逻辑判断，根本就不用等下单逻辑走完，我们直接给用户返回成功， 再在后台开一个线程，后台线程慢慢的去执行queue里边的消息，即不追求时效性，让用户先成功下单，后续再完善数据库数据

整体思路

用户下单之后，判断库存是否充足只需要到redis中去根据key找对应的value是否大于0即可，如果不充足，则直接结束，如果充足，继续在redis中判断用户是否可以下单，如果set集合中没有这条数据，说明他可以下单，如果set集合中没有这条记录，则将userId和优惠卷存入到redis中，并且返回0，整个过程需要保证是原子性的，我们可以使用lua来操作

当以上判断逻辑走完之后，我们可以判断当前redis中返回的结果是否是0 ，如果是0，则表示可以下单，则将之前说的信息存入到到queue中去，然后返回，然后再来个线程异步的下单，前端可以通过返回的订单id来判断是否下单成功。

难点

  • 怎么在redis中去快速校验一人一单，还有库存判断
  • 由于我们校验和tomct下单是两个线程，那么我们如何知道到底哪个单他最后是否成功，或者是下单完成，为了完成这件事我们在redis操作完之后，我们会将一些信息返回给前端，同时也会把这些信息丢到异步queue中去，后续操作中，可以通过这个id来查询我们tomcat中的下单逻辑是否完成了。

代码实现

需求：

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

新增优惠券，将优惠券信息入库并写入redis

@Override
    @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().setIfAbsent(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
    }

判断秒杀库存、一人一单，决定用户是否抢购成功，考虑到操作的原子性，采用lua脚本完成这一连串的操作

---
--- Generated by EmmyLua(https://github.com/EmmyLua)
--- Created by Lenovo.
--- DateTime: 2023/9/5 20:57
---
-- 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.库存不足，返回1
    return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
    -- 3.3.存在，说明是重复下单，返回2
    return 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

执行lua脚本，判断是否抢购成功，如果抢购成功，要放入堵塞队列中

@Override
    public Result seckillVoucher(Long voucherId) {
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        //判断是否开始，开始时间如果在当前时间之后就是尚未开始
        if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始");
        }
        //判断是否结束，结束时间如果在当前时间之前就是已经结束
        if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束");
        }
        Long userId = UserHolder.getUser().getId();
        long orderId = new RedisIdWorker(stringRedisTemplate).nextId("order");
        Long execute = stringRedisTemplate.execute(SILLL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(), String.valueOf(orderId)
        );
        int r = execute.intValue();
        if (r != 0) {
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }
        VoucherOrder voucherOrder = new VoucherOrder();
        //订单id
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setId(orderId);
        //将订单信息放入阻塞队列
        orderTakes.add(voucherOrder);
        return Result.ok(orderId);
    }

定义线程内部类，不断从堵塞队列中读取订单

//从阻塞队列里面取订单信息
    private class voucherOrderHander implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    VoucherOrder take = orderTakes.take();
                    handleVoucherOrder(take);
                } catch (Exception e) {
                    log.error("异常信息如下", e);
                }
            }
        }

获取订单信息的具体方法,这里依然加了分布式锁，是为了保险起见

 private void handleVoucherOrder(VoucherOrder take) {
            Long userId = take.getId();
            //创建锁对象
            RLock lock = redissonClient.getLock("lock:order:" + userId);
            //尝试获取锁
            boolean isLock = lock.tryLock();
            //获取锁失败
            if (!isLock) {
                log.error("不允许重复下单");
                return;
            }
            try {
                voucherOrderService.createVoucherOrder(take);
            } finally {
                //释放锁
                lock.unlock();
            }
        }
    }

这里又有一个问题，就是我们订单信息入库应该是在该类对象被创建的时候就要开启线程在堵塞队列等待读取是否有订单信息，然后顺利入库，所以我们用了aop的@PostConstruct，保证该对象被创建时，线程也能顺利创建，这里用了线程池来提交线程任务

@PostConstruct
    public void init() {
        SECKILL_ORDER_EXECUTOR.execute(new voucherOrderHander());
    }

完整代码实现

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private IVoucherOrderService voucherOrderService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedissonClient redissonClient;
    private static final DefaultRedisScript<Long> SILLL_SCRIPT;
    BlockingQueue<VoucherOrder> orderTakes = new ArrayBlockingQueue<>(1024 * 1024);
    //异步处理线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    static {
        SILLL_SCRIPT = new DefaultRedisScript<>();
        SILLL_SCRIPT.setLocation(new ClassPathResource("skill.lua"));
        SILLL_SCRIPT.setResultType(Long.class);
    }
    @PostConstruct
    public void init() {
        SECKILL_ORDER_EXECUTOR.execute(new voucherOrderHander());
    }
    //从阻塞队列里面取用户信息
    private class voucherOrderHander implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    VoucherOrder take = orderTakes.take();
                    handleVoucherOrder(take);
                } catch (Exception e) {
                    log.error("异常信息如下", e);
                }
            }
        }
        private void handleVoucherOrder(VoucherOrder take) {
            Long userId = take.getId();
            //创建锁对象
            RLock lock = redissonClient.getLock("lock:order:" + userId);
            //尝试获取锁
            boolean isLock = lock.tryLock();
            //获取锁失败
            if (!isLock) {
                log.error("不允许重复下单");
                return;
            }
            try {
                voucherOrderService.createVoucherOrder(take);
            } finally {
                //释放锁
                lock.unlock();
            }
        }
    }
    @Override
    public Result seckillVoucher(Long voucherId) {
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        //判断是否开始，开始时间如果在当前时间之后就是尚未开始
        if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始");
        }
        //判断是否结束，结束时间如果在当前时间之前就是已经结束
        if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀已经结束");
        }
        Long userId = UserHolder.getUser().getId();
        long orderId = new RedisIdWorker(stringRedisTemplate).nextId("order");
        Long execute = stringRedisTemplate.execute(SILLL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(), String.valueOf(orderId)
        );
        int r = execute.intValue();
        if (r != 0) {
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }
        VoucherOrder voucherOrder = new VoucherOrder();
        //订单id
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setId(orderId);
        //将订单信息放入阻塞队列
        orderTakes.add(voucherOrder);
        return Result.ok(orderId);
    }
@Transactional
    public 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;
        }
        save(voucherOrder);
    }

以上就是Redis实战之Redis实现异步秒杀优化详解的详细内容，更多关于Redis实现异步秒杀优化的资料请关注好代码网其它相关文章！

您可能有感兴趣的文章
Redis数据结构SortedSet的底层原理解析
redis主从哨兵模式如何实现一主二从
RediSearch加RedisJSON大于Elasticsearch的搜索存储引擎
Redis如何存储对象
Redis唯一ID生成器的如何实现