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 5f316f4..b678140 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,11 @@ public class TestController { @Autowired private RocketMQTemplate rocketMQTemplate; + @Autowired + private KafkaTemplate kafkaTemplate; // 注入 KafkaTemplate + + private final ObjectMapper objectMapper = new ObjectMapper(); + @PostMapping("/mockMq") @ApiOperation(value = "补偿处理消息") @@ -32,5 +43,21 @@ public class TestController { 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 消息已发送"); + } }