├── config
│   └── RedissonConfig.java                # 配置 Redisson 客戶端，創(chuàng)建 RedissonClient Bean
├── controller
│   └── DeliveryController.java            # 提供REST 接口模擬訂單創(chuàng)建和收貨操作，觸發(fā)延遲任務(wù)
├── enums
│   └── DelayQueueEnum.java                # 定義延遲隊(duì)列的業(yè)務(wù)枚舉及其關(guān)聯(lián)的處理類
├── hander
│   ├── DelayQueueHandler.java             # 延遲隊(duì)列處理器接口，定義 execute 方法
│   ├── EvaluationTimeoutHandler.java      # 處理評(píng)價(jià)超時(shí)邏輯的具體實(shí)現(xiàn)類
│   └── OrderPaymentTimeoutHandler.java    # 處理訂單支付超時(shí)邏輯的具體實(shí)現(xiàn)類
├── runner
│   └── RedisDelayQueueRunner.java         # 啟動(dòng)后監(jiān)聽并執(zhí)行延遲隊(duì)列任務(wù)，使用線程池并發(fā)處理
├── utils
│   ├── RedisDelayQueueUtil.java           # 封裝 Redis 延遲隊(duì)列的操作方法（添加/獲取元素）
│   └── SpringUtils.java                   # 工具類，用于在非 Spring 管理類中獲取 Bean
└── SpringbootApplication.java             # Spring Boot 主類，包含程序入口 main 方法

package com.zhou.demo.utils;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
/**
 * redis延遲隊(duì)列工具
 */
@Slf4j
@Component
public class RedisDelayQueueUtil {
    @Resource
    private RedissonClient redissonClient;
    /**
     * 將元素添加到延遲隊(duì)列中
     *
     * @param queueCode 隊(duì)列鍵（用于標(biāo)識(shí)不同的隊(duì)列）
     * @param value     要添加到隊(duì)列中的值（泛型類型）
     * @param delay     延遲時(shí)間（指定元素在隊(duì)列中延遲被消費(fèi)的時(shí)間）
     * @param timeUnit  時(shí)間單位（與延遲時(shí)間配合使用，如秒、毫秒等）
     */
    public <T> void addDelayQueue(String queueCode, T value, long delay, TimeUnit timeUnit) {
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);
            log.info("(添加延時(shí)隊(duì)列成功) 隊(duì)列鍵：{}，隊(duì)列值：{}，延遲時(shí)間：{}秒", queueCode, value, timeUnit.toSeconds(delay));
        } catch (Exception e) {
            log.error("(添加延時(shí)隊(duì)列失敗) {}", e.getMessage(), e);
            throw new RuntimeException("(添加延時(shí)隊(duì)列失敗)", e);
        }
    }
    /**
     * 獲取延遲隊(duì)列中的元素
     *
     * @param queueCode 隊(duì)列鍵
     * @param <T>       元素類型
     * @return 隊(duì)列中的元素
     */
    public <T> T getDelayQueue(String queueCode) throws InterruptedException {
        RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        return blockingDeque.take();
    }
}

package com.zhou.demo.runner;
import com.zhou.demo.enums.DelayQueueEnum;
import com.zhou.demo.hander.DelayQueueHandler;
import com.zhou.demo.utils.RedisDelayQueueUtil;
import com.zhou.demo.utils.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * Redis延遲隊(duì)列運(yùn)行器
 * 用于在Spring Boot啟動(dòng)后監(jiān)聽各個(gè)延遲隊(duì)列，并在線程池中執(zhí)行對(duì)應(yīng)的業(yè)務(wù)邏輯。
 *
 * @author zhouquan
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {
    @Resource
    private RedisDelayQueueUtil redisDelayQueueUtil;
    /**
     * 線程池，用于并發(fā)監(jiān)聽不同的延遲隊(duì)列
     */
    private final ExecutorService executorService = Executors.newCachedThreadPool();
    /**
     * 運(yùn)行狀態(tài)標(biāo)志，控制線程是否持續(xù)監(jiān)聽隊(duì)列
     */
    private volatile boolean running = true;
    /**
     * Spring Boot 啟動(dòng)完成后自動(dòng)運(yùn)行的方法
     * 遍歷所有延遲隊(duì)列枚舉，為每個(gè)隊(duì)列創(chuàng)建一個(gè)監(jiān)聽線程
     *
     * @param args 命令行參數(shù)
     */
    @Override
    public void run(String... args) {
        for (DelayQueueEnum queueEnum : DelayQueueEnum.values()) {
            executorService.execute(() -> {
                log.info("啟動(dòng)延遲隊(duì)列監(jiān)聽線程：{}", queueEnum.getCode());
                while (running) {
                    try {
                        Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
                        DelayQueueHandler handler = SpringUtils.getBean(queueEnum.getBeanClass());
                        handler.execute(value);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        log.warn("線程中斷：{}", queueEnum.getCode());
                    } catch (Exception ex) {
                        log.error("延遲隊(duì)列 [{}] 處理異常：{}", queueEnum.getCode(), ex.getMessage(), ex);
                    }
                }
            });
        }
        log.info("所有 Redis 延遲隊(duì)列監(jiān)聽啟動(dòng)完成");
    }
    /**
     * 在 Bean 銷毀前關(guān)閉線程池，釋放資源
     */
    @PreDestroy
    public void shutdown() {
        log.info("準(zhǔn)備關(guān)閉 Redis 延遲隊(duì)列監(jiān)聽線程池");
        running = false;
        executorService.shutdownNow();
    }
}

package com.zhou.demo.enums;
import com.zhou.demo.hander.EvaluationTimeoutHandler;
import com.zhou.demo.hander.OrderPaymentTimeoutHandler;
import com.zhou.demo.hander.DelayQueueHandler;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
/**
 * 延遲隊(duì)列業(yè)務(wù)枚舉
 *
 * @author 18324
 */
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum DelayQueueEnum {
    /**
     * 訂單超時(shí)
     */
    ORDER_PAYMENT_TIMEOUT("order_payment_timeout", "訂單支付超時(shí)", OrderPaymentTimeoutHandler.class),
    /**
     * 評(píng)價(jià)超時(shí)
     */
    EVALUATION_TIMEOUT("evaluation_timeout", "評(píng)價(jià)超時(shí)", EvaluationTimeoutHandler.class);
    /**
     * 延遲隊(duì)列 Redis Key
     */
    private String code;
    /**
     * 中文描述
     */
    private String name;
    /**
     * 延遲隊(duì)列具體業(yè)務(wù)實(shí)現(xiàn)的 Bean
     * 可通過(guò) Spring 的上下文獲取
     */
    private Class<? extends DelayQueueHandler<Long>> beanClass;
}

package com.zhou.demo.enums;
import com.zhou.demo.hander.EvaluationTimeoutHandler;
import com.zhou.demo.hander.OrderPaymentTimeoutHandler;
import com.zhou.demo.hander.DelayQueueHandler;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
/**
 * 延遲隊(duì)列業(yè)務(wù)枚舉
 *
 * @author 18324
 */
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum DelayQueueEnum {
    /**
     * 訂單超時(shí)
     */
    ORDER_PAYMENT_TIMEOUT("order_payment_timeout", "訂單支付超時(shí)", OrderPaymentTimeoutHandler.class),
    /**
     * 評(píng)價(jià)超時(shí)
     */
    EVALUATION_TIMEOUT("evaluation_timeout", "評(píng)價(jià)超時(shí)", EvaluationTimeoutHandler.class);
    /**
     * 延遲隊(duì)列 Redis Key
     */
    private String code;
    /**
     * 中文描述
     */
    private String name;
    /**
     * 延遲隊(duì)列具體業(yè)務(wù)實(shí)現(xiàn)的 Bean
     * 可通過(guò) Spring 的上下文獲取
     */
    private Class<? extends DelayQueueHandler<Long>> beanClass;
}