Redis消息队列实现异步秒杀

发布于:2022-11-09 ⋅ 阅读:(15) ⋅ 点赞:(0) ⋅ 评论:(0)

目录

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