Redis消息队列实现异步秒杀

目录

1、概述

2、基于Stream的消息队列-单消费者

3、基于Stream的消息队列-消费者组

4、基于Stream的消息队列实现异步秒杀

1、概述

1、消息队列（Message Queue），字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

• 消息队列：存储和管理消息，也被称为消息代理（Message Broker）
• 生产者：发送消息到消息队列
• 消费者：从消息队列获取消息并处理消息

2、Redis提供了三种不同的方式来实现消息队列：

• list结构：基于List结构模拟消息队列

1. 优点：

• 利用Redis存储，不受限于JVM内存上限
• 基于Redis的持久化机制，数据安全性有保证
• 可以满足消息有序性

1. 缺点：

• 无法避免消息丢失
• 只支持单消费者
• PubSub：发布订阅消息模型

1. 优点：

• 采用发布订阅模型，支持多生产、多消费

1. 缺点：

• 不支持数据持久化
• 无法避免消息丢失
• 消息堆积有上限，超出时数据丢失
• Stream：比较完善的消息队列模型

2、基于Stream的消息队列-单消费者

1、命令

Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

发送消息的命令：

读取消息的命令：

2、Stream类型的消息队列的XREAD命令特点：

• 消息可回溯（即可重复读取）
• 一个消息可以被多个消费者读取
• 可以阻塞读取
• 有消息漏读的风险

3、基于Stream的消息队列-消费者组

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

1. 创建消费者组：

key：队列名称

groupName：消费者组名称

ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息

MKSTREAM：队列不存在时自动创建队列

1. 从消费者组读取消息：

group：消费组名称

consumer：消费者名称，如果消费者不存在，会自动创建一个消费者

count：本次查询的最大数量

BLOCK milliseconds：当没有消息时最长等待时间

NOACK：无需手动ACK，获取到消息后自动确认

STREAMS key：指定队列名称

ID：获取消息的起始ID：

">"：从下一个未消费的消息开始

其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从list中的第一个消息开始

1. Stream类型的消息队列的XREADGROUP命令特点：

• 消息可回溯
• 可以多消费者争抢消息，加快消费速度
• 可以阻塞读取
• 没有消息漏读的风险
• 有消息确认机制，保证消息至少被消费一次

4、基于Stream的消息队列实现异步秒杀

1、实现思路

1. 创建一个Stream类型的消息队列，名为stream.orders

   

2. 修改之前的秒杀下单Lua脚本，判断有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId
3. 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

2、代码实现

--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,这里的value为下单的用户ID
local orderKey = 'seckill:order:' .. voucherId

--3.脚本业务
--3.1判断库存是否充足 get stockKey
if(tonumber(redis.call('get',stockKey)) <= 0) then
    --库存不足，返回1
    return 1
end
--3.2判断用户是否已经购买过 sismember orderKey userId
if(redis.call('sismember',orderKey,userId) == 1) then
    --存在，则该用户重复下单，返回2
    return 2
end
--3.3扣减库存 incrby stockKey -1
redis.call('incrby',stockKey,-1)
--3.4生成订单(将用户id添加到订单的set集合中)
redis.call('sadd',orderKey,userId)
--3.5将订单信息发送到消息队列中  XADD stream.orders * k1 v1 k2 v2 k3 v3
redis.call('xadd','stream.orders','*','userId',userId,'voucherId',voucherId,'id',orderId)
return 0

/**
 *  优惠券订单模块
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    @Autowired
    private RedisIDWorker redisIDWorker;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    //初始化脚本对象,用于加载Lua脚本  这里定义为static是为了在类加载时就初始化该脚本对象，并且只会初始化一次，提高IO性能
    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 ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    //在当前类初始化完毕后就执行该方法
    @PostConstruct
    private void init(){
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }
    //创建线程任务--从消息队列中读取订单信息并下单
    private class VoucherOrderHandler implements Runnable{
        //消息队列名称
        private String queueName = "stream.orders";
        @Override
        public void run() {
            while (true){
                try {
                    //从消息队列中读取消息  XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );
                    //判断是否读取到消息
                    if (list == null || list.isEmpty()) {
                        //读取失败，说明没有消息，重新读取
                        continue;
                    }
                    //读取成功，解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> map = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(map, new VoucherOrder(),true);
                    //下单
                    handleVoucherOrder(voucherOrder);
                    //ACK确认  XACK stream.orders g1 消息id
                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常",e);
                    //处理消息时出现异常，则读取pending-list中的未确认的消息
                    handlePendingList();
                }
            }
        }

        //读取pending-list中已读取但未确认的消息  XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders 0
        private void handlePendingList() {
            while (true){
                try {
                    //从pending-list中读取消息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );
                    //判断是否读取到消息
                    if (list == null || list.isEmpty()) {
                        //读取失败，说明pending-list中没有异常消息，结束循环
                        break;
                    }
                    //读取成功，解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> map = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(map, new VoucherOrder(),true);
                    //下单
                    handleVoucherOrder(voucherOrder);
                    //ACK确认
                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",record.getId());
                } catch (Exception e) {
                    log.error("处理pending-list订单异常",e);
                }
            }
        }
    }

    //创建代理对象
    private IVoucherOrderService proxy;

    /**
     * 创建订单到数据库,独立线程会执行此方法
     * @param voucherOrder
     */
    public void handleVoucherOrder(VoucherOrder voucherOrder){
        //获取用户ID  这里不能使用ThreadLocal获取用户信息，因为此时是新的独立线程在执行该方法
        Long userId = voucherOrder.getUserId();
        //创建锁对象
        SimpleRedisLock redisLock = new SimpleRedisLock("order"+userId.toString(),stringRedisTemplate);
        //尝试获取锁
        boolean isLock = redisLock.tryLock(2);
        //判断是否获取到
        if (!isLock){
            //获取锁失败，返回错误
            log.error("不允许重复下单");
            return;
        }
        try {
            //创建订单到数据库
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            //释放锁
            redisLock.unLock();
        }
    }

    /**
     * 优惠券秒杀下单功能：基于Redis消息队列实现异步秒杀
     * @param voucherId
     * @return
     */
    @Override
    public Result seckillVoucher(Long voucherId) {
        //获取当前用户
        Long userId = UserHolder.getUser().getId();
        //订单id
        long orderId = redisIDWorker.nextId("order");
        //执行Lua脚本
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(),String.valueOf(orderId));
        //判断结果是否为0
        int r = result.intValue();
        if (r != 0){
            //不为0，返回异常信息
            return Result.fail(r==1 ? "库存不足" : "不能重复购买");
        }
        //为0，则有购买资格
        //获取代理对象（事务）    保证事务提交之后再释放锁
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        //返回订单id给前端
        return Result.ok(orderId);
    }

    /**
     * 创建优惠券订单
     * @param voucherOrder
     */
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        //实现一人一单
        Long userId = voucherOrder.getUserId();
        //查询优惠券订单
        int count = query().eq("voucher_id", voucherOrder.getVoucherId()).eq("user_id", userId).count();
        //判断订单是否存在
        if (count>0){
            //该用户已经购买过了
            log.error("不能重复购买");
            return;
        }

        //扣减库存 update tb_seckill_voucher set stock=stock-1 where voucher_id=? and stock>0
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock -1")
                .eq("voucher_id", voucherOrder.getVoucherId())
                .gt("stock",0)
                .update();
        //判断扣减是否成功
        if (!success){
            log.error("库存不足");
            return;
        }
        //新增订单
        save(voucherOrder);
    }
}

3、总结

通过Redis的Stream类型消息队列实现异步秒杀的功能，利用消费者组的方式将多个消费者划分为同一个组，监听同一个队列，利用消息分流提高了消息消费的速度，减少消息堆积；利用消息标示确保每个消息都能被消费，解决了消息遗漏的问题；利用消息确认机制保证了每个消息都能至少被处理一次，成功地解决了基于阻塞队列带来的内存限制问题和数据安全问题。