![基于Redisson的延迟队列实现](https://www.zixueka.com/wp-content/uploads/2024/01/1706695673-32533f0a82784e0.jpg)
package com.dong.mytest.demo.client;
import cn.hutool.extra.spring.SpringUtil;
import com.dong.mytest.demo.common.dto.DelayMessage;
import com.dong.mytest.demo.common.util.DateUtil;
import com.dong.mytest.demo.service.delayqueue.DelayQueueConsumer;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* @author dong
*/
@Slf4j
@Component
public class RedissonDelayQueueClient implements InitializingBean {
@Resource
private RedissonClient redissonClient;
private final Map<String, RDelayedQueue<DelayMessage>> delayQueueMap = new ConcurrentHashMap<>(16);
public void addDelayMessage(DelayMessage delayMessage) {
log.info("delayMessage={}", delayMessage);
if (delayQueueMap.get(delayMessage.getQueueName()) == null) {
log.warn("queueName={},该延迟队列不存在,请确认后再试...", delayMessage.getQueueName());
return;
}
delayMessage.setCreateTime(DateUtil.getNowFormatStr());
RDelayedQueue<DelayMessage> rDelayedQueue = delayQueueMap.get(delayMessage.getQueueName());
rDelayedQueue.offer(delayMessage, delayMessage.getDelayTime(), delayMessage.getTimeUnit() == null ? TimeUnit.SECONDS : delayMessage.getTimeUnit());
}
@Override
public void afterPropertiesSet() throws Exception {
// 有新的延迟队列在这里添加,队列消费类需要继承DelayQueueConsumer,并且service名称为 ${queueName}Consumer
List<String> queueNameList = Lists.newArrayList("orderAutoCancelDelayQueue");
// 加载延迟队列
for (String queueName : queueNameList) {
DelayQueueConsumer delayQueueConsumer = SpringUtil.getBean(queueName + "Consumer");
if (delayQueueConsumer == null) {
throw new RuntimeException("queueName=" + queueName + ",delayQueueConsumer=null,请检查配置...");
}
// Redisson的延时队列是对另一个队列的再包装,使用时要先将延时消息添加到延时队列中,当延时队列中的消息达到设定的延时时间后,该延时消息才会进行进入到被包装队列中,因此,我们只需要对被包装队列进行监听即可。
RBlockingQueue<DelayMessage> rBlockingQueue = redissonClient.getBlockingDeque(queueName);
RDelayedQueue<DelayMessage> rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);
delayQueueMap.put(queueName, rDelayedQueue);
// 订阅新元素的到来,调用的是takeAsync(),异步执行
rBlockingQueue.subscribeOnElements(delayQueueConsumer::execute);
}
}
}
hmoban主题是根据ripro二开的主题,极致后台体验,无插件,集成会员系统
自学咖网 »
基于Redisson的延迟队列实现