mock数据kafak

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<String, String> 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<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 消息已发送");
+    }
 }