update

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/model/dto/ContentAnalyseDTO.java

@@ -13,4 +13,8 @@ public class ContentAnalyseDTO {
      *
      */
     private Long videoId;
+    /**
+     * 任务ID
+     */
+    private String taskId;
 }

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/impl/ContentServiceImpl.java

@@ -7,13 +7,12 @@ import com.tzld.piaoquan.content.understanding.common.enums.ExceptionEnum;
 import com.tzld.piaoquan.content.understanding.common.exception.CommonException;
 import com.tzld.piaoquan.content.understanding.dao.mapper.CuPipelineMapper;
 import com.tzld.piaoquan.content.understanding.dao.mapper.CuTaskMapper;
+import com.tzld.piaoquan.content.understanding.dao.mapper.CuVideoUnderstandingResultMapper;
 import com.tzld.piaoquan.content.understanding.model.dto.ContentDTO;
+import com.tzld.piaoquan.content.understanding.model.dto.ContentUnderstandingResultDTO;
 import com.tzld.piaoquan.content.understanding.model.dto.TaskResultDTO;
 import com.tzld.piaoquan.content.understanding.model.param.*;
-import com.tzld.piaoquan.content.understanding.model.po.CuPipeline;
-import com.tzld.piaoquan.content.understanding.model.po.CuPipelineExample;
-import com.tzld.piaoquan.content.understanding.model.po.CuTask;
-import com.tzld.piaoquan.content.understanding.model.po.CuTaskExample;
+import com.tzld.piaoquan.content.understanding.model.po.*;
 import com.tzld.piaoquan.content.understanding.model.vo.ContentUnderstandingResultVO;
 import com.tzld.piaoquan.content.understanding.model.vo.TaskResultVO;
 import com.tzld.piaoquan.content.understanding.service.ContentService;
@@ -49,6 +48,10 @@ public class ContentServiceImpl implements ContentService {
     private CuPipelineMapper cuPipelineMapper;
     @Autowired
     private CuTaskMapper cuTaskMapper;
+    @Autowired
+    private CuVideoUnderstandingResultMapper cuVideoUnderstandingResultMapper;
+
+    private static final Integer MAX_SIZE = 20;
 
     @Override
     public void analyse(ContentAnalyseParam param) {
@@ -147,6 +150,30 @@ public class ContentServiceImpl implements ContentService {
 
     @Override
     public ContentUnderstandingResultVO getVideoUnderstandingResult(GetVideoUnderstandingResultParam param) {
-        return null;
+        if (Objects.isNull(param) || Objects.isNull(param.getContents()) || param.getContents().isEmpty()) {
+            throw new CommonException(ExceptionEnum.PARAMS_INVALID, "参数不能为空");
+        }
+        if (param.getContents().size() > MAX_SIZE) {
+            throw new CommonException(ExceptionEnum.PARAMS_INVALID, "查询结果数不能超过" + MAX_SIZE + "条");
+        }
+        ContentUnderstandingResultVO contentUnderstandingResultVO = new ContentUnderstandingResultVO();
+        List<ContentUnderstandingResultDTO> contentUnderstandingResultDTOS = new ArrayList<>();
+        for (ContentDTO contentDTO : param.getContents()) {
+            if (Objects.isNull(contentDTO) || Objects.isNull(contentDTO.getPipelineId())
+                    || Objects.isNull(contentDTO.getContent()) || Objects.isNull(contentDTO.getVideoId())) {
+                continue;
+            }
+            CuVideoUnderstandingResultExample example = new CuVideoUnderstandingResultExample();
+            example.createCriteria().andVideoIdEqualTo(contentDTO.getVideoId()).andPipelineIdEqualTo(contentDTO.getPipelineId());
+            List<CuVideoUnderstandingResult> list = cuVideoUnderstandingResultMapper.selectByExample(example);
+            if (Objects.isNull(list) || list.isEmpty()) {
+                continue;
+            }
+            ContentUnderstandingResultDTO contentUnderstandingResultDTO = new ContentUnderstandingResultDTO();
+            BeanUtils.copyProperties(contentDTO, contentUnderstandingResultDTO);
+            contentUnderstandingResultDTOS.add(contentUnderstandingResultDTO);
+        }
+        contentUnderstandingResultVO.setResults(contentUnderstandingResultDTOS);
+        return contentUnderstandingResultVO;
     }
 }

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/impl/PipelineServiceImpl.java

@@ -2,7 +2,9 @@ package com.tzld.piaoquan.content.understanding.service.impl;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.TypeReference;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.tzld.piaoquan.content.understanding.common.enums.ExceptionEnum;
+import com.tzld.piaoquan.content.understanding.common.enums.TaskStatusEnum;
 import com.tzld.piaoquan.content.understanding.common.exception.CommonException;
 import com.tzld.piaoquan.content.understanding.dao.mapper.CuPipelineMapper;
 import com.tzld.piaoquan.content.understanding.dao.mapper.CuPipelineStepMapper;
@@ -13,10 +15,7 @@ import com.tzld.piaoquan.content.understanding.model.dto.ContentDTO;
 import com.tzld.piaoquan.content.understanding.model.dto.TreeNode;
 import com.tzld.piaoquan.content.understanding.model.param.ActionParam;
 import com.tzld.piaoquan.content.understanding.model.param.ContentAnalyseParam;
-import com.tzld.piaoquan.content.understanding.model.po.CuPipelineStep;
-import com.tzld.piaoquan.content.understanding.model.po.CuPipelineStepExample;
-import com.tzld.piaoquan.content.understanding.model.po.CuPrompt;
-import com.tzld.piaoquan.content.understanding.model.po.CuTask;
+import com.tzld.piaoquan.content.understanding.model.po.*;
 import com.tzld.piaoquan.content.understanding.service.Action;
 import com.tzld.piaoquan.content.understanding.service.PipelineService;
 import lombok.extern.slf4j.Slf4j;
@@ -25,7 +24,10 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import java.util.*;
+import java.util.concurrent.*;
 
 /**
  * @author supeng
@@ -54,6 +56,32 @@ public class PipelineServiceImpl implements PipelineService {
     @Autowired
     private CuTaskMapper cuTaskMapper;
 
+    /**
+     * 线程池队列大小
+     */
+    private static final int QUEUE_MAX_SIZE = 100000;
+    /**
+     * 线程命名
+     */
+    private static final ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder().setNameFormat("pipeline-service-pool-%d").build();
+    /**
+     * 线程池
+     */
+    private static ExecutorService pool;
+
+    @PostConstruct
+    public void init() {
+        //init thread pool
+        pool = new ThreadPoolExecutor(32, 32,
+                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(QUEUE_MAX_SIZE), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.AbortPolicy());
+    }
+
+    @PreDestroy
+    public void destroy() {
+        //gracefully shutdown
+        pool.shutdown();
+    }
+
     @Override
     public void execute(ContentAnalyseParam param) {
         // 1.获取pipeline配置
@@ -100,18 +128,22 @@ public class PipelineServiceImpl implements PipelineService {
         if (Objects.isNull(stepList) || stepList.isEmpty()) {
             throw new CommonException(ExceptionEnum.CONFIG_ERROR, "无配置 pipelineId：" + pipelineId);
         }
-        // 3.构建tree结构
-        TreeNode root = buildTree(stepList);
-        root.setType(contentType);
         if (Objects.isNull(content)) {
             throw new CommonException(ExceptionEnum.DATA_ERROR, "数据异常content：" + content);
         }
-        root.setInput(content);
-        ContentAnalyseDTO dto1 = new ContentAnalyseDTO();
-        dto1.setVideoId(contentDTO.getVideoId());
-        dto1.setContent(content);
-        // 4.按照步骤执行pipeline每一步动作
-        executeTaskTreeNodeBFS(root, dto1);
+        //异步执行
+        pool.execute(() -> {
+            // 3.构建tree结构
+            TreeNode root = buildTree(stepList);
+            root.setType(contentType);
+            root.setInput(content);
+            ContentAnalyseDTO dto1 = new ContentAnalyseDTO();
+            dto1.setTaskId(taskId);
+            dto1.setVideoId(contentDTO.getVideoId());
+            dto1.setContent(content);
+            // 4.按照步骤执行pipeline每一步动作
+            executeTaskTreeNodeBFS(root, dto1);
+        });
         return taskId;
     }
 
@@ -201,6 +233,7 @@ public class PipelineServiceImpl implements PipelineService {
         Queue<TreeNode> queue = new LinkedList<>();
         queue.offer(root);
 
+        String taskId = dto.getTaskId();
         Long videoId = dto.getVideoId();
 
         while (!queue.isEmpty()) {
@@ -220,7 +253,15 @@ public class PipelineServiceImpl implements PipelineService {
                         //保存数据到mongo
                         mongoTemplate.insert(resultMap, COLLECTION_NAME);
                         //保存数据到mysql
-
+                        CuTaskExample example = new CuTaskExample();
+                        example.createCriteria().andTaskIdEqualTo(taskId);
+                        CuTask cuTask = new CuTask();
+                        cuTask.setOutput(JSON.toJSONString(resultMap));
+                        cuTask.setTaskStatus(TaskStatusEnum.SUCCESS.getValue());
+                        int insert = cuTaskMapper.updateByExampleSelective(cuTask, example);
+                        if (insert <= 0) {
+                            log.error("step execute error step = {}, result = {} insert = {}", JSON.toJSONString(step), result, insert);
+                        }
                     } catch (Exception e) {
                         log.error("step execute error step = {}, result = {}", e, JSON.toJSONString(step), result);
                     }