Debug

src/main/java/com/backendsys/modules/queue/controller/TaskStatusController.java

@@ -1,37 +1,95 @@
 package com.backendsys.modules.queue.controller;
 
+import cn.hutool.core.convert.Convert;
+import com.backendsys.modules.common.config.security.annotations.Anonymous;
+import com.backendsys.modules.common.utils.Result;
+import com.backendsys.modules.queue.entity.Entire;
+import com.backendsys.modules.queue.service.TaskService;
+import com.backendsys.modules.sdk.comfyui.enums.TaskStatusEnums;
+import io.swagger.v3.oas.annotations.Operation;
+import jakarta.annotation.PostConstruct;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.http.HttpStatus;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.PendingMessages;
+import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
-import org.springframework.web.server.ResponseStatusException;
 
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
+import java.util.UUID;
 
 @RestController
 public class TaskStatusController {
 
     @Autowired
-    private RedisTemplate redisTemplate;
-
-    @GetMapping("/api/tasks/{taskId}/status")
-    public Map<String, String> getTaskStatus(@PathVariable String taskId) {
-        String statusKey = "task:status:" + taskId;
-        Map<Object, Object> entries = redisTemplate.opsForHash().entries(statusKey);
-
-        if (entries.isEmpty()) {
-            System.out.println("Task not found");
-        }
-
-        // 转换为String类型返回
-        return entries.entrySet().stream()
-                .collect(Collectors.toMap(
-                        e -> e.getKey().toString(),
-                        e -> e.getValue().toString()
-                ));
+    private TaskService taskService;
+
+    @Value("${comfyui.queue-key}")
+    private String QUEUE_KEY;
+    @Value("${spring.application.name}")
+    private String APPLICATION_NAME;
+
+
+    @PostConstruct
+    @Operation(summary = "初始化消费者组")
+    public void initConsumerGroup() {
+        String stream_key = APPLICATION_NAME + ":" + QUEUE_KEY;
+        String group_name = stream_key + ":group";
+        taskService.initConsumerGroup(stream_key, group_name);
+    }
+
+    @Operation(summary = "添加任务")
+    @GetMapping("/api/tasks/addTask")
+    public Result addTask() {
+        Map<String, Object> data = new LinkedHashMap<>();
+        data.put("progress", 0);
+        data.put("status", TaskStatusEnums.EXECUTION_START.getValue());
+
+        // 加入队列
+        String stream_key = APPLICATION_NAME + ":" + QUEUE_KEY;
+        RecordId record_id = taskService.addTask(stream_key, data);
+        data.put("task_id", record_id.getValue());
+
+        return Result.success().put("data", data);
+    }
+
+    @Anonymous
+    @Operation(summary = "消费任务")
+    @GetMapping("/api/tasks/consumeTask")
+    public Result consumeTask(String task_id) {
+        String stream_key = APPLICATION_NAME + ":" + QUEUE_KEY;
+        taskService.consumeTask(stream_key, task_id);
+        return Result.success();
+    }
+
+    @Anonymous
+    @Operation(summary = "移除任务")
+    @GetMapping("/api/tasks/removeTask")
+    public Result removeTask(String task_id) {
+        String stream_key = APPLICATION_NAME + ":" + QUEUE_KEY;
+        taskService.removeTask(stream_key, task_id);
+        return Result.success();
+    }
+
+    @Anonymous
+    @Operation(summary = "获取任务列表")
+    @GetMapping("/api/tasks/getTaskList")
+    public Result getTaskList() {
+        String stream_key = APPLICATION_NAME + ":" + QUEUE_KEY;
+        List<MapRecord<String, String, String>> task_list = taskService.getTaskList(stream_key);
+        return Result.success().put("data", task_list);
+    }
+
+    @Anonymous
+    @Operation(summary = "获取任务列表(未消费)")
+    @GetMapping("/api/tasks/getTaskPaddingList")
+    public Result getTaskPaddingList() {
+        String stream_key = APPLICATION_NAME + ":" + QUEUE_KEY;
+        List<MapRecord<String, String, String>> task_list = taskService.getTaskPaddingList(stream_key);
+        return Result.success().put("data", task_list);
     }
+
 }

src/main/java/com/backendsys/modules/queue/entity/Entire.java

@@ -0,0 +1,9 @@
+package com.backendsys.modules.queue.entity;
+
+import lombok.Data;
+
+@Data
+public class Entire {
+    String key;
+    Object value;
+}

src/main/java/com/backendsys/modules/queue/entity/GenerateRequest.java

@@ -1,15 +1,10 @@
 package com.backendsys.modules.queue.entity;
 
-
 import lombok.Data;
 
-import java.util.HashMap;
-import java.util.Map;
-
 @Data
 public class GenerateRequest {
     private String prompt;
     private String userId;
     private String model = "stable-diffusion-v2.1";              // 默认模型
-    private Map<String, Object> parameters = new HashMap<>();
 }

src/main/java/com/backendsys/modules/queue/service/QueueService.java

@@ -39,20 +39,22 @@
 //     * 开始排队
 //     */
 //    public void startProcessing(String queueKey) {
-//        new Thread(() -> {
-//            while (true) {
-//                // 从有序集合中取出第一个请求
-//                String requestId = Convert.toStr(redisTemplate.opsForZSet().popMin(queueKey));
-//                if (requestId != null) {
-//                    QueueRequest queueRequest = getRequestById(requestId);
-//                    int position = queueRequest.getPosition();
-//                    // 处理请求
-//                    processRequest(queueRequest);
-//                    // 可以通知用户处理完成
-//                    notifyUser(queueRequest, position);
-//                }
-//            }
-//        }).start();
+////        new Thread(() -> {
+////            while (true) {
+////                // 从有序集合中取出第一个请求
+////                String requestId = Convert.toStr(redisTemplate.opsForZSet().popMin(queueKey));
+////                if (requestId != null) {
+////                    QueueRequest queueRequest = getRequestById(requestId);
+////                    int position = queueRequest.getPosition();
+////                    // 处理请求
+////                    processRequest(queueRequest);
+////                    // 可以通知用户处理完成
+////                    notifyUser(queueRequest, position);
+////                }
+////            }
+////        }).start();
+//        String requestId = Convert.toStr(redisTemplate.opsForZSet().popMin(queueKey));
+//        System.out.println("requestId = " + requestId);
 //    }
 //
 //    private void processRequest(QueueRequest queueRequest) {

src/main/java/com/backendsys/modules/queue/service/TaskQueueService.java

@@ -1,4 +0,0 @@
-package com.backendsys.modules.queue.service;
-
-public interface TaskQueueService {
-}

src/main/java/com/backendsys/modules/queue/service/TaskService.java

@@ -0,0 +1,35 @@
+package com.backendsys.modules.queue.service;
+
+import com.backendsys.modules.queue.entity.Entire;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.PendingMessages;
+import org.springframework.data.redis.connection.stream.RecordId;
+
+import java.util.List;
+import java.util.Map;
+
+public interface TaskService {
+
+    // 初始化消费者组
+    void initConsumerGroup(String stream_key, String group_name);
+
+    // 获取任务队列
+    List<MapRecord<String, String, String>> getTaskList(String stream_key);
+    // 获取任务队列 (未消费)
+    List<MapRecord<String, String, String>> getTaskPaddingList(String stream_key);
+
+    // 添加任务
+    RecordId addTask(String stream_key, Map<String, Object> task_data);
+    // 移除任务
+    void removeTask(String stream_key, String task_id);
+
+    // 消费任务
+    void consumeTask(String stream_key, String task_id);
+
+//    // 消费任务
+//    void consumeTask(String stream_key);
+//    // 获取任务队列
+//    List<MapRecord<String, String, String>> getTaskList(String stream_key);
+//    List<MapRecord<String, String, String>> getTaskList(String stream_key, Entire entire);
+
+}

src/main/java/com/backendsys/modules/queue/service/impl/TaskQueueServiceImpl.java

@@ -1,61 +0,0 @@
-package com.backendsys.modules.queue.service.impl;
-
-import com.backendsys.modules.queue.entity.GenerateRequest;
-import com.backendsys.modules.queue.service.TaskQueueService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.data.redis.connection.stream.ObjectRecord;
-import org.springframework.data.redis.connection.stream.StreamRecords;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.stereotype.Service;
-
-import java.time.Instant;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-@Service
-public class TaskQueueServiceImpl implements TaskQueueService {
-
-    @Value("${spring.data.redis.stream.key}")
-    private String streamKey;
-
-    @Autowired
-    private RedisTemplate redisTemplate;
-
-    // 添加任务到Stream
-    public String addTask(Map<String, String> taskData) {
-        ObjectRecord<String, Map<String, String>> record =
-                StreamRecords.newRecord()
-                        .ofObject(taskData)
-                        .withStreamKey(streamKey);
-
-        return redisTemplate.opsForStream()
-                .add(record)
-                .getValue();
-    }
-
-    // 初始化任务状态（Redis Hash）
-    public void initTaskStatus(String taskId, GenerateRequest request) {
-        String statusKey = "task:status:" + taskId;
-
-        Map<String, String> status = new HashMap<>();
-        status.put("status", "QUEUED");
-        status.put("progress", "0");
-        status.put("prompt", request.getPrompt());
-        status.put("createdAt", Instant.now().toString());
-
-        redisTemplate.opsForHash().putAll(statusKey, status);
-
-        // 设置24小时过期
-        redisTemplate.expire(statusKey, 24, TimeUnit.HOURS);
-    }
-
-    // 获取队列位置
-    public String getQueuePosition(String taskId) {
-        // 使用XPENDING获取任务在消费者组中的位置
-        // 实际实现需根据业务逻辑计算
-        return "10"; // 示例值
-    }
-
-}

src/main/java/com/backendsys/modules/queue/service/impl/TaskServiceImpl.java

@@ -0,0 +1,202 @@
+package com.backendsys.modules.queue.service.impl;
+
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.core.util.StrUtil;
+import com.backendsys.modules.queue.entity.Entire;
+import com.backendsys.modules.queue.service.TaskService;
+import org.springframework.data.domain.Range;
+import jakarta.annotation.PostConstruct;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.connection.Limit;
+import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StreamOperations;
+import org.springframework.data.redis.stream.StreamListener;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.stereotype.Service;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static com.volcengine.model.tls.Const.STREAM;
+
+/**
+ * 任务队列服务
+ */
+@Service
+public class TaskServiceImpl implements TaskService, StreamListener<String, MapRecord<String,String,String>> {
+
+    @Autowired
+    private StreamMessageListenerContainer<String, MapRecord<String,String,String>> container;
+
+    @Autowired
+    private RedisTemplate redisTemplate;
+
+    /**
+     * 初始化消费者组
+     */
+    public void initConsumerGroup(String stream_key, String group_name) {
+        System.out.println("initConsumerGroup: [stream_key = " + stream_key + ", group_name = " + group_name + "]");
+        try {
+            redisTemplate.opsForStream().createGroup(stream_key, group_name);
+        } catch (RedisSystemException e) {
+            if (!e.getCause().getMessage().contains("BUSYGROUP")) {
+                throw e; // 忽略"组已存在"错误
+            }
+        }
+        container.receive(Consumer.from(group_name, "my-consumer"),
+                StreamOffset.create(STREAM, ReadOffset.lastConsumed()), this);
+    }
+
+    /**
+     * 获取任务队列 (显示前5条)
+     */
+    @Override
+    public List<MapRecord<String, String, String>> getTaskList(String stream_key) {
+        return redisTemplate.opsForStream()
+                .reverseRange(stream_key, Range.unbounded(), Limit.limit().count(5));
+    }
+
+
+
+
+
+    /**
+     * 获取任务队列 (未消费)
+     */
+    @Override
+    public List<MapRecord<String, String, String>> getTaskPaddingList(String stream_key) {
+        String group_name = stream_key + ":group";
+
+        // 1. 先确保流和消费者组存在，只需建一次
+        String stream = "my-stream";
+        String group = "my-group";
+        try {   // 如果组已存在会抛异常，忽略即可
+            redisTemplate.opsForStream().createGroup(stream, ReadOffset.latest(), group);
+        } catch (Exception ignored) {}
+
+        // 2. 轮询消费
+        StreamOperations<String, String, String> ops = redisTemplate.opsForStream();
+        while (true) {
+            List<MapRecord<String, String, String>> list = ops.read(
+                    Consumer.from(group, "sync-consumer"),
+                    StreamReadOptions.empty().count(5).block(Duration.ofSeconds(3)),
+                    StreamOffset.create(stream, ReadOffset.lastConsumed())
+            );
+            if (list.isEmpty()) continue;
+
+            for (MapRecord<String, String, String> r : list) {
+                System.out.println("收到消息 id=" + r.getId() + " body=" + r.getValue());
+                // 3. ack
+                redisTemplate.opsForStream().acknowledge(stream, group, r.getId());
+            }
+        }
+
+    }
+
+    /**
+     * 消费任务 (不移除)
+     */
+    @Override
+    public void consumeTask(String stream_key, String task_id) {
+        String group_name = stream_key + ":group";
+
+        // 用组里任意消费者读一条（不 ACK）
+        List<MapRecord<String, String, String>> records = redisTemplate.opsForStream()
+                .read(Consumer.from(group_name, "my-consumer"), StreamOffset.create(stream_key, ReadOffset.from(">")));
+        System.out.println("records = " + records);
+
+
+//        Long acknowledged = redisTemplate.opsForStream().acknowledge(stream_key, stream_key + ":group", task_id);
+//        System.out.println("acknowledged = " + acknowledged);
+    }
+
+
+
+
+
+
+    /**
+     * 添加任务
+     */
+    @Override
+    public RecordId addTask(String stream_key, Map<String, Object> task_data) {
+        ObjectRecord<String, Map<String, Object>> task = StreamRecords.newRecord()
+            .ofObject(task_data)
+            .withStreamKey(stream_key);
+        // 添加任务
+        return redisTemplate.opsForStream().add(task);
+    }
+
+    /**
+     * 移除任务
+     */
+    @Override
+    public void removeTask(String stream_key, String task_id) {
+        redisTemplate.opsForStream().delete(stream_key, task_id);
+    }
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+    }
+
+
+//        // 1. 安全获取消息
+//        List<MapRecord<String, String, String>> task_list = redisTemplate.opsForStream()
+//                .read(Consumer.from(stream_key + ":group", "consumer-1"),
+//                        StreamReadOptions.empty().count(1),
+//                        StreamOffset.create(stream_key, ReadOffset.lastConsumed()));
+
+//        System.out.println("(consumeTask) task_list = " + task_list);
+
+//        if (task_list != null && !task_list.isEmpty()) {
+//            // 2. 处理消息
+//            MapRecord<String, String, String> record = task_list.get(0);
+//            try {
+                // 实际业务处理（如调用文生图API）
+                // processTask(record.getValue());
+
+//                // 3. 确认消息（ACK）
+//                redisTemplate.opsForStream().acknowledge(stream_key, stream_key + ":group", record.getId());
+
+                // 在ACK后追加删除操作
+//                redisTemplate.opsForStream().delete(stream_key, record.getId());
+
+//            } catch (Exception e) {
+//                System.out.println("任务处理失败: " + record.getId() + ", " + e.getMessage());
+//            }
+//        }
+
+//    }
+
+    /**
+     * 获取任务队列
+     */
+//    @Override
+//    public List<MapRecord<String, String, String>> getTaskList(String stream_key) {
+//        return getTaskList(stream_key, null);
+//    }
+//    @Override
+//    public List<MapRecord<String, String, String>> getTaskList(String stream_key, Entire entire) {
+//
+//        // 1. 读取全部数据
+//        List<MapRecord<String, String, String>> task_list = redisTemplate.opsForStream().read(StreamOffset.fromStart(stream_key));
+//
+//        // 2. 根据 entire 过滤
+//        if (entire != null) {
+//            if (StrUtil.isNotEmpty(entire.getKey()) && ObjectUtil.isNotEmpty(entire.getValue())) {
+//                task_list.stream().filter(task -> {
+//                    return entire.getValue().equals(task.getValue().get(entire.getKey()));
+//                }).collect(Collectors.toList());
+//            }
+//        }
+//        return task_list;
+//    }
+
+
+}

src/main/java/com/backendsys/modules/queue/service/impl/__TaskServiceImpl_bak.java

@@ -0,0 +1,199 @@
+//package com.backendsys.modules.queue.service.impl;
+//
+//import com.backendsys.modules.queue.service.TaskService;
+//import org.springframework.beans.factory.annotation.Autowired;
+//import org.springframework.data.domain.Range;
+//import org.springframework.data.redis.RedisSystemException;
+//import org.springframework.data.redis.connection.Limit;
+//import org.springframework.data.redis.connection.stream.*;
+//import org.springframework.data.redis.core.RedisTemplate;
+//import org.springframework.stereotype.Service;
+//
+//import java.util.List;
+//import java.util.Map;
+//
+///**
+// * 任务队列服务
+// */
+//@Service
+//public class __TaskServiceImpl_bak implements TaskService {
+//
+//    @Autowired
+//    private RedisTemplate redisTemplate;
+//
+//    /**
+//     * 初始化消费者组
+//     */
+//    public void initConsumerGroup(String stream_key, String group_name) {
+//        System.out.println("initConsumerGroup: [stream_key = " + stream_key + ", group_name = " + group_name + "]");
+//        try {
+//            redisTemplate.opsForStream().createGroup(stream_key, group_name);
+//        } catch (RedisSystemException e) {
+//            if (!e.getCause().getMessage().contains("BUSYGROUP")) {
+//                throw e; // 忽略"组已存在"错误
+//            }
+//        }
+//    }
+//
+//    /**
+//     * 获取任务队列 (显示前5条)
+//     */
+//    @Override
+//    public List<MapRecord<String, String, String>> getTaskList(String stream_key) {
+//        return redisTemplate.opsForStream()
+//                .reverseRange(stream_key, Range.unbounded(), Limit.limit().count(5));
+//    }
+//
+//
+//
+//
+//
+//    /**
+//     * 获取任务队列 (未消费)
+//     */
+//    @Override
+//    public List<MapRecord<String, String, String>> getTaskPaddingList(String stream_key) {
+//        String group_name = stream_key + ":group";
+//
+////        /* 1. 已读未 ACK 的 Pending 消息 */
+////        List<MapRecord<String, String, String>> pendingRecords = Collections.emptyList();
+////        PendingMessagesSummary summary = redisTemplate.opsForStream().pending(stream_key, group_name);
+////        System.out.println("total = " + summary.getTotalPendingMessages());
+////        if (summary.getTotalPendingMessages() > 0) {
+////            Range<String> idRange = summary.getIdRange();
+////            System.out.println("idRange = " + idRange);
+////            pendingRecords = redisTemplate.opsForStream().range(stream_key, idRange);
+////            System.out.println("pendingRecords = " + pendingRecords);
+////        }
+////        return pendingRecords;
+//
+//        StreamInfo.XInfoGroups groups = redisTemplate.opsForStream().groups(stream_key);
+//        System.out.println("groups = " + groups);
+//
+//        String lastDelivered = groups.stream()
+//                .filter(g -> group_name.equals(g.groupName()))
+//                .findFirst()
+//                .map(StreamInfo.XInfoGroup::lastDeliveredId)
+//                .orElse("0-0");
+//        System.out.println("lastDelivered = " + lastDelivered);
+//
+//        List<MapRecord<String, String, String>> unread =
+//                redisTemplate.opsForStream()
+//                        .read(Consumer.from(group_name, "my-consumer"),
+//                                StreamOffset.create(stream_key,
+//                                        ReadOffset.from(lastDelivered)));
+//        System.out.println("unread = " + unread);
+//        return unread;
+//
+////        PendingMessagesSummary summary = redisTemplate.opsForStream().pending(stream_key, group_name);
+////        Range<String> idRange = summary.getIdRange();
+////
+////        long total = summary.getTotalPendingMessages();
+////        String minId = idRange.getLowerBound().getValue().orElse(null);
+////        String maxId = idRange.getUpperBound().getValue().orElse(null);
+////        System.out.println("total = " + total + ", minId: " + minId + ", maxId: " + maxId);
+////
+////        // 用得到的范围再去查询完整消息
+////        List<MapRecord<String, String, String>> records = redisTemplate.opsForStream().range(stream_key, idRange);
+////        return records;
+//
+//    }
+//
+//    /**
+//     * 消费任务 (不移除)
+//     */
+//    @Override
+//    public void consumeTask(String stream_key, String task_id) {
+//        String group_name = stream_key + ":group";
+//
+//        // 用组里任意消费者读一条（不 ACK）
+//        List<MapRecord<String, String, String>> records = redisTemplate.opsForStream()
+//                .read(Consumer.from(group_name, "my-consumer"), StreamOffset.create(stream_key, ReadOffset.from(">")));
+//        System.out.println("records = " + records);
+//
+//
+////        Long acknowledged = redisTemplate.opsForStream().acknowledge(stream_key, stream_key + ":group", task_id);
+////        System.out.println("acknowledged = " + acknowledged);
+//    }
+//
+//
+//
+//
+//
+//
+//    /**
+//     * 添加任务
+//     */
+//    @Override
+//    public RecordId addTask(String stream_key, Map<String, Object> task_data) {
+//        ObjectRecord<String, Map<String, Object>> task = StreamRecords.newRecord()
+//            .ofObject(task_data)
+//            .withStreamKey(stream_key);
+//        // 添加任务
+//        return redisTemplate.opsForStream().add(task);
+//    }
+//
+//    /**
+//     * 移除任务
+//     */
+//    @Override
+//    public void removeTask(String stream_key, String task_id) {
+//        redisTemplate.opsForStream().delete(stream_key, task_id);
+//    }
+//
+//
+//
+////        // 1. 安全获取消息
+////        List<MapRecord<String, String, String>> task_list = redisTemplate.opsForStream()
+////                .read(Consumer.from(stream_key + ":group", "consumer-1"),
+////                        StreamReadOptions.empty().count(1),
+////                        StreamOffset.create(stream_key, ReadOffset.lastConsumed()));
+//
+////        System.out.println("(consumeTask) task_list = " + task_list);
+//
+////        if (task_list != null && !task_list.isEmpty()) {
+////            // 2. 处理消息
+////            MapRecord<String, String, String> record = task_list.get(0);
+////            try {
+//                // 实际业务处理（如调用文生图API）
+//                // processTask(record.getValue());
+//
+////                // 3. 确认消息（ACK）
+////                redisTemplate.opsForStream().acknowledge(stream_key, stream_key + ":group", record.getId());
+//
+//                // 在ACK后追加删除操作
+////                redisTemplate.opsForStream().delete(stream_key, record.getId());
+//
+////            } catch (Exception e) {
+////                System.out.println("任务处理失败: " + record.getId() + ", " + e.getMessage());
+////            }
+////        }
+//
+////    }
+//
+//    /**
+//     * 获取任务队列
+//     */
+////    @Override
+////    public List<MapRecord<String, String, String>> getTaskList(String stream_key) {
+////        return getTaskList(stream_key, null);
+////    }
+////    @Override
+////    public List<MapRecord<String, String, String>> getTaskList(String stream_key, Entire entire) {
+////
+////        // 1. 读取全部数据
+////        List<MapRecord<String, String, String>> task_list = redisTemplate.opsForStream().read(StreamOffset.fromStart(stream_key));
+////
+////        // 2. 根据 entire 过滤
+////        if (entire != null) {
+////            if (StrUtil.isNotEmpty(entire.getKey()) && ObjectUtil.isNotEmpty(entire.getValue())) {
+////                task_list.stream().filter(task -> {
+////                    return entire.getValue().equals(task.getValue().get(entire.getKey()));
+////                }).collect(Collectors.toList());
+////            }
+////        }
+////        return task_list;
+////    }
+//
+//
+//}

src/main/resources/application-dev.yml

@@ -47,6 +47,7 @@ spring:
       host: 172.19.0.7
       port: 6388
       password: p1FM!fkfPdBQ%@5o
+
 #    cache:
 #      type: redis
 #      redis:
@@ -199,4 +200,5 @@ klingai:
 comfyui:
   host: 127.0.0.1
   token: $2b$12$.MR4qGaFetN1FPQzbfyIrehsyjnPJ12xAZhR/l7KZpLkUPQTCG4gy
-  is-save: true
+  is-save: true
+  queue-key: comfyui:queue

src/main/resources/application-local.yml

@@ -47,9 +47,6 @@ spring:
       host: 127.0.0.1
       port: 6388
       password: 123456
-      stream:
-        key: comfyui-stream-task      # 任务队列名称
-        group: comfyui-stream-group   # 消费者组名
 
 #    cache:
 #      type: redis
@@ -216,4 +213,5 @@ klingai:
 comfyui:
   host: 43.128.1.201
   token: $2b$12$.MR4qGaFetN1FPQzbfyIrehsyjnPJ12xAZhR/l7KZpLkUPQTCG4gy
-  is-save: true
+  is-save: true
+  queue-key: comfyui:queue

src/main/resources/application-prod.yml

@@ -47,6 +47,7 @@ spring:
       host: 172.19.0.6
       port: 6388
       password: stI2gmsq$Y9z3vdT
+
 #    cache:
 #      type: redis
 #      redis:
@@ -200,4 +201,5 @@ klingai:
 comfyui:
   host: 127.0.0.1
   token: $2b$12$.MR4qGaFetN1FPQzbfyIrehsyjnPJ12xAZhR/l7KZpLkUPQTCG4gy
-  is-save: true
+  is-save: true
+  queue-key: comfyui:queue