/*增加光明大模型处理框架*/

src/main/java/com/inspect/simulator/constant/AlgConstants.java

@@ -9,4 +9,6 @@ public class AlgConstants {
     public static final String XB = "xb";
 
     public static final String ALG_SUBTYPE_CODE = "alg_subtype_code";
+
+    public static final String LUMINOSITY_REQUEST_QUEUE = "luminosity:request:queue";
 }

src/main/java/com/inspect/simulator/service/impl/LuminosityRequestConsumerManager.java

@@ -0,0 +1,54 @@
+package com.inspect.simulator.service.impl;
+
+import com.inspect.simulator.constant.AlgConstants;
+import com.inspect.simulator.domain.algorithm.in.AnalyseRequest;
+import com.inspect.simulator.utils.redis.RedisService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import javax.annotation.PreDestroy;
+
+@Component
+public class LuminosityRequestConsumerManager {
+    private static final Logger log = LoggerFactory.getLogger(LuminosityRequestConsumerManager.class);
+
+    private RedisQueueConsumerAsync<AnalyseRequest> consumer;
+
+    @Resource
+    private RedisService redisService;
+
+
+    @PostConstruct
+    public void initConsumer() {
+        consumer = new RedisQueueConsumerAsync<AnalyseRequest>(
+                redisService.redisTemplate,
+                AlgConstants.LUMINOSITY_REQUEST_QUEUE,
+                AnalyseRequest.class,
+                1000,
+                500,
+                request -> {
+                    try {
+                        log.info("LuminosityRequestConsumerManager queueSize: {}, request: {}", getQueueSize(), request);
+                    } catch (Exception e) {
+                        log.info("LuminosityRequestConsumerManager error queueSize: {}, request: {}", getQueueSize(), request);
+                    }
+                }
+        );
+
+        consumer.start(); // 应用启动时即开始消费
+    }
+
+    @PreDestroy
+    public void stopConsumer() {
+        if (consumer != null) {
+            consumer.stop();
+        }
+    }
+
+    public long getQueueSize() {
+        return redisService.redisTemplate.opsForList().size(AlgConstants.LUMINOSITY_REQUEST_QUEUE);
+    }
+}

src/main/java/com/inspect/simulator/service/impl/RedisQueueConsumerAsync.java

@@ -0,0 +1,80 @@
+package com.inspect.simulator.service.impl;
+
+import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import java.time.Duration;
+import java.util.function.Consumer;
+
+public class RedisQueueConsumerAsync<T> implements Runnable {
+    private static final Logger logger = LoggerFactory.getLogger(RedisQueueConsumerAsync.class);
+
+    private final RedisTemplate<String, String> redisTemplate;
+    private final Gson gson = new Gson();
+    private final String queueKey;
+    private final Class<T> clazz;
+
+    private final long intervalMs;
+    private final long idleSleepMs;
+
+    private final Consumer<T> handler;
+
+    private volatile boolean running = false;
+
+    public RedisQueueConsumerAsync(RedisTemplate<String, String> redisTemplate,
+                                   String queueKey,
+                                   Class<T> clazz,
+                                   long intervalMs,
+                                   long idleSleepMs,
+                                   Consumer<T> handler) {
+        this.redisTemplate = redisTemplate;
+        this.queueKey = queueKey;
+        this.clazz = clazz;
+        this.intervalMs = intervalMs;
+        this.idleSleepMs = idleSleepMs;
+        this.handler = handler;
+    }
+
+    public void start() {
+        if (!running) {
+            running = true;
+            new Thread(this, "RedisQueueConsumerAsync-" + queueKey).start();
+        }
+    }
+
+    public void stop() {
+        running = false;
+    }
+
+    @Override
+    public void run() {
+        while (running) {
+            try {
+                String json = redisTemplate.opsForList().leftPop(queueKey, Duration.ofSeconds(1));
+
+                if (json == null) {
+                    Thread.sleep(idleSleepMs);
+                    continue;
+                }
+
+                long start = System.currentTimeMillis();
+
+                T obj = gson.fromJson(json, clazz);
+                handler.accept(obj);
+
+                long cost = System.currentTimeMillis() - start;
+                if (cost < intervalMs) {
+                    Thread.sleep(intervalMs - cost);
+                }
+
+            } catch (Exception e) {
+                logger.error("RedisQueueConsumerAsync error", e);
+            }
+        }
+
+        logger.info("RedisQueueConsumerAsync for [{}] stopped.", queueKey);
+    }
+}
+