队列跑通90%

db/__comfyui_task.sql

@@ -0,0 +1,21 @@
+# /**
+# Source Server Version: 8.0.31
+# Source Database: backendsys
+# Date: 2025/06/03 10:09:22
+# */
+#
+# DROP TABLE IF EXISTS `comfyui_task`;
+# CREATE TABLE `comfyui_task` (
+#     PRIMARY KEY (`id`),
+#     `id` BIGINT AUTO_INCREMENT COMMENT 'ID',
+#     `user_id` BIGINT NOT NULL COMMENT '用户ID',
+#     `client_id` VARCHAR(255) COMMENT 'Client ID',
+#     `task_type` VARCHAR(255) NOT NULL COMMENT '任务类型 (Text2Image, ..)',
+#     `task_status` TINYINT DEFAULT '-1' COMMENT '任务状态 (-1:未提交, 1:已提交)',
+#     `generate_request` TEXT COMMENT '任务请求原始参数 (JSONString)',
+#     `in_master` TINYINT DEFAULT '-1' COMMENT '队列状态 (-1:未进主队列, 1:已进主队列)',
+#     `create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+#     `update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+#     INDEX `idx_user_id` (`user_id`),
+#     INDEX `idx_in_master` (`in_master`)
+# ) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='任务表';

db/comfyui_task.sql

@@ -11,11 +11,18 @@ CREATE TABLE `comfyui_task` (
     `user_id` BIGINT NOT NULL COMMENT '用户ID',
     `client_id` VARCHAR(255) COMMENT 'Client ID',
     `task_type` VARCHAR(255) NOT NULL COMMENT '任务类型 (Text2Image, ..)',
-    `task_status` TINYINT DEFAULT '-1' COMMENT '任务状态 (-1:未提交, 1:已提交)',
+    `execute_prompt_id` VARCHAR(255) COMMENT '任务提交ID',
+    `execute_status` TINYINT DEFAULT '-1' COMMENT '任务执行状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)',
+    `execute_url` VARCHAR(2000) COMMENT '任务执行URL',
+    `execute_url_port` INT COMMENT '任务执行URL端口',
     `generate_request` TEXT COMMENT '任务请求原始参数 (JSONString)',
+    `generate_response` TEXT COMMENT '任务生成原始结果 (JSONString)',
+    `reason` TEXT COMMENT '当任务失败时展示失败原因（如触发平台的内容风控等）',
     `in_master` TINYINT DEFAULT '-1' COMMENT '队列状态 (-1:未进主队列, 1:已进主队列)',
+    `custom_params` VARCHAR(255) COMMENT '自定义参数',
     `create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
     `update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+    INDEX `idx_execute_prompt_id` (`execute_prompt_id`),
     INDEX `idx_user_id` (`user_id`),
     INDEX `idx_in_master` (`in_master`)
-) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='任务表';
+) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='任务执行表';

db/comfyui_task_execute.sql

@@ -1,26 +0,0 @@
-/**
-Source Server Version: 8.0.31
-Source Database: backendsys
-Date: 2025/06/03 10:09:22
-*/
-
-DROP TABLE IF EXISTS `comfyui_task_execute`;
-CREATE TABLE `comfyui_task_execute` (
-    PRIMARY KEY (`id`),
-    `id` BIGINT AUTO_INCREMENT COMMENT 'ID',
-    `task_id` BIGINT NOT NULL COMMENT '任务ID',
-    `user_id` BIGINT NOT NULL COMMENT '用户ID',
-    `client_id` VARCHAR(255) COMMENT 'Client ID',
-    `task_type` VARCHAR(255) NOT NULL COMMENT '任务类型 (Text2Image, ..)',
-    `execute_prompt_id` VARCHAR(255) NOT NULL COMMENT '任务提交ID',
-    `execute_status` TINYINT DEFAULT '-1' COMMENT '任务执行状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)',
-    `execute_url` VARCHAR(2000) COMMENT '任务执行URL',
-    `execute_url_port` INT COMMENT '任务执行URL端口',
-    `generate_request` TEXT COMMENT '任务请求原始参数 (JSONString)',
-    `generate_response` TEXT COMMENT '任务生成原始结果 (JSONString)',
-    `reason` TEXT COMMENT '当任务失败时展示失败原因（如触发平台的内容风控等）',
-    `create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
-    `update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
-    INDEX `idx_execute_prompt_id` (`execute_prompt_id`),
-    INDEX `idx_user_id` (`user_id`)
-) ENGINE=INNODB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='任务执行表';

src/main/java/com/backendsys/modules/TestController.java

@@ -127,6 +127,7 @@ public class TestController {
         Channel channel = connection.createChannel(false);
 
         // 2. 手动拉消息，第二个参数传 false 表示“不要自动 ack”
+//        GetResponse resp = channel.basicGet("demo.queue", false);
         GetResponse resp = channel.basicGet("demo.queue", false);
         if (resp == null) {
             channel.close();

src/main/java/com/backendsys/modules/common/config/rabbitmq/DemoRabbitListener.java

@@ -2,11 +2,18 @@ package com.backendsys.modules.common.config.rabbitmq;
 
 import com.rabbitmq.client.Channel;
 import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
 
+import org.springframework.amqp.rabbit.connection.Connection;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.GetResponse;
+
 /**
  * 自定义监听器
  */
@@ -14,6 +21,9 @@ import java.io.IOException;
 @Lazy(false)
 public class DemoRabbitListener {
 
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
 //    // [监听] 自动 ACK
 //    @ComfyuiRabbitListener(id = "demoContainer", queues = "demo.queue", autoStartup = "false" )
 //    public void receive(SysUser sysUser) {
@@ -22,7 +32,7 @@ public class DemoRabbitListener {
 
 
 //    // [监听] 手动 ACK
-//    @ComfyuiRabbitListener(id = "demoContainer", queues = "demo.queue", ackMode = "MANUAL")
+//    @RabbitListener(id = "demoContainer", queues = "demo.queue", ackMode = "MANUAL")
 //    public void receive(Message message, Channel channel) throws IOException, InterruptedException {
 //        try {
 //            // 1. 模拟耗时业务
@@ -31,18 +41,34 @@ public class DemoRabbitListener {
 //            // 2. 业务处理
 //            String body = new String(message.getBody(), StandardCharsets.UTF_8);
 //            System.out.println("处理消息 (8s): " + body);
-//
+
+
+//            // 外部调用ACK (消耗队列)：
+//            try {
+//                Connection connection = rabbitTemplate.getConnectionFactory().createConnection();
+//                Channel channel = connection.createChannel(false);
+//                GetResponse resp = channel.basicGet("demo.queue", false);
+//                if (resp != null) {
+//                    channel.basicAck(resp.getEnvelope().getDeliveryTag(), false);
+//                }
+//                channel.close();
+//                connection.close();
+//            } catch (Exception e) {
+//                System.out.println(e.getMessage());
+//            }
+
+
 //            // 3. 手动确认（deliveryTag + 是否批量）
 //            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
 //        } catch (Exception e) {
-//            System.err.println("处理失败 (10s): " + e.getMessage());
+//            System.err.println("(RabbitListener) 处理失败: " + e.getMessage());
 //            // 拒绝并重新入队
 //            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
 //        }
 //    }
 
     // [监听] 死信队列
-    @org.springframework.amqp.rabbit.annotation.RabbitListener(id = "demoDlxContainer", queues = "demo.dlx.queue", ackMode = "MANUAL")
+    @RabbitListener(id = "demoDlxContainer", queues = "demo.dlx.queue", ackMode = "MANUAL")
     public void handleDemoDlx(Message message, Channel channel) throws IOException {
         System.out.println("[Demo][RabbitMQ-死信队列]收到：" + new String(message.getBody()));
 

src/main/java/com/backendsys/modules/crt/service/impl/CrtGenerateServiceImpl.java

@@ -2,7 +2,10 @@ package com.backendsys.modules.crt.service.impl;
 
 import cn.hutool.core.convert.Convert;
 import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.json.JSONUtil;
 import com.backendsys.exception.CustException;
+import com.backendsys.modules.common.config.security.utils.SecurityUtil;
 import com.backendsys.modules.crt.dao.CrtDramaProjectSettingsDao;
 import com.backendsys.modules.crt.dao.CrtDramaProjectStoryboardDao;
 import com.backendsys.modules.crt.dao.CrtGenerateImageDao;
@@ -19,6 +22,7 @@ import com.backendsys.modules.sdk.comfyui.entity.ComfyuiQueue;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiText2Image;
 import com.backendsys.modules.sdk.comfyui.service.ComfyuiService;
 import com.backendsys.modules.sdk.comfyui.service.ComfyuiSocketService;
+import com.backendsys.modules.sdk.comfyui.service.ComfyuiTaskService;
 import com.backendsys.modules.sdk.comfyui.service.ComfyuiText2ImageService;
 import com.backendsys.modules.upload.enums.StyleEnums;
 import com.backendsys.modules.upload.utils.UploadUtil;
@@ -31,10 +35,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
 import java.util.stream.Collectors;
 
 @Service
@@ -47,6 +48,8 @@ public class CrtGenerateServiceImpl implements CrtGenerateService {
     private ComfyuiService comfyuiService;
     @Autowired
     private ComfyuiSocketService comfyuiSocketService;
+    @Autowired
+    private ComfyuiTaskService comfyuiTaskService;
 
     @Autowired
     private ComfyuiText2ImageService comfyuiText2ImageService;
@@ -153,8 +156,15 @@ public class CrtGenerateServiceImpl implements CrtGenerateService {
 
         // -- [ComfyUI] 发起生图任务 -----------------------------------------------
         Long task_id = comfyuiText2ImageService.generateText2Image(client_id, comfyuiText2Image);
-
-
+        System.out.println("[创建生图任务] task_id = " + task_id);
+
+        // -- [Rabbitmq] 推送到生成队列 --------------------------------------------
+        Long user_id = SecurityUtil.getUserId();
+        Map<String, Object> map = new HashMap<>();
+        map.put("drama_project_storyboard_id", drama_project_storyboard_id);
+        map.put("user_id", user_id);
+        String custom_params = JSONUtil.toJsonStr(map);
+        comfyuiTaskService.tryPushNext(user_id, custom_params);
 
 
 

src/main/java/com/backendsys/modules/sdk/comfyui/dao/ComfyuiTaskExecuteDao.java

@@ -1,9 +1,9 @@
 package com.backendsys.modules.sdk.comfyui.dao;
 
-import com.backendsys.modules.sdk.comfyui.entity.ComfyuiTaskExecute;
+import com.backendsys.modules.sdk.comfyui.entity.ComfyuiTask;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import org.apache.ibatis.annotations.Mapper;
 
 @Mapper
-public interface ComfyuiTaskExecuteDao extends BaseMapper<ComfyuiTaskExecute> {
+public interface ComfyuiTaskExecuteDao extends BaseMapper<ComfyuiTask> {
 }

src/main/java/com/backendsys/modules/sdk/comfyui/entity/ComfyuiTask.java

@@ -14,9 +14,17 @@ public class ComfyuiTask {
     private Long user_id;                 // 用户ID
     private String client_id;             // Client ID
     private String task_type;             // 任务类型 (Text2Image, ..)
-    private Integer task_status;          // 任务状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)
+    private String execute_prompt_id;     // 任务ID
+    private Integer execute_status;       // 任务状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)
+    private String execute_url;           // 任务执行URL
+    private Integer execute_url_port;     // 任务执行URL端口
+
     private String generate_request;      // 任务请求原始参数 (JSONString)
+    private String generate_response;     // 任务生成原始结果 (JSONString)
+    private String reason;                // 任务状态信息，当任务失败时展示失败原因（如触发平台的内容风控等）
+
     private Integer in_master;            // 队列状态 (-1:未进主队列, 1:已进主队列)
+    private String custom_params;         // 自定义参数
     private String create_time;
     private String update_time;
 

src/main/java/com/backendsys/modules/sdk/comfyui/entity/ComfyuiTaskExecute.java

@@ -1,29 +0,0 @@
-package com.backendsys.modules.sdk.comfyui.entity;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.Data;
-
-@Data
-@TableName("comfyui_task_execute")
-public class ComfyuiTaskExecute {
-
-    @TableId(type = IdType.AUTO)
-    private Long id;
-    private Long task_id;
-    private Long user_id;                 // 用户ID
-    private String client_id;             // Client ID
-    private String task_type;             // 任务类型 (Text2Image, ..)
-    private String execute_prompt_id;     // 任务ID
-    private Integer execute_status;       // 任务状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)
-    private String execute_url;           // 任务执行URL
-    private Integer execute_url_port;     // 任务执行URL端口
-
-    private String generate_request;      // 任务请求原始参数 (JSONString)
-    private String generate_response;     // 任务生成原始结果 (JSONString)
-    private String reason;                // 任务状态信息，当任务失败时展示失败原因（如触发平台的内容风控等）
-    private String create_time;
-    private String update_time;
-
-}

src/main/java/com/backendsys/modules/sdk/comfyui/entity/__ComfyuiTask.java

@@ -0,0 +1,23 @@
+//package com.backendsys.modules.sdk.comfyui.entity;
+//
+//import com.baomidou.mybatisplus.annotation.IdType;
+//import com.baomidou.mybatisplus.annotation.TableId;
+//import com.baomidou.mybatisplus.annotation.TableName;
+//import lombok.Data;
+//
+//@Data
+//@TableName("comfyui_task")
+//public class ComfyuiTask {
+//
+//    @TableId(type = IdType.AUTO)
+//    private Long id;
+//    private Long user_id;                 // 用户ID
+//    private String client_id;             // Client ID
+//    private String task_type;             // 任务类型 (Text2Image, ..)
+//    private Integer task_status;          // 任务状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)
+//    private String generate_request;      // 任务请求原始参数 (JSONString)
+//    private Integer in_master;            // 队列状态 (-1:未进主队列, 1:已进主队列)
+//    private String create_time;
+//    private String update_time;
+//
+//}

src/main/java/com/backendsys/modules/sdk/comfyui/rabbitmq/ComfyuiQueueDlxConfig.java

@@ -14,7 +14,7 @@ import org.springframework.context.annotation.Lazy;
 public class ComfyuiQueueDlxConfig {
 
     /* === 死信交换机/队列 === */
-    public static final String EXCHANGE = "comfyui.dlx";
+    public static final String EXCHANGE = "comfyui.dlx.exchange";
     public static final String QUEUE = "comfyui.dlx.queue";
     public static final String ROUTING_KEY = "comfyui.dlx.routekey";
 

src/main/java/com/backendsys/modules/sdk/comfyui/rabbitmq/ComfyuiRabbitListener.java

@@ -1,12 +1,29 @@
 package com.backendsys.modules.sdk.comfyui.rabbitmq;
 
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+import com.backendsys.modules.sdk.comfyui.entity.ComfyuiResponse;
+import com.backendsys.modules.sdk.comfyui.service.ComfyuiService;
+import com.backendsys.modules.sdk.comfyui.service.ComfyuiSocketService;
 import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.GetResponse;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.stereotype.Component;
+import reactor.core.publisher.Mono;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * 自定义监听器
@@ -15,10 +32,75 @@ import java.io.IOException;
 @Lazy(false)
 public class ComfyuiRabbitListener {
 
+    @Autowired
+    private ComfyuiService comfyuiService;
+    @Autowired
+    private ComfyuiQueueConfig config;
+    @Autowired
+    private DeliveryTagHolder deliveryTagHolder;
+
+
+    // [监听] 手动 ACK
+    // (从 ComfyuiTaskService.tryPushNext 过来)
+    @RabbitListener(id = "ComfyuiContainer", queues = "comfyui.queue", ackMode = "MANUAL")
+    public void receive(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
+        try {
+
+            System.out.println("[Rabbitmq] 开始监听队列");
+
+            // 2. 业务处理 (提交Comfyui任务)
+            String body = new String(message.getBody(), StandardCharsets.UTF_8);
+            // {"id": 24, "user_id": 1, "client_id": "..", "generate_request" :"..", ..}
+
+            // 收到的参数:
+            JSONObject body_object = JSONUtil.parseObj(body);
+            String prompt = body_object.getStr("generate_request");
+            String client_id = body_object.getStr("client_id");
+            String custom_params = body_object.getStr("custom_params");
+
+            JSONObject prompt_object = JSONUtil.parseObj(prompt);
+
+            // [ComfyUI] 发起任务
+            Mono<ComfyuiResponse> cfPromptResponseMono = comfyuiService.prompt(client_id, prompt_object, custom_params);
+            ComfyuiResponse response = cfPromptResponseMono.block();
+            System.out.println("[Comfyui][发起任务] response = " + response);
+
+            // 2. 暂存 deliveryTag
+            deliveryTagHolder.put(client_id, deliveryTag);
+
+            /*
+                外部调用ACK (确认消息)：
+                try {
+                    Connection connection = rabbitTemplate.getConnectionFactory().createConnection();
+                    Channel channel = connection.createChannel(false);
+                    GetResponse resp = channel.basicGet("demo.queue", false);
+                    if (resp != null) {
+                        channel.basicAck(resp.getEnvelope().getDeliveryTag(), false);
+                    }
+                    channel.close();
+                    connection.close();
+                } catch (Exception e) {
+                    System.err.println("(RabbitListener) 处理失败: " + e.getMessage());
+                }
+             */
+
+            // 3. 手动确认（deliveryTag + 是否批量）
+            // channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+
+        } catch (Exception e) {
+            System.err.println("(RabbitListener) 处理失败: " + e.getMessage());
+            // 拒绝并重新入队
+            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+        }
+    }
+
+
     // [监听] 死信队列
     @RabbitListener(id = "ComfyuiDlxContainer", queues = "comfyui.dlx.queue", ackMode = "MANUAL")
     public void handleComfyuiDlx(Message message, Channel channel) throws IOException {
-        System.out.println("[Comfyui][RabbitMQ-死信] 收到：" + new String(message.getBody()));
+
+        String body = new String(message.getBody());
+        System.out.println("[Comfyui][RabbitMQ-死信] 收到：" + StrUtil.sub(body, 0, 1000));
 
         // 业务：记录日志 / 重发 / 报警 / 人工补偿
         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

src/main/java/com/backendsys/modules/sdk/comfyui/rabbitmq/DeliveryTagHolder.java

@@ -0,0 +1,20 @@
+package com.backendsys.modules.sdk.comfyui.rabbitmq;
+
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class DeliveryTagHolder {
+    // key = userId（或 sessionId），value = deliveryTag
+    private final Map<String, Long> map = new ConcurrentHashMap<>();
+
+    public void put(String key, Long tag) {
+        map.put(key, tag);
+    }
+
+    public Long take(String key) {
+        return map.remove(key);   // 拿出来后自动删除
+    }
+}

src/main/java/com/backendsys/modules/sdk/comfyui/service/ComfyuiService.java

@@ -15,5 +15,6 @@ public interface ComfyuiService {
 
     // [ComfyUI] 执行任务 (通用)
     Mono<ComfyuiResponse> prompt(String client_id, JSONObject prompt);
+    Mono<ComfyuiResponse> prompt(String client_id, JSONObject prompt, String custom_params);
 
 }

src/main/java/com/backendsys/modules/sdk/comfyui/service/ComfyuiSocketService.java

@@ -14,6 +14,9 @@ public interface ComfyuiSocketService {
     Mono<Void> connectToSse(String clientId, Integer port, Boolean is_save);
     Mono<Void> connectToSse(String clientId, Integer port, Boolean is_save, Map<String, Object> params);
 
+    // 如果是在 websocket 周期，则需要手动输入 user_id
+    Mono<Void> connectToSse(String clientId, Integer port, Boolean is_save, Map<String, Object> params, Long input_user_id);
+
     // [ComfyUI] 断开 WebSocket 监听连接
     Mono<Void> disconnect(String clientId);
 

src/main/java/com/backendsys/modules/sdk/comfyui/service/ComfyuiTaskService.java

@@ -2,7 +2,8 @@ package com.backendsys.modules.sdk.comfyui.service;
 
 public interface ComfyuiTaskService {
 
-    // 推送主队列任务
+    // 推送主队列任务 (以用户为队列单位)
     void tryPushNext(long user_id);
+    void tryPushNext(long user_id, String custom_params);
 
 }

src/main/java/com/backendsys/modules/sdk/comfyui/service/impl/ComfyuiServiceImpl.java

@@ -4,12 +4,15 @@ import cn.hutool.core.convert.Convert;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONArray;
 import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.backendsys.modules.common.Filter.WebClientFilter;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiQueueItem;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiRequest;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiResponse;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiQueue;
 import com.backendsys.modules.sdk.comfyui.service.ComfyuiService;
+import com.backendsys.modules.sdk.comfyui.service.ComfyuiSocketService;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
@@ -30,6 +33,12 @@ public class ComfyuiServiceImpl implements ComfyuiService {
     @Value("${comfyui.ports}")
     private String COMFYUI_PORTS_STR;
 
+    @Autowired
+    private ComfyuiSocketService comfyuiSocketService;
+    @Value("${comfyui.is-save}")
+    private Boolean IS_SAVE;
+
+
     private WebClient webClient;
     public WebClient getWebClient() {
         if (webClient == null) {
@@ -130,13 +139,38 @@ public class ComfyuiServiceImpl implements ComfyuiService {
      */
     @Override
     public Mono<ComfyuiResponse> prompt(String client_id, JSONObject prompt) {
+        return prompt(client_id, prompt, null);
+    }
+    @Override
+    public Mono<ComfyuiResponse> prompt(String client_id, JSONObject prompt, String custom_params) {
+
+        // 获取最优端口
+        Integer port = getBestPort();
+
+        System.out.println("自定义参数 (custom_params) = " + custom_params);
+
+        System.out.println("[ComfyUI] 开始监听 WecSocket 事件");
+        JSONObject custom_params_object = (StrUtil.isNotEmpty(custom_params)) ? JSONUtil.parseObj(custom_params) : new JSONObject();
+        // -- [ComfyUI] 创建 WebSocket 监听进度 --------------------------
+        Map<String, Object> params = new LinkedHashMap<>();
+        Long drama_project_storyboard_id = Convert.toLong(custom_params_object.get("drama_project_storyboard_id"));
+        Long user_id = Convert.toLong(custom_params_object.get("user_id"));
+        if (drama_project_storyboard_id != null) {
+            params.put("drama_project_storyboard_id", drama_project_storyboard_id);
+        }
+        // - is_save: 是否转存到 cos/tos
+        comfyuiSocketService.connectToSse(client_id, port, IS_SAVE, params, user_id).subscribe();
+        // -------------------------------------------------------------
+        // 在任务结束后(SSE)(ComfyuiSocketService.connectToSse):
+        // 1) 手动确认消息(ACK)
+        // 2) 清除队列锁 (QUEUE_LOCK_KEY)
+        // 3) 尝试把下一条任务推进主队列
 
         ComfyuiRequest bodyValue = new ComfyuiRequest();
         bodyValue.setClient_id(client_id);
         bodyValue.setPrompt(prompt);
 
-        String url = "http://" + COMFYUI_HOST + ":8000/prompt";
-
+        String url = "http://" + COMFYUI_HOST + ":" + port + "/prompt";
         WebClient webClient = getWebClient();
         return webClient.post()
                 .uri(url)

src/main/java/com/backendsys/modules/sdk/comfyui/service/impl/ComfyuiSocketServiceImpl.java

@@ -1,20 +1,29 @@
 package com.backendsys.modules.sdk.comfyui.service.impl;
 
 import cn.hutool.core.convert.Convert;
+import cn.hutool.core.map.MapUtil;
 import cn.hutool.json.JSONArray;
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
+import com.backendsys.modules.common.config.redis.utils.RedisUtil;
 import com.backendsys.modules.common.config.security.utils.SecurityUtil;
 import com.backendsys.modules.crt.dao.CrtGenerateImageDao;
 import com.backendsys.modules.crt.entity.CrtGenerateImage;
 import com.backendsys.modules.sdk.comfyui.enums.TaskStatusEnums;
+import com.backendsys.modules.sdk.comfyui.rabbitmq.ComfyuiQueueConfig;
+import com.backendsys.modules.sdk.comfyui.rabbitmq.ComfyuiRabbitListener;
+import com.backendsys.modules.sdk.comfyui.rabbitmq.DeliveryTagHolder;
 import com.backendsys.modules.sdk.comfyui.service.ComfyuiSocketService;
+import com.backendsys.modules.sdk.comfyui.service.ComfyuiTaskService;
 import com.backendsys.modules.sdk.comfyui.utils.ComfyuiUtil;
 import com.backendsys.modules.sse.entity.SseResponse;
 import com.backendsys.modules.sse.entity.SseResponseEnum;
 import com.backendsys.modules.sse.utils.SseUtil;
 import com.backendsys.modules.upload.entity.SysFileResult;
 import com.backendsys.modules.upload.service.SysFileService;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.GetResponse;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -29,24 +38,42 @@ import reactor.netty.http.client.HttpClient;
 import java.net.URI;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.springframework.amqp.rabbit.connection.Connection;
+
 @Service
 public class ComfyuiSocketServiceImpl implements ComfyuiSocketService {
 
     @Autowired
     private SseUtil sseUtil;
 
+    @Autowired
+    private RedisUtil redisUtil;
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+
+    @Autowired
+    private ComfyuiQueueConfig config;
     @Autowired
     private ComfyuiUtil comfyuiUtil;
     @Autowired
+    private ComfyuiTaskService comfyuiTaskService;
+    @Autowired
     private SysFileService sysFileService;
     @Autowired
     private CrtGenerateImageDao crtGenerateImageDao;
+    @Autowired
+    private DeliveryTagHolder deliveryTagHolder;
+
 
+    @Value("${spring.application.name}")
+    private String APPLICATION_NAME;
     @Value("${comfyui.host}")
     private String COMFYUI_HOST;
     @Value("${comfyui.token}")
@@ -121,9 +148,14 @@ public class ComfyuiSocketServiceImpl implements ComfyuiSocketService {
     }
     @Override
     public Mono<Void> connectToSse(String client_id, Integer port, Boolean is_save, Map<String, Object> params) {
+        return connectToSse(client_id, port, is_save, params, null);
+    }
+    // 如果是在 websocket 周期，则需要手动输入 user_id
+    @Override
+    public Mono<Void> connectToSse(String client_id, Integer port, Boolean is_save, Map<String, Object> params, Long input_user_id) {
 
         // 由于 Websocket 获取不到上下文信息，所以 user_id 不能在 socket 周期获取
-        Long user_id = SecurityUtil.getUserId();
+        Long user_id = input_user_id != null ? input_user_id : SecurityUtil.getUserId();
 
         String wsUrl =  "ws://" + COMFYUI_HOST + ":" + port + "/ws";
         return Mono.defer(() -> {
@@ -132,12 +164,11 @@ public class ComfyuiSocketServiceImpl implements ComfyuiSocketService {
             }
             // 动态创建带有认证头的客户端
             WebSocketClient clientWithAuth = createWebSocketClientWithToken(COMFYUI_TOKEN);
-            return clientWithAuth.execute(URI.create(wsUrl + "?clientId=" + client_id), session -> {
+            String wsUrlFull = wsUrl + "?clientId=" + client_id + "&userId=" + user_id;
+            System.out.println("Listen: " + wsUrlFull);
+            return clientWithAuth.execute(URI.create(wsUrlFull), session -> {
                 // 保存会话
                 sessions.put(client_id, session);
-                // 接收消息
-                System.out.println("------ wsUrl: " + wsUrl + " ------");
-                System.out.println("------ connectToSse client_id: " + client_id + ", user_id: " + user_id + " ------");
 
                 Flux<String> incomingMessages = session.receive()
                         .map(WebSocketMessage::getPayloadAsText)
@@ -151,21 +182,82 @@ public class ComfyuiSocketServiceImpl implements ComfyuiSocketService {
                             dataChildren.put("drama_project_storyboard_id", drama_project_storyboard_id);
                             data.put("data", dataChildren);
 
-                            // == [任务执行完成] =======================================================
+                            // == [任务执行完成] =======================================================================
                             // { "type": "executed", .. }
                             String type = Convert.toStr(data.get("type"));
                             if (TaskStatusEnums.EXECUTED.getValue().equals(type)) {
 
+                                // 在任务结束后(SSE)(ComfyuiSocketService.connectToSse):
+                                // 1) 手动确认消息(ACK)
+                                // 2) 清除队列锁 (QUEUE_LOCK_KEY)
+                                // 3) 尝试把下一条任务推进主队列
+
+
+
+
+
+
+
+                                // -- [RabbitMQ][完成当前队列任务，放行下一个队列] ----------------------------------------
+                                try {
+                                    System.out.println("-- 准备放行队列 --");
+                                    Connection connection = rabbitTemplate.getConnectionFactory().createConnection();
+                                    Channel channel = connection.createChannel(false);
+//                                    GetResponse resp = channel.basicGet(config.QUEUE, false);
+
+                                    // 已经被 Listener 的这里拿不到?
+                                    // 这里是空的?
+                                    Long tag = deliveryTagHolder.take(client_id);
+
+                                    // 1) 手动确认消息 (ACK)
+                                    System.out.println("-- 手动确认消息 (ACK) ----------");
+//                                    channel.basicAck(resp.getEnvelope().getDeliveryTag(), false);
+                                    channel.basicAck(tag, false);
+
+
+
+                                    // 2) 清除队列锁 (QUEUE_LOCK_KEY)
+
+                                    String QUEUE_LOCK_KEY = APPLICATION_NAME + ":comfyui:queue:lock:user:" + Convert.toStr(user_id);
+                                    System.out.println("-- 清除队列锁 (QUEUE_LOCK_KEY): " + QUEUE_LOCK_KEY);
+                                    redisUtil.delete(QUEUE_LOCK_KEY);
+
+                                    channel.close();
+                                    connection.close();
+                                } catch (Exception e) {
+                                    System.out.println(e.getMessage());
+                                }
+                                // ----------------------------------------------------------------------------------
+
+
+
+
+
+
+
+
                                 JSONObject output = JSONUtil.parseObj(dataChildren.get("output"));
 
-                                // -- [生成图片 -> 转存图片 -> 新增记录] ------------------------------------------------------
+                                // -- [生成图片 -> 转存图片 -> 新增记录] ------------------------------------------------
                                 // 由于图片地址不是公开的，需要加 Token 访问，因此不能公开返回原始图片地址，比如：
                                 // http://43.128.1.201:8000/api/view?filename=fenjing_1_00012_.png&token=$2b$12$.MR4qGaFetN1FPQzbfyIrehsyjnPJ12xAZhR/l7KZpLkUPQTCG4gy
                                 // 因此，直接在 ComfyUI 的物理目录上构建 nginx 静态资源访问目录
                                 Object imagesObj = output.get("images");
                                 if (imagesObj != null) {
-                                    String prompt_id = Convert.toStr(dataChildren.get("prompt_id"));
 
+                                    // 3) 尝试把下一条任务推进主队列 (生图) ----------------------------------------------
+                                    System.out.println("-- 尝试把下一条任务推进主队列 (生图) --");
+                                    Map<String, Object> map = new HashMap<>();
+                                    map.put("drama_project_storyboard_id", drama_project_storyboard_id);
+                                    map.put("user_id", user_id);
+                                    String custom_params = JSONUtil.toJsonStr(map);
+
+                                    System.out.println("custom_params = " + custom_params);
+
+                                    comfyuiTaskService.tryPushNext(user_id, custom_params);
+                                    // -----------------------------------------------------------------------------
+
+                                    String prompt_id = Convert.toStr(dataChildren.get("prompt_id"));
 
                                     // 临时图片相对路径
                                     JSONArray images = JSONUtil.parseArray(imagesObj);
@@ -186,9 +278,9 @@ public class ComfyuiSocketServiceImpl implements ComfyuiSocketService {
                                     dataChildren.put("output", output);
                                     data.put("data", dataChildren);
 
-                                    // [DB] 执行任务
+                                    // [DB] 更新任务状态
                                     CompletableFuture.runAsync(() -> {
-                                        comfyuiUtil.executeComfyuiTask(prompt_id, JSONUtil.toJsonStr(dataChildren), 2);
+                                        comfyuiUtil.updateComfyuiTask(prompt_id, JSONUtil.toJsonStr(dataChildren), 2);
                                     });
 
                                 }

src/main/java/com/backendsys/modules/sdk/comfyui/service/impl/ComfyuiTaskServiceImpl.java

@@ -1,74 +1,109 @@
 package com.backendsys.modules.sdk.comfyui.service.impl;
 
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.util.StrUtil;
 import com.backendsys.modules.common.config.redis.utils.RedisUtil;
 import com.backendsys.modules.sdk.comfyui.dao.ComfyuiTaskDao;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiTask;
+import com.backendsys.modules.sdk.comfyui.rabbitmq.ComfyuiQueueConfig;
 import com.backendsys.modules.sdk.comfyui.service.ComfyuiTaskService;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 @Service
 public class ComfyuiTaskServiceImpl implements ComfyuiTaskService {
 
-    @Autowired
-    private RedisUtil redisUtil;
+    @Value("${spring.application.name}")
+    private String APPLICATION_NAME;
+
     @Autowired
     private RedisTemplate redisTemplate;
     @Autowired
     private RabbitTemplate rabbitTemplate;
-
-    private static final String MASTER_EX  = "master.job.exchange";
-    private static final String MASTER_RK  = "master.job";
-    private static final String TOKEN_KEY  = "user:%s:master_token";
+    @Autowired
+    private ComfyuiQueueConfig config;
 
     @Autowired
     private ComfyuiTaskDao comfyuiTaskDao;
 
     /**
-     * 推送主队列任务
+     * 推送主队列任务 (以用户为队列单位)
      * 当用户提交新任务、或某任务执行结束后，都调用它
      */
+    @Override
     @Transactional
     public void tryPushNext(long user_id) {
+        tryPushNext(user_id, null);
+    }
+    @Override
+    @Transactional
+    public void tryPushNext(long user_id, String custom_params) {
+
+        System.out.println("[推送主队列任务]");
         // 1. 如果该用户已在主队列里，直接返回
-        String token_key = String.format(TOKEN_KEY, user_id);
-        if (Boolean.TRUE.equals(redisTemplate.hasKey(token_key))) {
+        String QUEUE_LOCK_KEY = APPLICATION_NAME + ":comfyui:queue:lock:user:" + Convert.toStr(user_id);
+        System.out.println("-- (Redis) QUEUE_LOCK_KEY = " + QUEUE_LOCK_KEY);
+        if (Boolean.TRUE.equals(redisTemplate.hasKey(QUEUE_LOCK_KEY))) {
             return;
         }
 
         // 2. 抢令牌（原子）
-        Boolean ok = redisTemplate.opsForValue().setIfAbsent(token_key, "1", Duration.ofMinutes(10));
+        Boolean ok = redisTemplate.opsForValue().setIfAbsent(QUEUE_LOCK_KEY, "1", Duration.ofMinutes(10));
         if (!Boolean.TRUE.equals(ok)) {
             return;          // 抢失败（并发极小概率）
         }
 
-        // 3. 取该用户下一条未进主队列的任务
+        // 3. 该用户下一条未进主队列的任务
         LambdaQueryWrapper<ComfyuiTask> wrapperTask = new LambdaQueryWrapper<>();
         wrapperTask.eq(ComfyuiTask::getUser_id, user_id);
-        wrapperTask.eq(ComfyuiTask::getIn_master, 0);
-        ComfyuiTask comfyuiTask = comfyuiTaskDao.selectOne(wrapperTask);
-
-        if (comfyuiTask == null) {          // 用户已没任务
-            redisTemplate.delete(String.format(TOKEN_KEY, user_id));
+        wrapperTask.eq(ComfyuiTask::getIn_master, -1);
+        List<ComfyuiTask> taskListByUser = comfyuiTaskDao.selectList(wrapperTask);
+        if (taskListByUser == null || taskListByUser.isEmpty()) {
+            System.out.println("用户没有任务了");
+            redisTemplate.delete(String.format(QUEUE_LOCK_KEY, user_id));
             return;
+        } else {
+            System.out.println("用户队列数量: " + taskListByUser.size());
         }
 
+        ComfyuiTask taskByUser = taskListByUser.get(0);
+        wrapperTask.eq(ComfyuiTask::getId, taskByUser.getId());
+
         // 4. 打标 (进主队列)
         ComfyuiTask entity = new ComfyuiTask();
         entity.setIn_master(1);
+        taskByUser.setIn_master(1);
+        taskByUser.setCustom_params(custom_params);
         comfyuiTaskDao.update(entity, wrapperTask);
-        comfyuiTask.setIn_master(1);
 
-        // 5. 发消息
-        rabbitTemplate.convertAndSend(MASTER_EX, MASTER_RK, comfyuiTask);
+//        System.out.println("-- 在这里开始执行任务吗？ --");
+//
+//        String client_id = taskByUser.getClient_id();
+//        System.out.println("task_id: " + taskByUser.getId());
+//        System.out.println("client_id: " + client_id);
+//        String prompt_object = taskByUser.getGenerate_request();
+//        System.out.println("prompt_object: " + StrUtil.sub(prompt_object, 0, 200));
+//
+////        // [ComfyUI] 执行任务
+////        JSONObject prompt_object = JSONUtil.parseObj(prompt);
+////        Mono<ComfyuiResponse> cfPromptResponseMono = comfyuiService.prompt(client_id, prompt_object);
+////        ComfyuiResponse response = cfPromptResponseMono.block();
+////        response.setClient_id(client_id);
+
+        // 5. 发送消息 (rabbitmq)
+        System.out.println("-- 推送成功: (rabbitmq) task_id: " + taskByUser.getId());
+        rabbitTemplate.convertAndSend(config.EXCHANGE, config.ROUTING_KEY, taskByUser);
+        System.out.println("--------------------------------------------------------");
 
     }
 

src/main/java/com/backendsys/modules/sdk/comfyui/service/impl/ComfyuiText2ImageServiceImpl.java

@@ -335,15 +335,8 @@ public class ComfyuiText2ImageServiceImpl implements ComfyuiText2ImageService {
                 "}"
         ;
 
-
         // [DB] 初始化任务
         Long task_id = comfyuiUtil.initComfyuiTask(client_id, prompt, TaskTypeEnums.TEXT_2_IMAGE.getValue());
-
-        JSONObject prompt_object = JSONUtil.parseObj(prompt);
-        Mono<ComfyuiResponse> cfPromptResponseMono = comfyuiService.prompt(client_id, prompt_object);
-        ComfyuiResponse response = cfPromptResponseMono.block();
-        System.out.println("response = " + response);
-
         return task_id;
 
 //        // [ComfyUI] 执行任务

src/main/java/com/backendsys/modules/sdk/comfyui/utils/ComfyuiUtil.java

@@ -4,7 +4,6 @@ import com.backendsys.modules.common.config.security.utils.SecurityUtil;
 import com.backendsys.modules.sdk.comfyui.dao.ComfyuiTaskDao;
 import com.backendsys.modules.sdk.comfyui.dao.ComfyuiTaskExecuteDao;
 import com.backendsys.modules.sdk.comfyui.entity.ComfyuiTask;
-import com.backendsys.modules.sdk.comfyui.entity.ComfyuiTaskExecute;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -40,17 +39,17 @@ public class ComfyuiUtil {
      * [更新] 执行任务
      * - status: 任务状态 (-1:未开始, 1:进行中, 2:成功, 3:失败)
      */
-    public void executeComfyuiTask(String prompt_id, String response, Integer status) {
+    public void updateComfyuiTask(String prompt_id, String response, Integer status) {
         try {
-            LambdaQueryWrapper<ComfyuiTaskExecute> wrapper = new LambdaQueryWrapper<>();
-            wrapper.eq(ComfyuiTaskExecute::getExecute_prompt_id, prompt_id);
+            LambdaQueryWrapper<ComfyuiTask> wrapper = new LambdaQueryWrapper<>();
+            wrapper.eq(ComfyuiTask::getExecute_prompt_id, prompt_id);
 
-            ComfyuiTaskExecute comfyuiTaskExecute = new ComfyuiTaskExecute();
-            comfyuiTaskExecute.setExecute_status(status);
-            comfyuiTaskExecute.setExecute_prompt_id(prompt_id);
-            comfyuiTaskExecute.setGenerate_response(response);
+            ComfyuiTask comfyuiTask = new ComfyuiTask();
+            comfyuiTask.setExecute_status(status);
+            comfyuiTask.setExecute_prompt_id(prompt_id);
+            comfyuiTask.setGenerate_response(response);
 
-            comfyuiTaskExecuteDao.update(comfyuiTaskExecute, wrapper);
+            comfyuiTaskExecuteDao.update(comfyuiTask, wrapper);
         } catch (Exception e) {
             System.out.println(e.getMessage());
         }

src/main/resources/application-local.yml

@@ -228,5 +228,5 @@ comfyui:
   host: 43.128.1.201
   ports: 8000, 8001, 8002, 8003, 8004, 8005, 8006, 8007
   token: $2b$12$.MR4qGaFetN1FPQzbfyIrehsyjnPJ12xAZhR/l7KZpLkUPQTCG4gy
-  is-save: true
+  is-save: false
   # queue-key: comfyui:queue