redis延迟队列参考博客：https://blog.csdn.net/m0_37975854/article/details/119836978
jkd延迟队列参考博客：
https://www.jianshu.com/p/a1b3aa87f78b
一、实现延迟队列，a、可以使用mq的死信队列，b、可以使用 redis里面zset的延迟队列 c、可以使用jkd里面的延迟队列
二、zset实现延迟队列(跟sadd不同的就是有score)
zset参考博客：
https://blog.csdn.net/cm15835106905/article/details/126323705
; 三、实现代码
3.1、参考博客：
https://blog.csdn.net/m0_37975854/article/details/119836978
3.2、通过这篇博客学到了什么
- 接口定义规则，多个实现类。规定只能由这个接口实现的对象的class文件才能调用的方法，这么写：
Class<? extends RedisDelayedQueueListener> clazz， ? extends代表某个继承了RedisDelayedQueueListener接口的对象 - 如果想要拿到某个接口的全部实现类，使用:
- 继承ApplicationContextAware就是为了获取ApplicationContext(上下文)，
也可以使用注入的方式获取ApplicationContext；继承ApplicationContextAware则在项目启动的时候就执行setApplicationContext方法
; 3.3、代码实现
接口：
package com.yj.redis;
public interface RedisDelayedQueueListener<T> {
void invoke(T t);
}
实现类：
package com.yj.redis;
import com.yj.vo.customer.CustomerManagerVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
@Slf4j
public class PayQCordListener implements RedisDelayedQueueListener<CustomerManagerVO> {
@Override
public void invoke(CustomerManagerVO payStateReqVO) {
log.info("延迟时效时间：{}", new Date());
log.info("延迟失效,内容:{}", payStateReqVO);
log.info("延迟失效,内容:{},处理结果:{}", payStateReqVO,"测试处理");
}
}
创建redis中的zset和score（相当于创建延迟队列）：
package com.yj.redis;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class RedisDelayedQueue {
@Resource(name = "getRedisson")
private RedissonClient getRedisson;
private <T> void addQueue(T t, long delay, TimeUnit timeUnit, String queueName) {
log.info("开始延迟队列时间：{}", new Date());
log.info("添加延迟队列,监听名称:{},时间:{},时间单位:{},内容:{}" , queueName, delay, timeUnit,t);
RBlockingQueue<T> blockingFairQueue = getRedisson.getBlockingQueue(queueName);
RDelayedQueue<T> delayedQueue = getRedisson.getDelayedQueue(blockingFairQueue);
delayedQueue.offer(t, delay, timeUnit);
}
public <T> void addQueueSeconds(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.SECONDS, clazz.getName());
}
public <T> void addQueueMinutes(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.MINUTES, clazz.getName());
}
public <T> void addQueueHours(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.HOURS, clazz.getName());
}
public <T> void addQueueDays(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.DAYS, clazz.getName());
}
}
监听redis中数据是否过期，如果过期则进行消费
package com.yj.redis;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Map;
@Component
@Slf4j
public class RedisDelayedQueueInit implements ApplicationContextAware {
@Resource(name = "getRedisson")
private RedissonClient getRedisson;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
for (Map.Entry<String, RedisDelayedQueueListener> taskEventListenerEntry : map.entrySet()) {
String listenerName = taskEventListenerEntry.getValue().getClass().getName();
startThread(listenerName, taskEventListenerEntry.getValue());
}
}
private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
RBlockingQueue<T> blockingFairQueue = getRedisson.getBlockingQueue(queueName);
getRedisson.getDelayedQueue(blockingFairQueue);
Thread thread = new Thread(() -> {
log.info("启动监听队列线程" + queueName);
while (true) {
try {
T t = blockingFairQueue.take();
log.info("监听队列线程，监听名称：{},内容:{}", queueName, t);
redisDelayedQueueListener.invoke(t);
} catch (Exception e) {
log.info("监听队列线程错误,", e);
}
}
});
thread.setName(queueName);
thread.start();
}
}
Original: https://blog.csdn.net/M1275601161/article/details/127801705
Author: 进击的北极熊
Title: redis实现延迟队列
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/209484/
转载文章受原作者版权保护。转载请注明原作者出处！
