延时队列优化

yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListener.java

@@ -0,0 +1,39 @@
+package co.yixiang.yshop.module.message.redismq;
+
+/**
+ * 延时队列接口
+ * @author hupeng
+ * @date 2024.6.25
+ */
+public interface DelayedQueueListener<T> {
+    /**
+     * 是否启用
+     *
+     * @return boolean
+     */
+    default boolean isEnable() {
+        return true;
+    }
+
+    /**
+     * 队列键
+     *
+     * @return String
+     */
+    String delayedQueueKey();
+
+    /**
+     * 消费
+     *
+     * @param message Object
+     * @throws Exception Exception
+     */
+    void consume(T message) throws Exception;
+
+    /**
+     * 发生异常时最终处理
+     */
+    default void whenExceptionFinally() {
+    }
+
+}

yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueueListenerConfigurer.java

@@ -0,0 +1,58 @@
+package co.yixiang.yshop.module.message.redismq;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 配置
+ * @author hupeng
+ * @date 2024.6.25
+ */
+@Component
+public class DelayedQueueListenerConfigurer implements InitializingBean, DisposableBean {
+    private ThreadPoolExecutor delayedThreadPoolExecutor;
+
+    private final List<DelayedQueueListener<?>> delayedQueueListenerList;
+
+    private final RedissonClient redissonClient;
+
+    public DelayedQueueListenerConfigurer(List<DelayedQueueListener<?>> delayedQueueListenerList, RedissonClient redissonClient) {
+        this.delayedQueueListenerList = delayedQueueListenerList;
+        this.redissonClient = redissonClient;
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        if (delayedThreadPoolExecutor != null) {
+            delayedThreadPoolExecutor.shutdownNow();
+        }
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        Assert.notEmpty(delayedQueueListenerList, "delayedQueueListenerList must not be empty!");
+
+        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("delayed-queue-pool-%d").build();
+        int numberOfJob = delayedQueueListenerList.size();
+        delayedThreadPoolExecutor = new ThreadPoolExecutor(
+                numberOfJob,
+                numberOfJob,
+                0L,
+                TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(numberOfJob),
+                namedThreadFactory
+        );
+        delayedQueueListenerList.forEach(delayedQueueListener -> delayedThreadPoolExecutor.execute(new DelayedQueuePollTask<>(redissonClient, delayedQueueListener)));
+    }
+
+}

yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/DelayedQueuePollTask.java

@@ -0,0 +1,47 @@
+package co.yixiang.yshop.module.message.redismq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RBlockingDeque;
+import org.redisson.api.RedissonClient;
+
+/**
+ * 延迟队列轮询任务类
+ * @author hupeng
+ * @date 2024.6.25
+ */
+@Slf4j
+public class DelayedQueuePollTask<T> implements Runnable{
+    private final RedissonClient redissonClient;
+
+    private final DelayedQueueListener<T> delayedQueueListener;
+
+    public DelayedQueuePollTask(RedissonClient redissonClient, DelayedQueueListener<T> delayedQueueListener) {
+        this.redissonClient = redissonClient;
+        this.delayedQueueListener = delayedQueueListener;
+    }
+
+    @Override
+    public void run() {
+        String threadName = "delayed-queue-listener-" + delayedQueueListener.getClass().getSimpleName();
+        Thread.currentThread().setName(threadName);
+        if (!delayedQueueListener.isEnable()) {
+            return;
+        }
+        RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(delayedQueueListener.delayedQueueKey());
+        // 解决消息丢失问题，发送subscribe命令订阅redis队列
+        redissonClient.getDelayedQueue(blockingDeque);
+        while (!Thread.currentThread().isInterrupted()) {
+            try {
+                T message = blockingDeque.take();
+                delayedQueueListener.consume(message);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            } catch (Exception e) {
+                log.error(e.getMessage(), e);
+            } finally {
+                delayedQueueListener.whenExceptionFinally();
+            }
+        }
+    }
+
+}

yshop-drink-boot3/yshop-module-message/yshop-module-message-biz/src/main/java/co/yixiang/yshop/module/message/redismq/msg/OrderMsg.java

@@ -0,0 +1,12 @@
+package co.yixiang.yshop.module.message.redismq.msg;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+@Builder
+public class OrderMsg implements Serializable {
+    private String orderId;
+}