Redis Stream秒杀系统实现

// 前端调用
public Result seckillVoucher(Long voucherId) {
    // 生成唯一订单ID: 2024110500012345
    long orderId = redisIdWorker.nextId("order");
    // 执行Lua脚本进行原子操作
    Long result = stringRedisTemplate.execute(
            SECKILL_SCRIPT,
            Collections.emptyList(),
            "1001",  // 优惠券ID (iPhone特价券)
            "12345", // 用户ID
            "2024110500012345" // 订单ID
    );
    // 立即返回结果给用户
    if(result != 0){
        return Result.fail(result==1 ? "库存不足":"不能重复下单");
    }
    return Result.ok(2024110500012345L);
}

-- 参数：优惠券ID、用户ID、订单ID
local voucherId = ARGV[1]
local userId = ARGV[2]  
local orderId = ARGV[3]
-- 构建Redis Key
local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId
-- 1. 判断库存是否充足
local stock = redis.call('get', stockKey)
if tonumber(stock) <= 0 then
    return 1  -- 库存不足
end
-- 2. 判断用户是否已经下单 (set集合)
if redis.call('sismember', orderKey, userId) == 1 then
    return 2  -- 不能重复下单
end
-- 3. 扣减库存
redis.call('decr', stockKey)
-- 4. 记录用户购买记录
redis.call('sadd', orderKey, userId)
-- 5. 发送消息到Stream队列
redis.call('xadd', 'stream.order', '*',
    'voucherId', voucherId,
    'userId', userId, 
    'orderId', orderId
)
return 0  -- 成功

时间线:
10:00:00.000 - 用户A点击秒杀按钮
10:00:00.050 - Lua脚本执行:
                ✓ 库存检查: 库存1000 > 0
                ✓ 重复检查: 用户A未购买
                ✓ 库存-1 → 999
                ✓ 记录用户A到已购集合
                ✓ 发送消息到stream.order
10:00:00.100 - 返回订单ID: 2024110500012345
10:00:00.150 - 异步线程处理订单入库
10:00:01.000 - 订单创建完成

时间线:
10:00:00.200 - 用户B点击秒杀按钮  
10:00:00.250 - Lua脚本执行:
                ✓ 库存检查: 库存999 > 0
                ✗ 重复检查: 用户B已在已购集合中
                → 返回2 (不能重复下单)
10:00:00.300 - 前端显示:"不能重复下单"

时间线:
10:00:05.000 - 用户Z点击秒杀按钮
10:00:05.050 - Lua脚本执行:
                ✗ 库存检查: 库存0 <= 0
                → 返回1 (库存不足)
10:00:05.100 - 前端显示:"库存不足"

// VoucherOrderHandler - 订单处理线程
while(true){
    // 从消息队列读取订单
    List<MapRecord<String, Object, Object>> list = stringRedisTemplate
        .opsForStream()
        .read(
            Consumer.from("g1", "c1"),  // 消费者组g1，消费者c1
            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
            StreamOffset.create("stream.order", ReadOffset.lastConsumed()) // 读取新消息
        );
    if(!list.isEmpty()){
        MapRecord<String, Object, Object> record = list.get(0);
        Map<Object, Object> values = record.getValue();
        // 构建订单对象
        VoucherOrder order = new VoucherOrder();
        order.setId(Long.parseLong((String)values.get("orderId")));
        order.setUserId(Long.parseLong((String)values.get("userId")));
        order.setVoucherId(Long.parseLong((String)values.get("voucherId")));
        // 保存到数据库
        voucherOrderService.save(order);
        // 确认消息已处理
        stringRedisTemplate.opsForStream()
            .acknowledge("stream.order", "g1", record.getId());
        log.info("订单处理成功: {}", order.getId());
    }
}

private void handlePendingList() {
    while(true){
        try {
            // 读取未确认的消息 (处理异常情况)
            List<MapRecord<String, Object, Object>> list = stringRedisTemplate
                .opsForStream()
                .read(
                    Consumer.from("g1", "c1"),
                    StreamReadOptions.empty().count(1),
                    StreamOffset.create("stream.order", ReadOffset.from("0")) // 从pending-list读取
                );
            if(list.isEmpty()) break; // 没有异常消息
            MapRecord<String, Object, Object> record = list.get(0);
            // 重新处理订单...
            createVoucherOrder(voucherOrder);
            // 确认消息
            stringRedisTemplate.opsForStream()
                .acknowledge("stream.order", "g1", record.getId());
        } catch (Exception e) {
            // 处理失败，等待后重试
            Thread.sleep(20);
        }
    }
}

// 模拟10万用户同时秒杀
for(int i = 1; i <= 100000; i++){
    new Thread(() -> {
        Result result = seckillVoucher(1001L); // iPhone秒杀
        if(result.getCode() == 200){
            System.out.println("用户" + Thread.currentThread().getId() + "秒杀成功");
        } else {
            System.out.println("用户" + Thread.currentThread().getId() + "秒杀失败");
        }
    }).start();
}

seckill:stock:1001: "1000"     # 库存1000
seckill:order:1001: []         # 空集合
stream.order: []               # 空消息队列

seckill:stock:1001: "500"      # 库存剩余500
seckill:order:1001: ["12345", "12346", ...]  # 500个用户ID
stream.order: [消息1, 消息2, ...]            # 500条待处理消息

seckill:stock:1001: "0"        # 库存为0
seckill:order:1001: [1000个用户ID]          # 1000个购买用户
stream.order: []               # 消息全部处理完成