From 40e38bda603a1397232026f8d303b1f3ab8260a9 Mon Sep 17 00:00:00 2001 From: zren25 Date: Mon, 31 Mar 2025 17:28:35 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0mq=E5=A4=84=E7=90=86dcc?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../center/controller/TestController.java | 32 ++++++++ .../ai/analytic/center/job/DccCorpusJob.java | 61 +++++++++++++++ .../center/mq/CorpusDccMqProducer.java | 74 +++++++++++++++++++ .../center/mq/CorpusProcessKafkaProducer.java | 37 ++++++++-- 4 files changed, 199 insertions(+), 5 deletions(-) create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqProducer.java diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java index ff6857d..48071ea 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java +++ b/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 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 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 消息已发送"); + } } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java new file mode 100644 index 0000000..56a0eff --- /dev/null +++ b/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(); + } + + + +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusDccMqProducer.java new file mode 100644 index 0000000..04b46c6 --- /dev/null +++ b/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 { + + @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()); + } + } + +} + diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java index 4d01458..561d963 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java +++ b/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 recordMessages) { @@ -71,11 +85,24 @@ public class CorpusProcessKafkaProducer { log.info(" dcc 语料开始处理"); tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus); long startTimeDify = System.currentTimeMillis(); - CompletableFuture.runAsync(() -> { - tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone); - }, runTelephoneExecutor); - // 关闭线程池 - runTelephoneExecutor.shutdown(); + + rocketMqTemplate.asyncSend(dccMqTipic, MessageBuilder.withPayload(message).build(), + new SendCallback() { + @Override + 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) {