feat: 定时任务消费评估任务

cmvr-iot-common/src/main/java/com/cmvr/common/utils/http/CallAPIUtil.java

@@ -84,6 +84,20 @@ public class CallAPIUtil {
         }
     }
 
+    public static HttpResponse doPost(String url, Map<String, String> headers, Object jsonObj) {
+        HttpRequest request = HttpUtil.createPost(url)
+                .header("Content-Type", ContentType.JSON.toString())
+                .body(JSONUtil.toJsonStr(jsonObj));
+
+        if (headers != null) {
+            request.addHeaders(headers);
+        }
+
+        try (HttpResponse response = request.execute()) {
+            return response;
+        }
+    }
+
     /**
      * 发送 POST 请求 - 文件上传 (multipart/form-data)
      *

cmvr-iot-evaluation/src/main/java/com/cmvr/evaluation/service/impl/AeCallbackServiceImpl.java

@@ -12,9 +12,9 @@ import com.cmvr.evaluation.service.ExViProjectService;
 import com.cmvr.evaluation.service.IAeEvaluationService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
+import java.time.LocalDateTime;
 import java.util.Objects;
 
 
@@ -25,7 +25,6 @@ public class AeCallbackServiceImpl implements AeCallbackService {
 
     private final IAeEvaluationService aeEvaluationService;
     private final ExViProjectService exViProjectService;
-    private final StringRedisTemplate redisTemplate;
 
     @Override
     public void callback(AeEvaluation aeEvaluation) {
@@ -43,36 +42,25 @@ public class AeCallbackServiceImpl implements AeCallbackService {
         }
         JSONObject jsonObject = JSON.parseObject(result);
         String overallResult = jsonObject.getString("overall_result");
+        log.info("时间：{} id:{} overall_result:{}", LocalDateTime.now(), aeId, overallResult);
         int status;
         if (StrUtil.isNotEmpty(overallResult) && overallResult.equals("成功")) {
-            status = 3;
+            status = 2;
         } else {
-            status = 4;
+            status = 3;
         }
         LambdaUpdateWrapper<AeEvaluation> wrapper = Wrappers.lambdaUpdate();
         wrapper.eq(AeEvaluation::getAeId, aeId)
                 .set(AeEvaluation::getResult, result)
                 .set(AeEvaluation::getStatus, status); // 评估完成/失败
 
-        aeEvaluationService.update(wrapper);
+        boolean update = aeEvaluationService.update(wrapper);
+        log.info("时间:{} id:{} update:{}", LocalDateTime.now(), aeId, update);
 
         // 判断是否是最后一次评估 更改项目状态
         boolean finished = aeEvaluationService.isAllEvaluationsFinished(byId.getInstId());
         if (finished) {
             exViProjectService.updateStatus(byId.getTaskId());
         }
-
-        // 继续下一个评估
-        String str = redisTemplate.opsForList().leftPop("ae:evaluation:queue");
-        if (StrUtil.isEmpty(str)) {
-            return;
-        }
-        AeEvaluation evaluation = JSON.parseObject(str, AeEvaluation.class);
-        try {
-            Thread.sleep(2000);
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-        aeEvaluationService.runEvaluation(evaluation);
     }
 }

cmvr-iot-evaluation/src/main/java/com/cmvr/evaluation/service/impl/AeEvaluationServiceImpl.java

@@ -1,13 +1,14 @@
 package com.cmvr.evaluation.service.impl;
 
 import cn.hutool.core.bean.BeanUtil;
-import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.StrUtil;
+import cn.hutool.http.HttpResponse;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.cmvr.common.exception.GlobalException;
@@ -184,15 +185,11 @@ public class AeEvaluationServiceImpl extends ServiceImpl<AeEvaluationMapper, AeE
                 // 入库
                 this.saveBatch(evaluations);
 
-                // 提取第一个对象出来
+                // 提交评估
-                AeEvaluation first = CollUtil.getFirst(evaluations);
+                for (AeEvaluation e : evaluations) {
-                List<AeEvaluation> rest = CollUtil.sub(evaluations, 1, evaluations.size());
+                    redisTemplate.opsForList()
-                // 剩下的先入队列
+                            .rightPush("ae:evaluation:queue", JSON.toJSONString(e));
-                for (AeEvaluation e : rest) {
-                    redisTemplate.opsForList().rightPush("ae:evaluation:queue", JSON.toJSONString(e));
                 }
-
-                runEvaluation(first);
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
@@ -205,6 +202,12 @@ public class AeEvaluationServiceImpl extends ServiceImpl<AeEvaluationMapper, AeE
     }
 
     public void runEvaluation(AeEvaluation evaluation) {
+        LambdaUpdateWrapper<AeEvaluation> wrapper = Wrappers.lambdaUpdate();
+        wrapper.eq(AeEvaluation::getAeId, evaluation.getAeId())
+                .eq(AeEvaluation::getStatus, 0)
+                .set(AeEvaluation::getStatus, 1);
+
+        this.update(wrapper);
         // 构建评估请求
         Map<String, String> body = new HashMap<>();
         body.put("aeId", evaluation.getAeId());
@@ -212,10 +215,17 @@ public class AeEvaluationServiceImpl extends ServiceImpl<AeEvaluationMapper, AeE
         body.put("videoPath", evaluation.getVideoPath());
         body.put("content", evaluation.getAvContent());
         body.put("aeType", evaluation.getAeType());
-        CallAPIUtil.doPostJson(apiProperties.getEvaluation(), null, body);
+        HttpResponse httpResponse = CallAPIUtil.doPost(apiProperties.getEvaluation(), null, body);
+        if (httpResponse.getStatus() != 200) {
+            String responseBody = httpResponse.body();
+            LambdaUpdateWrapper<AeEvaluation> wrapper1 = Wrappers.lambdaUpdate();
+            wrapper1.eq(AeEvaluation::getAeId, evaluation.getAeId())
+                    .eq(AeEvaluation::getStatus, 1)
+                    .set(AeEvaluation::getStatus, 4)
+                    .set(AeEvaluation::getResult, responseBody);
 
-        evaluation.setStatus(1);
+            this.update(wrapper1);
-        this.updateById(evaluation);
+        }
     }
 
 
@@ -237,13 +247,13 @@ public class AeEvaluationServiceImpl extends ServiceImpl<AeEvaluationMapper, AeE
             return false;
         }
 
-        long unfinished = this.count(
+        long finished = this.count(
                 Wrappers.lambdaQuery(AeEvaluation.class)
                         .eq(AeEvaluation::getInstId, instId)
-                        .ne(AeEvaluation::getStatus, 2)
+                        .ge(AeEvaluation::getStatus, 2)
         );
 
-        return unfinished == 0;
+        return finished == total;
     }
 
     private Map<String, List<TeNodeInst>> buildSingleIterationMap(List<TeNodeInst> view) {

cmvr-iot-quartz/pom.xml

@@ -35,6 +35,12 @@
             <artifactId>cmvr-iot-common</artifactId>
         </dependency>
 
+        <!-- 评估-->
+        <dependency>
+            <groupId>com.cmvr</groupId>
+            <artifactId>cmvr-iot-evaluation</artifactId>
+        </dependency>
+
     </dependencies>
 
 </project>

cmvr-iot-quartz/src/main/java/com/cmvr/quartz/task/AeEvaluationTask.java

@@ -0,0 +1,40 @@
+package com.cmvr.quartz.task;
+
+import com.alibaba.fastjson2.JSON;
+import com.cmvr.evaluation.model.domain.AeEvaluation;
+import com.cmvr.evaluation.service.IAeEvaluationService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * 评估任务调度
+ */
+@Slf4j
+@Component("evaluation")
+@RequiredArgsConstructor
+public class AeEvaluationTask {
+
+    private final StringRedisTemplate redisTemplate;
+
+    private final IAeEvaluationService aeEvaluationService;
+
+    public void schedule() {
+        String json = redisTemplate.opsForList().leftPop("ae:evaluation:queue");
+
+        if (json == null) {
+            return;
+        }
+
+        AeEvaluation evaluation = JSON.parseObject(json, AeEvaluation.class);
+
+        try {
+            aeEvaluationService.runEvaluation(evaluation);
+        } catch (Exception e) {
+            log.error("评估执行失败 aeId={}", evaluation.getAeId(), e);
+            // 失败重回队列
+            redisTemplate.opsForList().rightPush("ae:evaluation:queue", json);
+        }
+    }
+}