继上一章节基于Redisson实现延迟队列
我们实现了延迟队列的基本功能，但是存在一个问题就是对于延迟队列，我们能不能有一个监听机制监听消息过期的流程呢？也就是说一旦到期就自动监听并触发呢。这时我粗略的想法是采用线程去监听队列的消息。在项目启动时我们就初始化并用线程池去循环监听队列中的消息，话不多说，按照思路开干。稍微封装下
定义一个接口对于过期消息的监听，一旦到期触发onExpired。对于以后我们消费者只要实现这接口即可实现消息的过期监听。后面会详细讲到
@FunctionalInterface
public interface RedisDelayedQueueListener<T> {
/**
* 对于过期消息的监听
*
* @param t 泛型
*/
void onExpired(T t);
}
这时我们再自定义一个注解用于初始化并监听队列中的消息
其中的name就是队列的名称
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface RedisDelayedQueue {
/**
* 队列名称
*
* @return string
*/
String name();
}
定义一个队列消费任务DelayPollTask，完成队列的消费逻辑。这里参考了redis Strem源码
@Slf4j
public class DelayPollTask<T> implements Runnable{
private final RedisDelayedQueueListener<T> listener;
private final RedissonClient redissonClient;
private final String name;
public DelayPollTask(RedisDelayedQueueListener<T> listener, RedissonClient redissonClient, String name) {
this.listener = listener;
this.redissonClient = redissonClient;
this.name = name;
}
@Override
public void run() {
loop();
}
@SuppressWarnings({"all"})
private void loop(){
while (true) {
try {
Thread.sleep(0);
System.out.println(name);
RBlockingQueue<T> retryQueue = redissonClient.getBlockingQueue(name);
// 订阅redis队列 解决项目重新启动并不会消费之前队列里的消息的问题
redissonClient.getDelayedQueue(retryQueue);
T t = retryQueue.take();
log.info("poll--->" + t);
listener.onExpired(t);
} catch (Exception e) {
log.error("DelayPollTask error", e);
}
}
}
}
定义一个队列初始化器DelayQueueInitializer初始化并开启线程监听队列中的消息
@Slf4j
@Component
public class DelayQueueInitializer implements InitializingBean {
@Autowired
private RedissonClient redissonClient;
/**
* 线程池来监听队列的消息 线程的配置以实际项目为主
*/
private final ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 30,
10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1000),
new NamedThreadFactory("delayQueueJob", false));
/**
* 任务项，用于收集注册进来的线程任务
*/
private final List<Runnable> tasks = new ArrayList<>();
@Override
public void afterPropertiesSet() {
// 初始化逻辑
init();
}
@SuppressWarnings({"unchecked", "rawtypes"})
private void init() {
Collection<RedisDelayedQueueListener> listeners = SpringUtil.getBeansOfType(RedisDelayedQueueListener.class).values();
for (RedisDelayedQueueListener listener : listeners) {
RedisDelayedQueue annotation = listener.getClass().getAnnotation(RedisDelayedQueue.class);
if (annotation != null) {
register(annotation.name(), listener);
}
}
log.info("成功注册" + tasks.size() + "条队列");
// 开启监听
start();
}
/**
* 注册监听器
*
* @param name 队列的名称
* @param listener 监听器
* @param <T> 泛型
*/
private <T> void register(String name, RedisDelayedQueueListener<T> listener) {
tasks.add(new DelayPollTask<>(listener, redissonClient, name));
}
/**
* 开启对队列的监听
*/
private void start() {
tasks.forEach(executor::execute);
log.info("成功开启" + tasks.size() + "条队列");
}
}
接下来我们测试下
定义一个订单类OrderDTO
@NoArgsConstructor
@AllArgsConstructor
@Data
public class OrderDTO implements Serializable {
private String orderNo;
private String orderName;
@Override
public boolean equals(Object object) {
if (this == object) return true;
if (!(object instanceof OrderDTO)) return false;
OrderDTO orderDTO = (OrderDTO) object;
return Objects.equals(getOrderNo(), orderDTO.getOrderNo()) && Objects.equals(getOrderName(), orderDTO.getOrderName());
}
@Override
public int hashCode() {
return Objects.hash(getOrderNo(), getOrderName());
}
}
重点来了，只要实现RedisDelayedQueueListener接口即可实现对于订单过期的监听
@RedisDelayedQueue标记队列的名称，用于消费之前队列中的消息。以后对于扩展也方便
@RedisDelayedQueue(name = "Queue-1")
@Component
public class OrderListener implements RedisDelayedQueueListener<OrderDTO> {
@Override
public void onExpired(OrderDTO orderDTO) {
System.out.println("订单编号" + orderDTO.getOrderNo());
System.out.println("订单名称" + orderDTO.getOrderName());
// 业务流程...
}
}
测试：
稍微封装下队列的操作
@Component
public class RedissonDelayQueue {
@Autowired
private RedissonClient redissonClient;
/**
* 放入延迟队列
*
* @param data 数据
* @param delay 延迟时间
* @param timeUnit 时间单位
* @param queueName 队列名称
*/
public <T> void addQueue(T data, long delay, TimeUnit timeUnit, String queueName) {
// redisson 延迟队列
RBlockingQueue<T> blockingQueue = redissonClient.getBlockingQueue(queueName);
RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
delayedQueue.offer(data, delay, timeUnit);
}
public void delete(Object data, String queueName) {
RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(queueName);
RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
delayedQueue.remove(data);
}
}
测试：
OrderDTO orderDTO = new OrderDTO();
orderDTO.setOrderNo("123");
orderDTO.setOrderName("xxx");
// Queue-1 必须和@RedisDelayedQueue(name = "Queue-1") 中的name一直
redissonDelayQueue.addQueue(orderDTO, 3, TimeUnit.SECONDS, "Queue-1");
这时上面定义的OrderListener就能消费队列中的消息了，注意OrderListener在监听时一定要处理好异常，否则一旦报错可能后续监听将失效了
希望大佬们能发现其中的不足，提出宝贵的意见~
