如何设计一个高并发的秒杀系统？超卖问题如何解决？

-- 原子扣减库存脚本
local stockKey = KEYS[1]
local userKey = KEYS[2]
local userId = ARGV[1]

-- 1. 判断用户是否已经抢购过
if redis.call('sismember', userKey, userId) == 1 then
    return -1 -- 已抢购
end

-- 2. 判断库存是否充足
local stock = tonumber(redis.call('get', stockKey))
if stock <= 0 then
    return 0 -- 库存不足
end

-- 3. 扣减库存并标记用户已抢购
redis.call('decr', stockKey)
redis.call('sadd', userKey, userId)
return 1 -- 抢购成功

-- 不用版本号，直接用库存数当条件，更简洁
UPDATE seckill_product 
SET stock = stock - 1 
WHERE id = #{productId} 
AND stock >= 1;  -- 核心兜底条件

/**
 * 原子性预扣减库存（Lua脚本实现）
 * @param goodsId 商品ID
 * @param userId 用户ID
 * @param limitCount 每人限购数量
 * @return 0-成功 1-库存不足 2-已达限购上限
 */
public int deductStock(Long goodsId, Long userId, int limitCount) {
    String luaScript = """
        local stockKey = KEYS[1]
        local userKey = KEYS[2]
        local limit = tonumber(ARGV[1])
        
        -- 1. 判断用户是否已达限购上限
        local userBuyCount = redis.call('HGET', userKey, ARGV[2])
        if userBuyCount and tonumber(userBuyCount) >= limit then
            return 2
        end
        
        -- 2. 判断库存是否充足
        local stock = redis.call('GET', stockKey)
        if not stock or tonumber(stock) <= 0 then
            return 1
        end
        
        -- 3. 扣减库存+记录用户购买数量
        redis.call('DECR', stockKey)
        redis.call('HINCRBY', userKey, ARGV[2], 1)
        return 0
    """;

    List<String> keys = Arrays.asList(
        "seckill:stock:" + goodsId,
        "seckill:user:" + goodsId
    );

    return (Integer) redisTemplate.execute(
        new DefaultRedisScript<>(luaScript, Integer.class),
        keys,
        limitCount,
        userId.toString()
    );
}

/**
 * 防止同一用户重复下单
 */
public Result seckill(Long goodsId, Long userId) {
    String lockKey = "seckill:lock:" + goodsId + ":" + userId;
    RLock lock = redissonClient.getLock(lockKey);

    try {
        // 尝试获取锁，最多等待0秒，锁默认30秒过期（看门狗自动续期）
        if (!lock.tryLock(0, TimeUnit.SECONDS)) {
            return Result.error("请勿重复提交请求");
        }

        // 执行库存预扣减
        int result = deductStock(goodsId, userId, 1);
        if (result == 1) {
            return Result.error("商品已售罄");
        }
        if (result == 2) {
            return Result.error("您已参与过本次秒杀");
        }

        // 生成订单ID，发送消息到Kafka
        String orderId = UUID.randomUUID().toString();
        kafkaTemplate.send("seckill-order-topic", 
            new SeckillMessage(orderId, goodsId, userId));

        return Result.success("排队成功，请等待订单生成");
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return Result.error("系统繁忙，请稍后再试");
    } finally {
        // 释放锁
        if (lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
}

/**
 * Kafka秒杀订单消费者（幂等性实现）
 */
@KafkaListener(topics = "seckill-order-topic", groupId = "seckill-group")
public void onMessage(ConsumerRecord<String, String> record) {
    SeckillMessage message = JSON.parseObject(record.value(), SeckillMessage.class);
    String orderId = message.getOrderId();

    // 1. 幂等性判断：如果订单已存在，直接返回
    if (orderService.existOrder(orderId)) {
        log.info("订单已存在，重复消费，orderId: {}", orderId);
        return;
    }

    // 2. 事务处理：创建订单+扣减数据库库存
    try {
        orderService.createOrderWithStockDeduct(message);
    } catch (Exception e) {
        log.error("订单创建失败，orderId: {}", orderId, e);
        // 发送到死信队列，人工处理
        kafkaTemplate.send("seckill-dlq-topic", record.value());
    }
}

-- 最终库存扣减（行锁实现，绝对不会超卖）
UPDATE seckill_goods 
SET stock_count = stock_count - 1,
    version = version + 1
WHERE goods_id = #{goodsId} 
  AND stock_count > 0 
  AND version = #{version};

@RestController
public class SeckillController {
    @Autowired
    private SeckillStockService stockService;
    @Autowired
    private RocketMQTemplate mqTemplate;
    @Autowired
    private StringRedisTemplate redis;

    // 秒杀接口
    @PostMapping("/seckill/exec")
    public Result exec(@RequestParam String itemId, 
                       @RequestParam String captcha,
                       HttpServletRequest request) {
        String uid = getLoginUserId(request);
        // 1. 验证码校验（防脚本）
        if (!checkCaptcha(uid, captcha)) {
            return Result.fail("验证码错误");
        }
        // 2. 用户级限流：每人只允许一次请求
        String limitKey = "seckill:limit:" + itemId + ":" + uid;
        Boolean limited = redis.opsForValue()
                .setIfAbsent(limitKey, "1", Duration.ofMinutes(1));
        if (Boolean.FALSE.equals(limited)) {
            return Result.fail("请勿重复提交");
        }
        // 3. 库存预扣
        boolean deducted = stockService.deductStock(itemId, 10); // 10个分片
        if (!deducted) {
            return Result.fail("已抢光");
        }
        // 4. 发送 MQ 消息，异步下单
        SeckillMessage msg = new SeckillMessage(uid, itemId);
        mqTemplate.asyncSend("seckill_order", msg, new SendCallback() {
            @Override public void onSuccess(SendResult result) {}
            @Override public void onException(Throwable e) {
                // 发送失败，补偿恢复库存 + 删除 limitKey
                // ... 异常处理逻辑，可发一条补偿消息
            }
        });
        // 5. 立即返回排队结果，前端轮询查结果
        return Result.ok("排队中，请稍候查看结果");
    }

    // 查询秒杀结果
    @GetMapping("/seckill/result")
    public Result queryResult(@RequestParam String itemId, HttpServletRequest req) {
        String uid = getLoginUserId(req);
        String resultKey = "seckill:result:" + itemId + ":" + uid;
        String result = redis.opsForValue().get(resultKey);
        if (result == null) {
            return Result.ok("排队中");
        }
        return Result.ok(result); // "success" 或 "fail"
    }
}