增加mq处理dcc

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java

@@ -1,13 +1,19 @@
 package com.volvo.ai.analytic.center.controller;
 
 
+import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
 import com.volvo.common.core.util.ResultMsg;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -24,6 +30,15 @@ public class TestController {
     @Autowired
     private RocketMQTemplate rocketMQTemplate;
 
+    @Autowired
+    private KafkaTemplate<String, String> kafkaTemplate; // 注入 KafkaTemplate
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    @Value("spring.kafka.producer.topic")
+    private  String kafkaTopic; // 注入 KafkaTemplate
+    @Value("")
+    private  String kafkaTopicGroup;
 
     @PostMapping("/mockMq")
     @ApiOperation(value = "补偿处理消息")
@@ -31,5 +46,22 @@ public class TestController {
         rocketMQTemplate.syncSend("COMMUNITY_POST_TO_AI_TOPIC", message);
         return ResultMsg.ok("ok");
     }
+
+    @PostMapping("/mockKafka")
+    @ApiOperation(value = "生成 Kafka 数据")
+    public ResultMsg<Object> mockKafka(@RequestBody String message) {
+        try {
+            AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
+            log.info("mockKafka：{}",message);
+            for (int i = 0; i < 20; i++){
+                aicorpusTelephone.setSourceId(aicorpusTelephone.getSourceId().concat("_"+i));
+                kafkaTemplate.send("topic_voc_covert_text_log", JSONObject.toJSONString(aicorpusTelephone)); // 发送 Kafka 消息
+            }
+            log.info("Kafka 消息已发送: {}", message);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+        return ResultMsg.ok("Kafka 消息已发送");
+    }
 }
 

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java

@@ -0,0 +1,61 @@
+package com.volvo.ai.analytic.center.job;
+
+import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
+import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
+import com.volvo.ai.analytic.center.service.TmOdsVdqwMessagearchivingService;
+import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
+import com.volvo.common.core.util.ResultMsg;
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Arrays;
+import java.util.List;
+
+
+@Slf4j
+@Component
+@RestController
+public class DccCorpusJob {
+
+    @Autowired
+    private TmOdsVdqwMessagearchivingService tmOdsVdqwMessagearchivingService;
+
+    @Autowired
+    private TmTelephoneCorpusService tmTelephoneCorpusService;
+
+    @Autowired
+    private  TmTelephoneCorpusMapper tmTelephoneCorpusMapper;
+
+    /**
+     * dcc语料处理
+     */
+    @XxlJob("dccCorpusJob")
+    public ResultMsg dccCorpusJob(@RequestBody String paramJson) {
+        try {
+            // 获取任务参数
+            String param = XxlJobHelper.getJobParam();
+            if(StringUtils.isEmpty(param)){
+                param = paramJson;
+            }
+
+            // 分页查询 过滤已跑批并发送的
+
+
+            tmOdsVdqwMessagearchivingService.runQiWeiCorpusDify(param);
+        } catch (Exception e) {
+            log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
+            throw new RuntimeException(e);
+        }
+        return ResultMsg.ok();
+    }
+
+
+
+}

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqProducer.java

@@ -0,0 +1,74 @@
+
+package com.volvo.ai.analytic.center.mq;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.volvo.ai.analytic.center.constant.Constant;
+import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
+import com.volvo.ai.analytic.center.dto.corpus.DisplayDTO;
+import com.volvo.ai.analytic.center.entity.TmTelephoneCorpus;
+import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Component;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * @ClassName CorpusProcessKafkaConsumer
+ * @Description
+ * @Author renzhen
+ * @Date 2025-03-04 10:18
+ * @Version 1.0
+ **/
+
+@Slf4j
+@Component
+@RestController
+@RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.corpus.dcctopic}",
+		topic = "${rocketmq.consumer.corpus.dcctopicgroup}",
+		enableMsgTrace = true)
+public class CorpusDccMqProducer  implements RocketMQListener<MessageExt>  {
+
+	@Autowired
+	private TmTelephoneCorpusService tmTelephoneCorpusService;
+
+	private final ObjectMapper objectMapper = new ObjectMapper();
+
+	@Override
+	public void onMessage(MessageExt messageExt) {
+		long startTime = System.currentTimeMillis();
+		try {
+			String message = new String(messageExt.getBody());
+			log.info("dcc_mq message: " + message);
+			AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
+			DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);
+			log.info("corpusDccMqProducer display getSegments: {}", display.getSegments());
+			tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone);
+			log.info("corpusDccMqProducer mq 处理完成: {}", aicorpusTelephone.getSourceId());
+			log.info("dcc_mq 处理完成，耗时：{}", System.currentTimeMillis() - startTime);
+		} catch (JsonProcessingException e) {
+			log.info(" dcc mq 处理失败：{}", e.getMessage());
+		}
+	}
+
+}
+

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java

@@ -9,13 +9,20 @@ import com.volvo.ai.analytic.center.entity.TmTelephoneCorpus;
 import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+import javax.annotation.Resource;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -33,6 +40,7 @@ import java.util.concurrent.Executors;
 @Slf4j
 @Component
 @RestController
+@RefreshScope
 public class CorpusProcessKafkaProducer {
 
 	@Autowired
@@ -40,6 +48,12 @@ public class CorpusProcessKafkaProducer {
 
 	private final ObjectMapper objectMapper = new ObjectMapper();
 
+	@Value("${rocketmq.producer.corpus.dcctopic}")
+	private String dccMqTipic;
+
+	@Resource
+	private RocketMQTemplate rocketMqTemplate;
+
 	@PostMapping("corpusProcessKafkaConsumer")
 	@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
 	public void listen(List<String> recordMessages) {
@@ -71,11 +85,24 @@ public class CorpusProcessKafkaProducer {
 								log.info(" dcc 语料开始处理");
 								tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus);
 								long startTimeDify = System.currentTimeMillis();
-								CompletableFuture.runAsync(() -> {
+
-									tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone);
+								rocketMqTemplate.asyncSend(dccMqTipic, MessageBuilder.withPayload(message).build(),
-								}, runTelephoneExecutor);
+										new SendCallback() {
-								// 关闭线程池
+											@Override
-								runTelephoneExecutor.shutdown();
+											public void onSuccess(SendResult sendResult) {
+												log.info("dcc发送MQ成功  消息体:{}", message);
+											}
+											@Override
+											public void onException(Throwable e) {
+												log.error("dcc 送MQ异常  消息体:{}, 异常：", message, e);
+											}
+										}, 10000);
+
+//								CompletableFuture.runAsync(() -> {
+//									tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone);
+//								}, runTelephoneExecutor);
+//								// 关闭线程池
+//								runTelephoneExecutor.shutdown();
 								log.info(" dify处理耗时：{}", System.currentTimeMillis() - startTimeDify);
 							}
 						} catch (Exception e) {