From 79acc23dde7eb48d959323d60bdabbd3b603a4e6 Mon Sep 17 00:00:00 2001 From: zren25 Date: Mon, 31 Mar 2025 13:02:02 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9kafka=E6=B6=88=E8=B4=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../center/controller/TestController.java | 43 +++++++++---- .../center/mq/CorpusProcessKafkaProducer.java | 64 +++++++++++++------ 2 files changed, 76 insertions(+), 31 deletions(-) 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 902fdf7..2fb4f97 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,16 +1,10 @@ package com.volvo.ai.analytic.center.controller; -import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; -import com.volvo.ai.analytic.center.dto.req.CallItem; -import com.volvo.ai.analytic.center.dto.req.DiFyReq; -import com.volvo.ai.analytic.center.dto.req.DiffDefeatanAlysis; -import com.volvo.ai.analytic.center.dto.req.DifyImageWorkFlow; -import com.volvo.ai.analytic.center.dto.resp.DiffDefeatAnalyseOutputResult; -import com.volvo.ai.analytic.center.feign.DiFyFeign; -import com.volvo.ai.analytic.center.service.DiFyService; -import com.volvo.ai.analytic.center.service.MqMessageRecordService; +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; @@ -18,26 +12,33 @@ 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; import org.springframework.web.bind.annotation.RestController; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - @RestController @Api(tags = "测试类API") @RequestMapping("/ai-analytic-center") @Slf4j +@RefreshScope 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 = "补偿处理消息") @@ -45,5 +46,21 @@ 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); + for (int i = 0; i < 100000; i++){ + aicorpusTelephone.setSourceId(aicorpusTelephone.getSourceId().concat("_"+i)); + kafkaTemplate.send(kafkaTopic, 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/mq/CorpusProcessKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java index 3738347..e02a9e4 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 @@ -8,6 +8,8 @@ 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.kafka.clients.consumer.ConsumerRecord; import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.annotation.KafkaListener; @@ -16,7 +18,10 @@ import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RestController; 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 @@ -38,26 +43,49 @@ public class CorpusProcessKafkaProducer { @PostMapping("corpusProcessKafkaConsumer") @KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}") - public void listen(String message) { + public void listen(List> recordMessage) { try { - log.info("CorpusProcessKafkaProducer Received message: {}" , message); + log.info("CorpusProcessKafkaProducer Received message: {}", recordMessage); + // 获取消息列表 + int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2; + log.info("获取的线程数：{}", optimalThreadPoolSize); + ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); + if(CollectionUtils.isNotEmpty(recordMessage)){ + log.info("CorpusProcessKafkaProducer List size: {}", recordMessage.size()); + for (ConsumerRecord record : recordMessage) { + executor.submit(() -> { + try { + String message = (String) record.value(); + AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class); + log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay()); + DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class); + log.info("aicorpusTelephone display getSegments: {}", display.getSegments()); + TmTelephoneCorpus tmTelephoneCorpus = new TmTelephoneCorpus(); + BeanUtils.copyProperties(aicorpusTelephone, tmTelephoneCorpus); + tmTelephoneCorpus.setCreateBy("kafka"); + tmTelephoneCorpus.setCreateTime(LocalDateTime.now()); + ExecutorService runTelephoneExecutor = Executors.newFixedThreadPool(optimalThreadPoolSize); - AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class); - log.info("aicorpusTelephone categoryCode:{}, display: {}" ,aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay()); - DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class); - log.info("aicorpusTelephone display getSegments: {}" , display.getSegments()); - TmTelephoneCorpus tmTelephoneCorpus = new TmTelephoneCorpus(); - BeanUtils.copyProperties(aicorpusTelephone, tmTelephoneCorpus); - tmTelephoneCorpus.setCreateBy("kafka"); - tmTelephoneCorpus.setCreateTime(LocalDateTime.now()); + // 条件：只处理dcc的 10s通话时间以上 + log.info("CorpusProcessKafkaProducer getCategoryCode: {}，audioFileId：{}，sourceId：{}", tmTelephoneCorpus.getCategoryCode(), aicorpusTelephone.getAudioFileId(), aicorpusTelephone.getSourceId()); + if (Constant.CHANNEL_DCC.equals(tmTelephoneCorpus.getCategoryCode())) { + log.info(" dcc 语料开始处理: {}"); + tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus); - // 条件： 只处理dcc的 10s通话时间以上 - log.info("CorpusProcessKafkaProducer getCategoryCode: {}，audioFileId：{}，sourceId：{}" , tmTelephoneCorpus.getCategoryCode(), aicorpusTelephone.getAudioFileId(), aicorpusTelephone.getSourceId()); - if (Constant.CHANNEL_DCC.equals(tmTelephoneCorpus.getCategoryCode())){ - log.info(" dcc 语料开始处理: {}"); - tmTelephoneCorpusService.saveTelephoneCorpus(tmTelephoneCorpus); - tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone); - } + CompletableFuture.runAsync(() -> { + tmTelephoneCorpusService.runTelephoneCorpusDify(aicorpusTelephone); + }, runTelephoneExecutor); + // 关闭线程池 + runTelephoneExecutor.shutdown(); + } + } catch (Exception e) { + log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}", e.getMessage()); + } + }); + + } + executor.shutdown(); + } // 在这里可以添加对解析后的对象的进一步处理逻辑