support submit task

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/rocketmq/consumer/ContentUnderstandingPipelineTaskConsumer.java

@@ -34,15 +34,15 @@ import java.util.Map;
 import java.util.Objects;
 
 /**
- * 旧版视频理解 需要延迟处理的视频 消费者
+ * 内容理解任务 消费者
  *
  * @author supeng
  */
 @Slf4j
 @Component
 @RocketMQMessageListener(endpoints = "${rocketmq.consumer.endpoints:rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080}",
-        topic = "${rocketmq.consumer.pipelinetask.topic:topic_content_understanding_pipeline_task}",
-        consumerGroup = "${rocketmq.consumer.pipelinetask.group:group_content_understanding_pipeline_task}", tag = "*", consumptionThreadCount = 8)
+        topic = "${rocketmq.consumer.pipelinetask.normal.topic:topic_content_understanding_pipeline_task}",
+        consumerGroup = "${rocketmq.consumer.pipelinetask.normal.group:group_content_understanding_pipeline_task}", tag = "*", consumptionThreadCount = 8)
 public class ContentUnderstandingPipelineTaskConsumer implements RocketMQListener {
 
     @Autowired

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/rocketmq/consumer/ContentUnderstandingPipelingTaskDelayConsumer.java

@@ -0,0 +1,62 @@
+package com.tzld.piaoquan.content.understanding.rocketmq.consumer;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.TypeReference;
+import com.tzld.piaoquan.content.understanding.service.ContentService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
+import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.apis.message.MessageView;
+import org.apache.rocketmq.client.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * 内容理解任务 消费者 延迟处理
+ *
+ * @author supeng
+ */
+@Slf4j
+@Component
+@RocketMQMessageListener(endpoints = "${rocketmq.consumer.endpoints:rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080}",
+        topic = "${rocketmq.consumer.pipelinetask.delay.topic:topic_content_understanding_pipeline_task_delay}",
+        consumerGroup = "${rocketmq.consumer.pipelinetask.delay.group:group_content_understanding_pipeline_task_delay}", tag = "*", consumptionThreadCount = 2)
+public class ContentUnderstandingPipelingTaskDelayConsumer implements RocketMQListener {
+
+    @Autowired
+    private ContentService contentService;
+
+    @Override
+    public ConsumeResult consume(MessageView messageView) {
+        log.info("Receive message {}", messageView);
+        MessageId messageId = messageView.getMessageId();
+        try {
+            ByteBuffer body = messageView.getBody();
+            Charset charset = StandardCharsets.UTF_8;
+            String messageStr = charset.decode(body).toString();
+            Map<String, Object> messageMap = JSON.parseObject(messageStr, new TypeReference<Map<String, Object>>() {
+            });
+            Object videoIdObj = messageMap.get("videoId");
+            if (Objects.isNull(videoIdObj)) {
+                //错误消息 清理掉
+                return ConsumeResult.SUCCESS;
+            }
+            Long videoId = Long.parseLong(videoIdObj.toString());
+            boolean flag = contentService.remainVideoUnderstandingHandler(videoId);
+            if (flag) {
+                log.info("Message is acknowledged successfully, messageId={}", messageId);
+                return ConsumeResult.SUCCESS;
+            }
+        } catch (Exception e) {
+            log.error("Message is failed to be acknowledged, messageId={}", messageId, e);
+        }
+        return ConsumeResult.FAILURE;
+    }
+}

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/rocketmq/consumer/OldVersionRemainVideoUnderstandingConsumer.java

@@ -27,8 +27,8 @@ import java.util.Objects;
 @Slf4j
 @Component
 @RocketMQMessageListener(endpoints = "${rocketmq.consumer.endpoints:rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080}",
-        topic = "${rocketmq.consumer.oldversion.topic:topic_old_version_remain_video_understanding}",
-        consumerGroup = "${rocketmq.consumer.oldversion.group:group_old_version_remain_video_understanding}", tag = "*", consumptionThreadCount = 1)
+        topic = "${rocketmq.consumer.oldversion.delay.topic:topic_old_version_remain_video_understanding}",
+        consumerGroup = "${rocketmq.consumer.oldversion.delay.group:group_old_version_remain_video_understanding}", tag = "*", consumptionThreadCount = 1)
 public class OldVersionRemainVideoUnderstandingConsumer implements RocketMQListener {
 
     @Autowired

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/rocketmq/producer/ContentUnderstandingPipelineTaskDelayProducer.java

@@ -0,0 +1,38 @@
+package com.tzld.piaoquan.content.understanding.rocketmq.producer;
+
+import com.alibaba.fastjson.JSON;
+import com.tzld.piaoquan.content.understanding.model.dto.VideoUnderstandingMessageDTO;
+import com.tzld.piaoquan.content.understanding.util.RetryUtil;
+import org.apache.rocketmq.client.apis.message.MessageId;
+import org.apache.rocketmq.client.apis.producer.SendReceipt;
+import org.apache.rocketmq.client.core.RocketMQClientTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * 内容理解任务 任务 生产者 延迟处理
+ *
+ * @author supeng
+ */
+@Component
+public class ContentUnderstandingPipelineTaskDelayProducer {
+    @Autowired
+    private RocketMQClientTemplate rocketMQClientTemplate;
+
+    @Value("${rocketmq.producer.pipelinetask.delay.topic:topic_content_understanding_pipeline_task_delay}")
+    private String pipelineTaskTopicDelay;
+
+    public MessageId syncSendDelayMessage(VideoUnderstandingMessageDTO messageDTO, Duration duration) {
+        return RetryUtil.executeWithRetry(() -> {
+            SendReceipt sendReceipt = rocketMQClientTemplate.syncSendDelayMessage(pipelineTaskTopicDelay, JSON.toJSONString(messageDTO), duration);
+            if (Objects.isNull(sendReceipt) || Objects.isNull(sendReceipt.getMessageId())) {
+                throw new RuntimeException("OldVersionRemainVideoUnderstandingProducer syncSendDelayMessage failure messageDTO = " + messageDTO);
+            }
+            return sendReceipt.getMessageId();
+        }, "OldVersionRemainVideoUnderstandingProducer");
+    }
+}

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/rocketmq/producer/ContentUnderstandingPipelineTaskProducer.java

@@ -24,7 +24,7 @@ public class ContentUnderstandingPipelineTaskProducer {
     @Autowired
     private RocketMQClientTemplate rocketMQClientTemplate;
 
-    @Value("${rocketmq.producer.pipelinetask.topic:topic_content_understanding_pipeline_task}")
+    @Value("${rocketmq.producer.pipelinetask.normal.topic:topic_content_understanding_pipeline_task}")
     private String pipelineTaskTopic;
 
     public MessageId syncSendNormalMessage(PipelineTaskMessageDTO messageDTO) {

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/rocketmq/producer/OldVersionRemainVideoUnderstandingProducer.java

@@ -24,7 +24,7 @@ public class OldVersionRemainVideoUnderstandingProducer {
     @Autowired
     private RocketMQClientTemplate rocketMQClientTemplate;
 
-    @Value("${rocketmq.producer.oldversion.topic:topic_old_version_remain_video_understanding}")
+    @Value("${rocketmq.producer.oldversion.delay.topic:topic_old_version_remain_video_understanding}")
     private String oldVersionRemainTopic;
 
     public MessageId syncSendDelayMessage(VideoUnderstandingMessageDTO messageDTO, Duration duration) {

content-understanding-server/src/main/resources/application-dev.yml

@@ -80,14 +80,23 @@ rocketmq:
   consumer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_test
-      group: group_old_version_remain_video_understanding_test
+      delay:
+        topic: topic_old_version_remain_video_understanding_test
+        group: group_old_version_remain_video_understanding_test
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_test
-      group: group_content_understanding_pipeline_task_test
+      normal:
+        topic: topic_content_understanding_pipeline_task_test
+        group: group_content_understanding_pipeline_task_test
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_test
+        group: group_content_understanding_pipeline_task_delay_test
   producer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_test
+      delay:
+        topic: topic_old_version_remain_video_understanding_test
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_test
+      normal:
+        topic: topic_content_understanding_pipeline_task_test
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_test

content-understanding-server/src/main/resources/application-pre.yml

@@ -80,14 +80,23 @@ rocketmq:
   consumer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_pre
-      group: group_old_version_remain_video_understanding_pre
+      delay:
+        topic: topic_old_version_remain_video_understanding_pre
+        group: group_old_version_remain_video_understanding_pre
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_pre
-      group: group_content_understanding_pipeline_task_pre
+      normal:
+        topic: topic_content_understanding_pipeline_task_pre
+        group: group_content_understanding_pipeline_task_pre
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_pre
+        group: group_content_understanding_pipeline_task_delay_pre
   producer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_pre
+      delay:
+        topic: topic_old_version_remain_video_understanding_pre
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_pre
+      normal:
+        topic: topic_content_understanding_pipeline_task_pre
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_pre

content-understanding-server/src/main/resources/application-prod.yml

@@ -80,14 +80,23 @@ rocketmq:
   consumer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding
-      group: group_old_version_remain_video_understanding
+      delay:
+        topic: topic_old_version_remain_video_understanding
+        group: group_old_version_remain_video_understanding
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task
-      group: group_content_understanding_pipeline_task
+      normal:
+        topic: topic_content_understanding_pipeline_task
+        group: group_content_understanding_pipeline_task
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay
+        group: group_content_understanding_pipeline_task_delay
   producer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding
+      delay:
+        topic: topic_old_version_remain_video_understanding
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task
+      normal:
+        topic: topic_content_understanding_pipeline_task
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay

content-understanding-server/src/main/resources/application-stress.yml

@@ -82,14 +82,23 @@ rocketmq:
   consumer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_test
-      group: group_old_version_remain_video_understanding_test
+      delay:
+        topic: topic_old_version_remain_video_understanding_test
+        group: group_old_version_remain_video_understanding_test
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_test
-      group: group_content_understanding_pipeline_task_test
+      normal:
+        topic: topic_content_understanding_pipeline_task_test
+        group: group_content_understanding_pipeline_task_test
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_test
+        group: group_content_understanding_pipeline_task_delay_test
   producer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_test
+      delay:
+        topic: topic_old_version_remain_video_understanding_test
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_test
+      normal:
+        topic: topic_content_understanding_pipeline_task_test
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_test

content-understanding-server/src/main/resources/application-test.yml

@@ -82,14 +82,23 @@ rocketmq:
   consumer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_test
-      group: group_old_version_remain_video_understanding_test
+      delay:
+        topic: topic_old_version_remain_video_understanding_test
+        group: group_old_version_remain_video_understanding_test
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_test
-      group: group_content_understanding_pipeline_task_test
+      normal:
+        topic: topic_content_understanding_pipeline_task_test
+        group: group_content_understanding_pipeline_task_test
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_test
+        group: group_content_understanding_pipeline_task_delay_test
   producer:
     endpoints: rmq-cn-vym47zv2i03-vpc.cn-hangzhou.rmq.aliyuncs.com:8080
     oldversion:
-      topic: topic_old_version_remain_video_understanding_test
+      delay:
+        topic: topic_old_version_remain_video_understanding_test
     pipelinetask:
-      topic: topic_content_understanding_pipeline_task_test
+      normal:
+        topic: topic_content_understanding_pipeline_task_test
+      delay:
+        topic: topic_content_understanding_pipeline_task_delay_test