Redisson延迟队列的封装

继上一章节基于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在监听时一定要处理好异常，否则一旦报错可能后续监听将失效了

希望大佬们能发现其中的不足，提出宝贵的意见~