import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
@Service
public class OrderService {
    @Resource
    private RedissonClient redissonClient;
    // 隊(duì)列的名稱，可以理解為用于存放待關(guān)閉訂單的“信箱”
    private static final String ORDER_QUEUE_KEY = "order-close-queue";
    public void createOrder(Order order) {
        // 1. 保存訂單到數(shù)據(jù)庫(kù) (省略具體邏輯)
        saveToDB(order);
        log.info("訂單 [{}] 創(chuàng)建成功，等待支付...", order.getId());
        // 2. 將訂單ID放入延遲隊(duì)列，設(shè)置30分鐘后“爆炸”
        try {
            // 獲取一個(gè)阻塞隊(duì)列，這是消費(fèi)者最終要監(jiān)聽的目標(biāo)隊(duì)列
            RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(ORDER_QUEUE_KEY);
            // 基于阻塞隊(duì)列，創(chuàng)建一個(gè)延遲隊(duì)列
            RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
            // 將訂單ID放入延遲隊(duì)列，延遲30分鐘
            delayedQueue.offer(order.getId().toString(), 30, TimeUnit.MINUTES);
            log.info("訂單 [{}] 已放入延遲隊(duì)列，30分鐘后將自動(dòng)關(guān)閉", order.getId());
        } catch (Exception e) {
            log.error("放入延遲隊(duì)列失敗", e);
            // 這里可以考慮補(bǔ)償機(jī)制，比如記錄日志后由定時(shí)任務(wù)兜底
        }
    }
}

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class OrderCloseListener implements CommandLineRunner {
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private OrderService orderService; // 注入你的訂單Service
    private static final String ORDER_QUEUE_KEY = "order-close-queue";
    @Override
    public void run(String... args) throws Exception {
        // 程序啟動(dòng)后，在一個(gè)獨(dú)立的線程中監(jiān)聽隊(duì)列
        new Thread(() -> {
            log.info("訂單關(guān)閉監(jiān)聽線程已啟動(dòng)，等待到期訂單...");
            // 獲取與生產(chǎn)者相同的阻塞隊(duì)列
            RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(ORDER_QUEUE_KEY);
            // 注意：這里必須調(diào)用一次getDelayedQueue，目的是在消費(fèi)端也初始化相關(guān)的監(jiān)聽器
            // 雖然不調(diào)用也能工作，但官方推薦調(diào)用以保證可靠性 
            redissonClient.getDelayedQueue(blockingQueue); 
            while (true) {
                try {
                    // 阻塞等待，直到有訂單到期。take()方法會(huì)一直阻塞直到拿到數(shù)據(jù)
                    String orderId = blockingQueue.take();
                    log.info("收到到期訂單ID：{}，開始執(zhí)行關(guān)閉操作", orderId);
                    // 執(zhí)行真正的關(guān)單業(yè)務(wù)邏輯
                    orderService.closeExpiredOrder(Long.parseLong(orderId));
                } catch (InterruptedException e) {
                    log.error("監(jiān)聽線程被中斷", e);
                    Thread.currentThread().interrupt();
                    break; // 線程中斷時(shí)退出循環(huán)
                } catch (Exception e) {
                    log.error("處理到期訂單時(shí)發(fā)生錯(cuò)誤", e);
                    // 防止單個(gè)消息處理失敗導(dǎo)致循環(huán)中斷，繼續(xù)監(jiān)聽下一個(gè)
                }
            }
        }, "OrderClose-Listener").start();
    }
}