Merge branch 'dev/dev' into dev/dev_xp

# Conflicts: # src/main/java/com/ai/da/service/GenerateService.java
2024-06-25 16:59:41 +08:00
parent be001477ce 8c65592394
commit e614e3a07a
13 changed files with 339 additions and 90 deletions
--- a/src/main/java/com/ai/da/common/RabbitMQ/GenerateConsumer.java
+++ b/src/main/java/com/ai/da/common/RabbitMQ/GenerateConsumer.java
@@ -46,6 +46,8 @@ public class GenerateConsumer {
    private String generateResultKey;
    @Value("${redis.key.toProductImageResultKey}")
    private String toProductImageResultKey;
+    @Value("${redis.key.relightResultKey}")
+    private String relightResultKey;

    public void generate(Message msg, Channel channel, String consumerName) {
        log.info("============start listening==========");
@@ -204,73 +206,125 @@ public class GenerateConsumer {

        long end = System.currentTimeMillis();
        log.info("tasks_id : {}, end , message : {}, 执行时长： {} 毫秒", generateResult.get("tasks_id"), generateResult.get("message"), (end - start));
-        log.info("============ProcessGenerateResult End listening==========");
-
+        log.info("============ProcessToProductImageResult End listening==========");
    }

-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer1(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 1");
+    private void processRelightResult(Message msg, Channel channel) {
+        log.info("============processRelightResult listening==========");
+        long start = System.currentTimeMillis();
+
+        Map<String, String> generateResult = JSONObject.parseObject(msg.getBody(), Map.class);
+        log.info("toProductImage response : {}", generateResult);
+
+        try {
+            log.info("tasks_id : {} start ", generateResult.get("tasks_id"));
+            if (generateResult.get("status").equals("SUCCESS")) {
+                String url = generateResult.get("image_url");
+                String taskId = generateResult.get("tasks_id");
+                String category = generateResult.get("category");
+                generateService.processRelightResult(taskId, url, category);
+            } else {
+                // 修改redis中的数据状态为exception
+                String key = relightResultKey + ":" + generateResult.get("tasks_id");
+                redisUtil.addToString(key, new Gson().toJson(new GenerateResultVO(generateResult.get("tasks_id"), null, null, "Fail")), CommonConstant.GENERATE_RESULT_EXPIRE_TIME);
+                // 将异常信息存到exception中
+                HashMap<String, String> exceptionInfo = new HashMap<>();
+                exceptionInfo.put(generateResult.get("tasks_id"), generateResult.get("data"));
+                // 存redis
+                redisUtil.addToMap(exceptionMapKey, exceptionInfo);
+            }
+        } catch (Exception e) {
+            log.error(e.getMessage());
+            try {
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+                // 将消息从redis排队队列中删除,需保证被消费的消息存储到db之后再从redis删除
+                redisUtil.removeFromZSet(consumptionOrderKey, generateResult.get("tasks_id"));
+            } catch (IOException exception) {
+                log.error("手动确认，取消返回队列，不再重新消费");
+            }
+            // 将入参和错误信息存入数据库
+            String exceptionMessage = JSONObject.toJSONString(generateResult) +
+                    " Exception message ： " + e.getMessage();
+            HashMap<String, String> exceptionInfo = new HashMap<>();
+            exceptionInfo.put(String.valueOf(generateResult.get("tasks_id")), exceptionMessage);
+            // 存redis
+            redisUtil.addToMap(exceptionMapKey, exceptionInfo);
+        }
+
+        long end = System.currentTimeMillis();
+        log.info("tasks_id : {}, end , message : {}, 执行时长： {} 毫秒", generateResult.get("tasks_id"), generateResult.get("message"), (end - start));
+        log.info("============ProcessRelightResult End listening==========");
    }

-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer2(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 2");
-    }
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer1(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 1");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer2(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 2");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer3(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 3");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer4(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 4");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer5(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 5");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer6(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 6");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer7(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 7");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer8(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 8");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+//    @RabbitHandler
+//    public void generateConsumer9(Message msg, Channel channel) {
+//        generate(msg, channel, "consumer 9");
+//    }
+//
+//    @RabbitListener(queues = MQConfig.GENERATE_RESULT_QUEUE)
+//    @RabbitHandler
+//    public void getGenerateResult(Message msg, Channel channel) {
+//        processGenerateResult(msg, channel);
+//    }
+//
+//    @RabbitListener(queues = MQConfig.TO_PRODUCT_IMAGE_RESULT_QUEUE)
+//    @RabbitHandler
+//    public void getToProductImageResult(Message msg, Channel channel) {
+//        processToProductImageResult(msg, channel);
+//    }

-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
+    @RabbitListener(queues = MQConfig.RELIGHT_RESULT_QUEUE)
    @RabbitHandler
-    public void generateConsumer3(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 3");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer4(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 4");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer5(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 5");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer6(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 6");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer7(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 7");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer8(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 8");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_QUEUE)
-    @RabbitHandler
-    public void generateConsumer9(Message msg, Channel channel) {
-        generate(msg, channel, "consumer 9");
-    }
-
-    @RabbitListener(queues = MQConfig.GENERATE_RESULT_QUEUE)
-    @RabbitHandler
-    public void getGenerateResult(Message msg, Channel channel) {
-        processGenerateResult(msg, channel);
-    }
-
-    @RabbitListener(queues = MQConfig.TO_PRODUCT_IMAGE_RESULT_QUEUE)
-    @RabbitHandler
-    public void getToProductImageResult(Message msg, Channel channel) {
-        processToProductImageResult(msg, channel);
+    public void getRelightResult(Message msg, Channel channel) {
+        processRelightResult(msg, channel);
    }
 }
--- a/src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java
+++ b/src/main/java/com/ai/da/common/RabbitMQ/MQConfig.java
@@ -27,6 +27,8 @@ public class MQConfig {
    public static final String GENERATE_RESULT_QUEUE = "GenerateImage-dev";

    public static final String TO_PRODUCT_IMAGE_RESULT_QUEUE = "ToProductImage-local";
+
+    public static final String RELIGHT_RESULT_QUEUE = "Relight-local";
    public MQConfig() {
    }

--- a/src/main/java/com/ai/da/common/RabbitMQ/SRConsumer.java
+++ b/src/main/java/com/ai/da/common/RabbitMQ/SRConsumer.java
@@ -218,17 +218,17 @@ public class SRConsumer {
        taskListService.updateTaskStatusOrOutputRedis(uniqueId, "fail", null);
    }

-    @RabbitListener(queues = MQConfig.SR_QUEUE)
-    @RabbitHandler
-    public void SRConsumer1(Message msg, Channel channel) {
-        superResolution(msg, channel, "consumer 1");
-    }
-
-
-    @RabbitListener(queues = MQConfig.SR_RESULT_QUEUE)
-    @RabbitHandler
-    public void SRResultConsumer1(Message msg, Channel channel) {
-        getSRResult(msg, channel, "consumer 1");
-    }
+//    @RabbitListener(queues = MQConfig.SR_QUEUE)
+//    @RabbitHandler
+//    public void SRConsumer1(Message msg, Channel channel) {
+//        superResolution(msg, channel, "consumer 1");
+//    }
+//
+//
+//    @RabbitListener(queues = MQConfig.SR_RESULT_QUEUE)
+//    @RabbitHandler
+//    public void SRResultConsumer1(Message msg, Channel channel) {
+//        getSRResult(msg, channel, "consumer 1");
+//    }

 }