基于Redis的Stream结构作为消息队列，实现异步秒杀下单

柔光的暖阳◎ 2024-03-26 17:56 71阅读 0赞

**需求**：

*  创建一个Stream类型的消息队列，名为stream.orders
 *  修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId
 *  项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

**导入依懒**

<dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
                <exclusions>
                    <exclusion>
                        <artifactId>spring-data-redis</artifactId>
                        <groupId>org.springframework.data</groupId>
                    </exclusion>
                    <exclusion>
                        <artifactId>lettuce-core</artifactId>
                        <groupId>io.lettuce</groupId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.data</groupId>
                <artifactId>spring-data-redis</artifactId>
                <version>2.6.2</version>
            </dependency>
            <dependency>
                <groupId>io.lettuce</groupId>
                <artifactId>lettuce-core</artifactId>
                <version>6.1.6.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-pool2</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <scope>runtime</scope>
                <version>5.1.46</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>com.baomidou</groupId>
                <artifactId>mybatis-plus-boot-starter</artifactId>
                <version>3.4.3</version>
            </dependency>
            
            <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
                <version>5.7.17</version>
            </dependency>
           
            <dependency>
                <groupId>org.redisson</groupId>
                <artifactId>redisson</artifactId>
                <version>3.13.6</version>
            </dependency>
    
        </dependencies>

**编写Controller**

package com.hmdp.controller;
    
    
    import com.hmdp.dto.Result;
    import com.hmdp.service.IVoucherOrderService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    
    /**
     * @author 24412
     */
    @RestController
    @RequestMapping("/voucher-order")
    public class VoucherOrderController {
        @Autowired
        private IVoucherOrderService voucherOrderService;
        @PostMapping("seckill/{id}")
        public Result seckillVoucher(@PathVariable("id") Long voucherId) {
            return voucherOrderService.seckillVoucher(voucherId);
        }
    }

编写service

package com.hmdp.service.impl;
    
    import cn.hutool.core.bean.BeanUtil;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import com.hmdp.dto.Result;
    import com.hmdp.entity.VoucherOrder;
    import com.hmdp.mapper.VoucherOrderMapper;
    import com.hmdp.service.ISeckillVoucherService;
    import com.hmdp.service.IVoucherOrderService;
    import com.hmdp.utils.RedisIdWorker;
    import com.hmdp.utils.UserHolder;
    import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.RLock;
    import org.redisson.api.RedissonClient;
    import org.springframework.aop.framework.AopContext;
    import org.springframework.core.io.ClassPathResource;
    import org.springframework.data.redis.connection.stream.*;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.data.redis.core.script.DefaultRedisScript;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.Resource;
    import java.time.Duration;
    import java.util.Collections;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * @author 24412
     */
    @Service
    @Slf4j
    public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
        @Resource
        private ISeckillVoucherService deskillVoucherService;
    
    
        @Resource
        private StringRedisTemplate stringRedisTemplate;
    
        @Resource
        private RedissonClient redissonClient;
    
        /**
         *  代理对象
         */
        private IVoucherOrderService currentProxy;
    
        // Lua 脚本
        private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    
        static {
            SECKILL_SCRIPT = new DefaultRedisScript<>();
            // 设置 Lua 脚本
            SECKILL_SCRIPT.setLocation(new ClassPathResource("SeckillVoucher.lua"));
            SECKILL_SCRIPT.setResultType(Long.class);
        }
    
        /**
         *   异步处理线程池
         */
        private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    
        /**
         *  在当前类初始完毕后执行 VoucherOrderHandler 中的 run 方法
         */
    
        @PostConstruct
        public void init() {
            SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
        }
    
        /**
         *  项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单
         */
        public class VoucherOrderHandler implements Runnable {
            String queueName = "stream.orders";
            String groupName = "orderGroup";
            String consumerName = "ConsumerOne";
    
            @Override
            public void run() {
                while (true) {
                    try {
                        // 1. 获取消息队列中的订单信息
                        // XREAD GROUP orderGroup consumerOne COUNT 1 BLOCK 2000 STREAMS stream.orders >
                        // 队列 stream.orders、消费者组 orderGroup、消费者 consumerOne、每次读 1 条消息、阻塞时间 2s、从下一个未消费的消息开始。
                        List<MapRecord<String, Object, Object>> readingList = stringRedisTemplate.opsForStream().read(
                                Consumer.from(groupName, consumerName),
                                StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                                StreamOffset.create(queueName, ReadOffset.lastConsumed()));
    
                        // 2. 判断消息是否获取成功
                        if (readingList.isEmpty() || readingList == null) {
                            // 获取失败说明没有消息，则继续下一次循环
                            continue;
                        }
    
                        // 3. 解析消息中的订单信息
                        // MapRecord：String 代表 消息ID；两个 Object 代表 消息队列中的 Key-Value
                        MapRecord<String, Object, Object> record = readingList.get(0);
                        Map<Object, Object> recordValue = record.getValue();
                        VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(recordValue, new VoucherOrder(), true);
    
                        // 4. 获取成功则下单
                        handleVoucherOrder(voucherOrder);
    
                        // 5. 确认消息 XACK stream.orders orderGroup id
                        stringRedisTemplate.opsForStream().acknowledge(groupName, consumerName, record.getId());
                    } catch (Exception e) {
                        log.error("订单处理异常", e);
                        handlePendingMessages();
                    }
                }
            }
    
             /**
             *  处理 pending-list 中的消息
             */
            private void handlePendingMessages() {
                while (true) {
                    try {
                        // 1. 获取 pending-list 中的订单信息
                        // XREAD GROUP orderGroup consumerOne COUNT 1 STREAM stream.orders 0
                        List<MapRecord<String, Object, Object>> readingList = stringRedisTemplate.opsForStream().read(
                                Consumer.from(groupName, consumerName),
                                StreamReadOptions.empty().count(1),
                                StreamOffset.create(queueName, ReadOffset.from("0"))
                        );
    
                        // 2. 判断消息是否获取成功
                        if (readingList.isEmpty() || readingList == null) {
                            // 获取失败 pending-list 中没有异常消息，结束循环
                            break;
                        }
    
                        // 3. 解析消息中的订单信息并下单
                        MapRecord<String, Object, Object> record = readingList.get(0);
                        Map<Object, Object> recordValue = record.getValue();
                        VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(recordValue, new VoucherOrder(), true);
                        handleVoucherOrder(voucherOrder);
    
                        // 4. XACK
                        stringRedisTemplate.opsForStream().acknowledge(queueName, groupName, record.getId());
                    } catch (Exception e) {
                        log.error("订单处理异常（pending-list）", e);
                        try {
                            // 稍微休眠一下再进行循环
                            Thread.sleep(20);
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                    }
                }
            }
        }
    
    
        /**
         * 创建代金券订单
         * @param voucherOrder
         * @return
         */
        private void handleVoucherOrder(VoucherOrder voucherOrder) {
            Long userId = voucherOrder.getUserId();
            RLock lock = redissonClient.getLock("lock:order:" + userId);
            boolean isLocked = lock.tryLock();
            if (!isLocked) {
                log.error("不允许重复下单！");
                return;
            }
            try {
                // 该方法非主线程调用，代理对象需要在主线程中获取。
                currentProxy.createVoucherOrder(voucherOrder);
            } finally {
                lock.unlock();
            }
        }
    
    
        /**
         * 秒杀代金券
         * @param voucherId
         * @return
         */
        @Override
        public Result seckillVoucher(Long voucherId) {
            // 1. 执行 Lua 脚本（有购买资格：向 stream.orders 中添加消息，内容包括 voucherId、userId、orderId）
            Long userId = UserHolder.getUser().getId();
            long orderId = RedisIdWorker.nextId("order");
            Long executeResult = stringRedisTemplate.execute(
                    SECKILL_SCRIPT,
                    Collections.emptyList(),
                    voucherId.toString(), userId.toString(), String.valueOf(orderId)
            );
    
            // 2. Lua 脚本的执行结果0则有购买资格
            int result = executeResult.intValue();
            if (result != 0) {
                return Result.fail(result == 1 ? "库存不足！" : "请勿重复下单！");
            }
    
            // 4. 获取代理对象
            currentProxy = (IVoucherOrderService) AopContext.currentProxy();
    
            // 5. 返回订单号（告诉用户下单成功，业务结束；执行异步下单操作数据库）
            return Result.ok(orderId);
        }
    
        /**
         * 异步下单
         * @param voucherOrder
         */
        @Transactional
        @Override
        public void createVoucherOrder(VoucherOrder voucherOrder) {
            Long userId = voucherOrder.getUserId();
            // 1. 一人一单
            Integer count = query()
                    .eq("voucher_id", voucherOrder.getVoucherId())
                    .eq("user_id", userId)
                    .count();
            if (count > 0) {
                log.error("不可重复下单！");
                return;
            }
    
            // 2. 减扣库存
            boolean isAccomplished = deskillVoucherService.update()
                    .setSql("stock = stock - 1")
                    .eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0)
                    .update();
            if (!isAccomplished) {
                log.error("库存不足！");
                return;
            }
    
            // 3. 下单
            boolean isSaved = save(voucherOrder);
            if (!isSaved) {
                log.error("下单失败！");
                return;
            }
        }
    }

**lua脚本**

---
    --- Created by sun.
    --- DateTime: 2022/10/18 17:10
    ---
    
    local voucherId = ARGV[1]
    local userId = ARGV[2]
    local orderId = ARGV[3]
    
    local stockKey = "seckill:stock:" .. voucherId
    local orderKey = "seckill:order:" .. voucherId
    
    -- 判断库存是否充足（不足，返回 1）
    if (tonumber(redis.call('GET', stockKey)) <= 0) then
        return 1;
    end;
    
    -- 判断用户是否下单（重复下单，返回 2）
    if (redis.call('SISMEMBER', orderKey, userId) == 1) then
        return 2;
    end;
    
    -- 下单成功：扣减库存、保存用户。
    redis.call('INCRBY', stockKey, -1);
    redis.call('SADD', orderKey, userId);
    -- 发送消息到 stream.orders 队列中（*：消息的唯一ID 由 Redis 自动生成）：XADD stream.orders * key field ...
    redis.call('XADD', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId);
    return 0;

**获取用户id工具类**

package com.hmdp.utils;
    
    import com.hmdp.dto.UserDTO;
    
    /**
     * @author 24412
     */
    public class UserHolder {
        private static final ThreadLocal<UserDTO> tl = new ThreadLocal<>();
    
        public static void saveUser(UserDTO user){
            tl.set(user);
        }
    
        public static UserDTO getUser(){
            return tl.get();
        }
    
        public static void removeUser(){
            tl.remove();
        }
    }

**配置Redisson客户端**

package com.hmdp.config;
    
    import org.redisson.Redisson;
    import org.redisson.api.RedissonClient;
    import org.redisson.config.Config;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * @BelongsProject: hm-dianping
     * @BelongsPackage: com.hmdp.config
     * @Author: wang fei
     * @CreateTime: 2023-02-14  16:09
     * @Description: TODO 配置Redisson客户端
     * @Version: 1.0
     */
    @Configuration
    public class RedissonConfig {
        @Bean
        public RedissonClient redisson() {
            // 配置类
            Config config = new Config();
            // 添加 Redis 地址：此处是单节点地址，也可以通过 config.useClusterServers() 添加集群地址
            config.useSingleServer().setAddress("redis://xxxxxx").setPassword("wangf");
            // 创建客户端
            return Redisson.create(config);
    
        }
    }

**VoucherOrder**

package com.hmdp.entity;
    
    import com.baomidou.mybatisplus.annotation.IdType;
    import com.baomidou.mybatisplus.annotation.TableId;
    import com.baomidou.mybatisplus.annotation.TableName;
    import lombok.Data;
    import lombok.EqualsAndHashCode;
    import lombok.experimental.Accessors;
    
    import java.io.Serializable;
    import java.time.LocalDateTime;
    
    
    /**
     * @author 24412
     */
    @Data
    @EqualsAndHashCode(callSuper = false)
    @Accessors(chain = true)
    @TableName("tb_voucher_order")
    public class VoucherOrder implements Serializable {
    
        private static final long serialVersionUID = 1L;
    
        /**
         * 主键
         */
        @TableId(value = "id", type = IdType.INPUT)
        private Long id;
    
        /**
         * 下单的用户id
         */
        private Long userId;
    
        /**
         * 购买的代金券id
         */
        private Long voucherId;
    
        /**
         * 支付方式 1：余额支付；2：支付宝；3：微信
         */
        private Integer payType;
    
        /**
         * 订单状态，1：未支付；2：已支付；3：已核销；4：已取消；5：退款中；6：已退款
         */
        private Integer status;
    
        /**
         * 下单时间
         */
        private LocalDateTime createTime;
    
        /**
         * 支付时间
         */
        private LocalDateTime payTime;
    
        /**
         * 核销时间
         */
        private LocalDateTime useTime;
    
        /**
         * 退款时间
         */
        private LocalDateTime refundTime;
    
        /**
         * 更新时间
         */
        private LocalDateTime updateTime;
    
    
    }