-- 參數(shù)：優(yōu)惠券ID、用戶ID、訂單ID
local voucherId = ARGV[1]
local userId = ARGV[2]
local orderId = ARGV[3]
-- 庫(kù)存Key和訂單Key
local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId
-- 判斷庫(kù)存是否充足
if (tonumber(redis.call('GET', stockKey)) <= 0 then
    return 1 -- 庫(kù)存不足
end
-- 判斷用戶是否已下單
if (redis.call('SISMEMBER', orderKey, userId) == 1 then
    return 2 -- 用戶已下單
end
-- 扣減庫(kù)存并記錄訂單
redis.call('DECR', stockKey)
redis.call('SADD', orderKey, userId)
-- 將訂單信息發(fā)送到消息隊(duì)列
redis.call('XADD', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0 -- 成功

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    private ISeckillVoucherService seckillVoucherService;
    @Resource
    private RedisIdWorker redisIdWorker;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonClient redissonClient;
    /**
     * 當(dāng)前類初始化完畢就立馬執(zhí)行該方法
     */
    @PostConstruct
    private void init() {
        // 創(chuàng)建消息隊(duì)列
        DefaultRedisScript<Long> mqScript = new DefaultRedisScript<>();
        mqScript.setLocation(new ClassPathResource("lua/stream-mq.lua"));
        mqScript.setResultType(Long.class);
        Long result = null;
        try {
            result = stringRedisTemplate.execute(mqScript,
                    Collections.emptyList(),
                    QUEUE_NAME,
                    GROUP_NAME);
        } catch (Exception e) {
            log.error("隊(duì)列創(chuàng)建失敗", e);
            return;
        }
        int r = result.intValue();
        String info = r == 1 ? "隊(duì)列創(chuàng)建成功" : "隊(duì)列已存在";
        log.debug(info);
        // 執(zhí)行線程任務(wù)
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }
    /**
     * 線程池
     */
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    /**
     * 隊(duì)列名
     */
    private static final String QUEUE_NAME = "stream.orders";
    /**
     * 組名
     */
    private static final String GROUP_NAME = "g1";
    /**
     * 線程任務(wù): 不斷從消息隊(duì)列中獲取訂單
     */
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    // 1、從消息隊(duì)列中獲取訂單信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 1000 STREAMS streams.order >
                    List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(
                            Consumer.from(GROUP_NAME, "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),
                            StreamOffset.create(QUEUE_NAME, ReadOffset.lastConsumed())
                    );
                    // 2、判斷消息獲取是否成功
                    if (messageList == null || messageList.isEmpty()) {
                        // 2.1 消息獲取失敗，說(shuō)明沒(méi)有消息，進(jìn)入下一次循環(huán)獲取消息
                        continue;
                    }
                    // 3、消息獲取成功，可以下單
                    // 將消息轉(zhuǎn)成VoucherOrder對(duì)象
                    MapRecord<String, Object, Object> record = messageList.get(0);
                    Map<Object, Object> messageMap = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    // 4、ACK確認(rèn) SACK stream.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, GROUP_NAME, record.getId());
                } catch (Exception e) {
                    log.error("處理訂單異常", e);
                    // 處理異常消息
                    handlePendingList();
                }
            }
        }
    }
    private void handlePendingList() {
        while (true) {
            try {
                // 1、從pendingList中獲取訂單信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 1000 STREAMS streams.order 0
                List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(
                        Consumer.from(GROUP_NAME, "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),
                        StreamOffset.create(QUEUE_NAME, ReadOffset.from("0"))
                );
                // 2、判斷pendingList中是否有效性
                if (messageList == null || messageList.isEmpty()) {
                    // 2.1 pendingList中沒(méi)有消息，直接結(jié)束循環(huán)
                    break;
                }
                // 3、pendingList中有消息
                // 將消息轉(zhuǎn)成VoucherOrder對(duì)象
                MapRecord<String, Object, Object> record = messageList.get(0);
                Map<Object, Object> messageMap = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
                handleVoucherOrder(voucherOrder);
                // 4、ACK確認(rèn) SACK stream.orders g1 id
                stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, GROUP_NAME, record.getId());
            } catch (Exception e) {
                log.error("處理訂單異常", e);
                // 這里不用調(diào)自己，直接就進(jìn)入下一次循環(huán)，再?gòu)膒endingList中取，這里只需要休眠一下，防止獲取消息太頻繁
                try {
                    Thread.sleep(20);
                } catch (InterruptedException ex) {
                    log.error("線程休眠異常", ex);
                }
            }
        }
    }
    /**
     * 創(chuàng)建訂單
     *
     * @param voucherOrder
     */
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        RLock lock = redissonClient.getLock(RedisConstants.LOCK_ORDER_KEY + userId);
        boolean isLock = lock.tryLock();
        if (!isLock) {
            // 索取鎖失敗，重試或者直接拋異常（這個(gè)業(yè)務(wù)是一人一單，所以直接返回失敗信息）
            log.error("一人只能下一單");
            return;
        }
        try {
            // 創(chuàng)建訂單（使用代理對(duì)象調(diào)用，是為了確保事務(wù)生效）
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            lock.unlock();
        }
    }
    /**
     * 加載 判斷秒殺券庫(kù)存是否充足 并且 判斷用戶是否已下單 的Lua腳本
     */
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/stream-seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }
    /**
     * VoucherOrderServiceImpl類的代理對(duì)象
     * 將代理對(duì)象的作用域進(jìn)行提升，方面子線程取用
     */
    private IVoucherOrderService proxy;
    /**
     * 搶購(gòu)秒殺券
     *
     * @param voucherId
     * @return
     */
    @Transactional
    @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId = ThreadLocalUtls.getUser().getId();
        long orderId = redisIdWorker.nextId(SECKILL_VOUCHER_ORDER);
        // 1、執(zhí)行Lua腳本，判斷用戶是否具有秒殺資格
        Long result = null;
        try {
            result = stringRedisTemplate.execute(
                    SECKILL_SCRIPT,
                    Collections.emptyList(),
                    voucherId.toString(),
                    userId.toString(),
                    String.valueOf(orderId)
            );
        } catch (Exception e) {
            log.error("Lua腳本執(zhí)行失敗");
            throw new RuntimeException(e);
        }
        if (result != null && !result.equals(0L)) {
            // result為1表示庫(kù)存不足，result為2表示用戶已下單
            int r = result.intValue();
            return Result.fail(r == 2 ? "不能重復(fù)下單" : "庫(kù)存不足");
        }
        // 2、result為0，下單成功，直接返回ok
        // 索取鎖成功，創(chuàng)建代理對(duì)象，使用代理對(duì)象調(diào)用第三方事務(wù)方法， 防止事務(wù)失效
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        this.proxy = proxy;
        return Result.ok();
    }
    /**
     * 創(chuàng)建訂單
     *
     * @param voucherOrder
     * @return
     */
    @Transactional
    @Override
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();
        // 1、判斷當(dāng)前用戶是否是第一單
        int count = this.count(new LambdaQueryWrapper<VoucherOrder>()
                .eq(VoucherOrder::getUserId, userId));
        if (count >= 1) {
            // 當(dāng)前用戶不是第一單
            log.error("當(dāng)前用戶不是第一單");
            return;
        }
        // 2、用戶是第一單，可以下單，秒殺券庫(kù)存數(shù)量減一
        boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
                .eq(SeckillVoucher::getVoucherId, voucherId)
                .gt(SeckillVoucher::getStock, 0)
                .setSql("stock = stock -1"));
        if (!flag) {
            throw new RuntimeException("秒殺券扣減失敗");
        }
        // 3、將訂單保存到數(shù)據(jù)庫(kù)
        flag = this.save(voucherOrder);
        if (!flag) {
            throw new RuntimeException("創(chuàng)建秒殺券訂單失敗");
        }
    }
}

-- 優(yōu)惠券id
local voucherId = ARGV[1];
-- 用戶id
local userId = ARGV[2];
local stockKey = 'seckill:stock:' .. voucherId;
local orderKey = 'seckill:order:' .. voucherId;
local stock = redis.call('GET', stockKey);
if (tonumber(stock) <= 0) then
    return 1;
end
if (redis.call('SISMEMBER', orderKey, userId) == 1) then
    return 2;
end
redis.call('INCRBY', stockKey, -1);
redis.call('SADD', orderKey, userId);
return 0;

@PostConstruct
private void init() {
    DefaultRedisScript<Long> mqScript = new DefaultRedisScript<>();
    mqScript.setLocation(new ClassPathResource("lua/stream-mq.lua"));
    mqScript.setResultType(Long.class);
    Long result = stringRedisTemplate.execute(mqScript, Collections.emptyList(), QUEUE_NAME, GROUP_NAME);
    if (result == 1) {
        log.debug("隊(duì)列創(chuàng)建成功");
    } else {
        log.debug("隊(duì)列已存在");
    }
    SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}

@Override
public Result seckillVoucher(Long voucherId) {
    Long userId = ThreadLocalUtls.getUser().getId();
    long orderId = redisIdWorker.nextId(SECKILL_VOUCHER_ORDER);
    Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), 
                                            voucherId.toString(), userId.toString(), String.valueOf(orderId));
    if (result != 0) {
        return Result.fail(result == 2 ? "不能重復(fù)下單" : "庫(kù)存不足");
    }
    IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    this.proxy = proxy;
    return Result.ok();
}

private class VoucherOrderHandler implements Runnable {
    @Override
    public void run() {
        while (true) {
            try {
                List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(
                    Consumer.from(GROUP_NAME, "c1"),
                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),
                    StreamOffset.create(QUEUE_NAME, ReadOffset.lastConsumed())
                );
                if (messageList == null || messageList.isEmpty()) {
                    continue;
                }
                MapRecord<String, Object, Object> record = messageList.get(0);
                Map<Object, Object> messageMap = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
                handleVoucherOrder(voucherOrder);
                stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, GROUP_NAME, record.getId());
            } catch (Exception e) {
                log.error("處理訂單異常", e);
                handlePendingList();
            }
        }
    }
}

private void handleVoucherOrder(VoucherOrder voucherOrder) {
    Long userId = voucherOrder.getUserId();
    RLock lock = redissonClient.getLock(RedisConstants.LOCK_ORDER_KEY + userId);
    boolean isLock = lock.tryLock();
    if (!isLock) {
        log.error("一人只能下一單");
        return;
    }
    try {
        proxy.createVoucherOrder(voucherOrder);
    } finally {
        lock.unlock();
    }
}

@Override
public void createVoucherOrder(VoucherOrder voucherOrder) {
    Long userId = voucherOrder.getUserId();
    Long voucherId = voucherOrder.getVoucherId();
    int count = this.count(new LambdaQueryWrapper<VoucherOrder>().eq(VoucherOrder::getUserId, userId));
    if (count >= 1) {
        log.error("當(dāng)前用戶不是第一單");
        return;
    }
    boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
        .eq(SeckillVoucher::getVoucherId, voucherId)
        .gt(SeckillVoucher::getStock, 0)
        .setSql("stock = stock -1"));
    if (!flag) {
        throw new RuntimeException("秒殺券扣減失敗");
    }
    flag = this.save(voucherOrder);
    if (!flag) {
        throw new RuntimeException("創(chuàng)建秒殺券訂單失敗");
    }
}