fix 修复 延迟队列在投递消息未到达时间的时候 服务死机导致重启收不到消息
This commit is contained in:
parent
3acbf6efee
commit
9b0938e0d6
@ -224,8 +224,12 @@ public class QueueUtils {
|
|||||||
/**
|
/**
|
||||||
* 订阅阻塞队列(可订阅所有实现类 例如: 延迟 优先 有界 等)
|
* 订阅阻塞队列(可订阅所有实现类 例如: 延迟 优先 有界 等)
|
||||||
*/
|
*/
|
||||||
public static <T> void subscribeBlockingQueue(String queueName, Consumer<T> consumer) {
|
public static <T> void subscribeBlockingQueue(String queueName, Consumer<T> consumer, boolean isDelayed) {
|
||||||
RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
|
RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
|
||||||
|
if (isDelayed) {
|
||||||
|
// 订阅延迟队列
|
||||||
|
CLIENT.getDelayedQueue(queue);
|
||||||
|
}
|
||||||
queue.subscribeOnElements(consumer);
|
queue.subscribeOnElements(consumer);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -1,5 +1,6 @@
|
|||||||
package org.dromara.demo.controller.queue;
|
package org.dromara.demo.controller.queue;
|
||||||
|
|
||||||
|
import cn.dev33.satoken.annotation.SaIgnore;
|
||||||
import org.dromara.common.core.domain.R;
|
import org.dromara.common.core.domain.R;
|
||||||
import org.dromara.common.redis.utils.QueueUtils;
|
import org.dromara.common.redis.utils.QueueUtils;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
@ -22,6 +23,7 @@ import java.util.concurrent.TimeUnit;
|
|||||||
* @author Lion Li
|
* @author Lion Li
|
||||||
* @version 3.6.0
|
* @version 3.6.0
|
||||||
*/
|
*/
|
||||||
|
@SaIgnore
|
||||||
@Slf4j
|
@Slf4j
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
@RestController
|
@RestController
|
||||||
@ -40,7 +42,7 @@ public class DelayedQueueController {
|
|||||||
QueueUtils.subscribeBlockingQueue(queueName, (String orderNum) -> {
|
QueueUtils.subscribeBlockingQueue(queueName, (String orderNum) -> {
|
||||||
// 观察接收时间
|
// 观察接收时间
|
||||||
log.info("通道: {}, 收到数据: {}", queueName, orderNum);
|
log.info("通道: {}, 收到数据: {}", queueName, orderNum);
|
||||||
});
|
}, true);
|
||||||
return R.ok("操作成功");
|
return R.ok("操作成功");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user