local voucherId = ARGV[1]; -- 優(yōu)惠劵id
local userId = ARGV[2]; -- 用戶id

-- 庫存key
local stockKey = 'seckill:stock' .. voucherId; -- 拼接
-- 訂單key
local stockKey = 'seckill:stock' .. voucherId; -- 拼接
-- 判斷庫存是否充足
if(tonumber(redis.call('get',stockKey) <= 0)) then
    -- 庫存不足，返回1
    return 1;
end;
-- 判斷用戶是否下單
if(redis.call('sismember',orderKey,userId)) then
    -- 存在，說明重復(fù)下單，返回2
    return 2;
end
-- 扣減庫存 incrby stockKey -1
redis.call('incrby',stockKey,-1);
-- 下單（保存用戶） sadd orderKey userId
redis.call('sadd',orderKey,userId);
return 0;

@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonClient redissonClient;
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT; // 泛型內(nèi)填入返回值類型
    static { // 靜態(tài)屬性要使用靜態(tài)代碼塊進(jìn)行初始化
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setResultType(Long.class);
        SECKILL_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    }
    public Result seckillVoucherMax(Long voucherId) {
        // 獲取用戶信息
        Long userId = UserHolder.getUser().getId();
        // 1.執(zhí)行Lua腳本來判斷用戶資格
        Long result = stringRedisTemplate.execute(
                            SECKILL_SCRIPT,
                            Collections.emptyList(), // Lua無需接受key
                            voucherId.toString(),
                            userId.toString()
                        );
        // 2.判斷結(jié)果是否為0
        int r = result.intValue();
        if(r != 0) {
            // 不為0代表無資格購買
            return Result.fail(r == 1 ? "庫存不足" : "不能重復(fù)下單");
        }
        // 3.有購買資格則將下單信息保存到阻塞隊列中
        // ... 
        return Result.ok();
    }

}

@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonClient redissonClient;
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT; // 泛型內(nèi)填入返回值類型
    static { // 靜態(tài)屬性要使用靜態(tài)代碼塊進(jìn)行初始化
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setResultType(Long.class);
        SECKILL_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    }
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024); // 創(chuàng)建阻塞隊列
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();  // 創(chuàng)建線程池
    // 讓大類在開始初始化時就能夠執(zhí)行線程任務(wù)
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderTask());
    }
    // 創(chuàng)建線程任務(wù)
    private class VoucherOrderTask implements Runnable {
        @Override
        public void run() {
            while(true){
                try {
                    // 獲取隊列中的訂單信息
                    VoucherOrder voucherOrder = orderTasks.take();// 取出頭部信息
                    // 創(chuàng)建訂單
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    log.error("處理訂單異常",e);
                }
            }
        }
    }
    // 創(chuàng)建訂單
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        RLock lock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId().toString());
        boolean isLock = lock.tryLock();
        // 判斷是否獲取鎖成功
        if (!isLock) {
            // 獲取鎖失敗，返回錯誤或重試
            log.error("不允許重復(fù)下單");
            return ;
        }
        try {
            proxy.createVoucherOrderMax(voucherOrder);
        } finally {
            lock.unlock();
        }
    }
    @Override
    public void createVoucherOrderMax(VoucherOrder voucherOrder) {
        // 一人一單
        Long userId = voucherOrder.getUserId();
        // 查詢訂單
        int count = query().eq("user_id",userId).eq("voucher_id", voucherOrder.getVoucherId()).count();
        // 判斷是否存在
        if(count > 0){
            // 用戶已經(jīng)購買過
            log.error("用戶已經(jīng)購買過");
            return ;
        }
        // CAS改進(jìn)：將庫存判斷改成stock > 0以此來提高性能
        boolean success = seckillVoucherService.update()
                .setSql("stock= stock -1") // set stock = stock - 1
                .eq("voucher_id", voucherOrder.getVoucherId()).eq("stock",0) // where id = ? and stock > 0
                .update();
        if (!success) {
            //扣減庫存
            log.error("庫存不足！");
            return ;
        }
        //6.創(chuàng)建訂單
        save(voucherOrder);
    }
    private IVoucherOrderService proxy; // 代理對象
    public Result seckillVoucherMax(Long voucherId) {
        // 獲取用戶信息
        Long userId = UserHolder.getUser().getId();
        // 1.執(zhí)行Lua腳本來判斷用戶資格
        Long result = stringRedisTemplate.execute(
                            SECKILL_SCRIPT,
                            Collections.emptyList(), // Lua無需接受key
                            voucherId.toString(),
                            userId.toString()
                        );
        // 2.判斷結(jié)果是否為0
        int r = result.intValue();
        if(r != 0) {
            // 不為0代表無資格購買
            return Result.fail(r == 1 ? "庫存不足" : "不能重復(fù)下單");
        }
        // 3.有購買資格則將下單信息保存到阻塞隊列中
        Long orderId = redisIdWorker.nextId("order");
        // 創(chuàng)建訂單
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        // 放入阻塞隊列
        orderTasks.add(voucherOrder);
        // 4.獲取代理對象(線程異步執(zhí)行，需要手動在方法內(nèi)獲取)
        proxy = (IVoucherOrderService)AopContext.currentProxy(); // 獲取當(dāng)前類的代理對象  (需要引入aspectjweaver依賴，并且在實(shí)現(xiàn)類加入@EnableAspectJAutoProxy(exposeProxy = true)以此來暴露代理對象)
        return Result.ok();
    }

}

private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

VoucherOrder voucherOrder = orderTasks.take();
handleVoucherOrder(voucherOrder);

private class VoucherOrderTask implements Runnable {
    @Override
    public void run() {
        while (true) {
            try {
                // 獲取隊列中的訂單信息
                VoucherOrder voucherOrder = orderTasks.take(); // 獲取訂單
                // 創(chuàng)建訂單
                handleVoucherOrder(voucherOrder);
            } catch (Exception e) {
                log.error("處理訂單異常", e);
            }
        }
    }
}

@PostConstruct
private void init() {
    SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderTask());
}

import redis.clients.jedis.Jedis;

public class ListProducer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379); // 連接 Redis
        String queueName = "myQueue";

        // 發(fā)送消息
        for (int i = 1; i <= 5; i++) {
            String message = "Message " + i;
            jedis.lpush(queueName, message); // 將消息推入隊列
            System.out.println("Sent: " + message);
        }

        jedis.close(); // 關(guān)閉連接
    }
}

import redis.clients.jedis.Jedis;

public class ListConsumer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379); // 連接 Redis
        String queueName = "myQueue";

        while (true) {
            // 阻塞獲取消息，超時時間為 0（無限等待）
            var result = jedis.brpop(0, queueName);
            String message = result.get(1); // 獲取消息內(nèi)容
            System.out.println("Received: " + message);
        }
    }
}

import redis.clients.jedis.Jedis;

public class PubSubProducer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379); // 連接 Redis
        String channelName = "myChannel";

        // 發(fā)布消息
        for (int i = 1; i <= 5; i++) {
            String message = "Message " + i;
            jedis.publish(channelName, message); // 發(fā)布消息到頻道
            System.out.println("Published: " + message);
        }

        jedis.close(); // 關(guān)閉連接
    }
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class PubSubConsumer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379); // 連接 Redis
        String channelName = "myChannel";

        // 創(chuàng)建訂閱者
        JedisPubSub subscriber = new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                System.out.println("Received: " + message);
            }
        };

        // 訂閱頻道
        jedis.subscribe(subscriber, channelName);
    }
}

XGROUP CREATE mystream group1 $ MKSTREAM
XGROUP CREATE mystream group2 $ MKSTREAM

@Configuration
public class RedisStreamConfig {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String STREAM_KEY = "mystream";
    private static final String GROUP_NAME = "mygroup";

    @PostConstruct
    public void init() {
        // 檢查消費(fèi)者組是否存在，若不存在則創(chuàng)建
        try {
            // 如果消費(fèi)者組不存在則會拋出異常，我們捕獲異常進(jìn)行創(chuàng)建
            redisTemplate.opsForStream().groups(STREAM_KEY);
        } catch (Exception e) {
            // 創(chuàng)建消費(fèi)者組，起始位置為 $ 表示從末尾開始消費(fèi)新消息
            redisTemplate.opsForStream().createGroup(STREAM_KEY, GROUP_NAME);
        }
    }
}

@Service  
public class RedisStreamProducerService {  // 定義生產(chǎn)者服務(wù)類 RedisStreamProducerService

    private static final String STREAM_KEY = "mystream";  // 定義 Redis Stream 的名稱，這里指定隊列名為 "mystream"

    @Autowired  
    private StringRedisTemplate redisTemplate;

    public void sendMessage(String content) {  // 定義一個方法，發(fā)送消息到 Redis Stream，參數(shù) content 是消息的內(nèi)容
        Map<String, String> map = new HashMap<>();  // 創(chuàng)建一個 Map 用來存儲消息內(nèi)容
        map.put("content", content);  // 將消息內(nèi)容添加到 Map 中，鍵是 "content"，值是傳入的內(nèi)容

        // 在消息隊列中添加消息，調(diào)用 StringRedisTemplate 的 opsForStream 方法
        RecordId recordId = redisTemplate.opsForStream()  // 獲取操作 Redis Stream 的操作對象
                .add(StreamRecords.objectBacked(map)  // 創(chuàng)建一個 Stream 記錄，將 Map 轉(zhuǎn)化為對象記錄
                .withStreamKey(STREAM_KEY));  // 設(shè)置該記錄屬于的 Stream（消息隊列）的名稱
        // 輸出記錄的 ID，表示消息已經(jīng)成功發(fā)送
        System.out.println("消息發(fā)送成功，id: " + recordId.getValue());  // 打印消息的 ID，表明該消息已經(jīng)被成功加入到 Stream 中
    }
}

RecordId recordId = redisTemplate.opsForStream().add(StreamRecords.objectBacked(map).withStreamKey("mystream"));

@Slf4j  
@Service  
public class RedisStreamConsumerService { 
    private static final String STREAM_KEY = "mystream";  // Redis Stream 的名稱，這里指定隊列名為 "mystream"
    private static final String GROUP_NAME = "mygroup";  // 消費(fèi)者組的名稱，多個消費(fèi)者可以通過組名共享消費(fèi)隊列
    private static final String CONSUMER_NAME = "consumer-1";  // 消費(fèi)者的名稱，消費(fèi)者名稱在同一消費(fèi)者組內(nèi)必須唯一

    @Autowired  
    private StringRedisTemplate redisTemplate;

    @PostConstruct  // 使用該注解能讓方法在 Spring 完成依賴注入后自動調(diào)用，用于初始化任務(wù)
    @Async  // 將該方法標(biāo)記為異步執(zhí)行，允許它在單獨(dú)的線程中運(yùn)行，不會阻塞主線程，@EnableAsync 需要在配置類中啟用
    public void start() {  // 啟動方法，在應(yīng)用啟動時執(zhí)行
        // 無限循環(huán)，不斷從 Redis Stream 中讀取消息（可以改為定時任務(wù)等方式）
        while (true) {
            try {
                // 設(shè)置 Stream 讀取的阻塞超時，設(shè)置最多等待 2 秒
                StreamReadOptions options = StreamReadOptions.empty().block(Duration.ofSeconds(2));
                // 從指定的消費(fèi)者組中讀取消息，">" 表示只消費(fèi)未被消費(fèi)過的消息
                List<MapRecord<String, Object, Object>> messages = redisTemplate.opsForStream().read(
                        Consumer.from(GROUP_NAME, CONSUMER_NAME),  // 指定消費(fèi)者組和消費(fèi)者名稱
                        options,  // 設(shè)置讀取選項(xiàng)，包含阻塞時間
                        StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed())  // 從最后消費(fèi)的消息開始讀取
                );
                // 如果沒有消息，繼續(xù)循環(huán)讀取
                if (messages == null || messages.isEmpty()) {
                    continue;  
                }
                // 處理每一條讀取到的消息
                for (MapRecord<String, Object, Object> message : messages) {
                    String messageId = message.getId();  // 獲取消息的唯一標(biāo)識符（ID）
                    Map<Object, Object> value = message.getValue();  // 獲取消息內(nèi)容（以 Map 形式存儲）
                    log.info("接收到消息，id={}，內(nèi)容={}", messageId, value);  // 打印日志，記錄消息 ID 和內(nèi)容
                    // 在這里加入業(yè)務(wù)邏輯處理
                    // 例如處理消息并執(zhí)行相應(yīng)的操作
                    // ...

                    // 消息處理成功后，需要確認(rèn)消息已經(jīng)被消費(fèi)（通過 XACK 命令）
                    redisTemplate.opsForStream().acknowledge(STREAM_KEY, GROUP_NAME, messageId);  // 確認(rèn)消費(fèi)的消息
                }
            } catch (Exception e) {
                log.error("讀取 Redis Stream 消息異常", e);  // 異常捕獲，記錄錯誤日志
            }
        }
    }
}

MapRecord<String, Object, Object> message = redisTemplate.opsForStream().read(Consumer.from("mygroup", "consumer1"), options, StreamOffset.create("mystream", ReadOffset.lastConsumed()));

@SpringBootApplication
@EnableAsync // 啟動異步支持
public class MyApplication {
    public static void main(String[] args) {
        SpringApplication.run(MyApplication.class, args);
    }
}