diff --git a/ruoyi-common/src/main/java/com/ruoyi/common/utils/redis/QueueUtils.java b/ruoyi-common/src/main/java/com/ruoyi/common/utils/redis/QueueUtils.java index 842e4565b..059735247 100644 --- a/ruoyi-common/src/main/java/com/ruoyi/common/utils/redis/QueueUtils.java +++ b/ruoyi-common/src/main/java/com/ruoyi/common/utils/redis/QueueUtils.java @@ -224,8 +224,12 @@ public class QueueUtils { /** * 订阅阻塞队列(可订阅所有实现类 例如: 延迟 优先 有界 等) */ - public static void subscribeBlockingQueue(String queueName, Consumer consumer) { + public static void subscribeBlockingQueue(String queueName, Consumer consumer, boolean isDelayed) { RBlockingQueue queue = CLIENT.getBlockingQueue(queueName); + if (isDelayed) { + // 订阅延迟队列 + CLIENT.getDelayedQueue(queue); + } queue.subscribeOnElements(consumer); } diff --git a/ruoyi-demo/src/main/java/com/ruoyi/demo/controller/queue/DelayedQueueController.java b/ruoyi-demo/src/main/java/com/ruoyi/demo/controller/queue/DelayedQueueController.java index e6dbee863..511621cf7 100644 --- a/ruoyi-demo/src/main/java/com/ruoyi/demo/controller/queue/DelayedQueueController.java +++ b/ruoyi-demo/src/main/java/com/ruoyi/demo/controller/queue/DelayedQueueController.java @@ -40,7 +40,7 @@ public class DelayedQueueController { QueueUtils.subscribeBlockingQueue(queueName, (String orderNum) -> { // 观察接收时间 log.info("通道: {}, 收到数据: {}", queueName, orderNum); - }); + }, true); return R.ok("操作成功"); }