From a587dd679a59a6c69cb7c034f68423f9931b19f5 Mon Sep 17 00:00:00 2001 From: zren25 Date: Thu, 8 May 2025 19:02:56 +0800 Subject: [PATCH] =?UTF-8?q?=E9=93=AD=E7=89=8C=E7=BB=93=E6=9E=84=E8=B0=83?= =?UTF-8?q?=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analytic/center/config/KafkaConfig.java | 88 +++++++++++++ .../center/controller/TestController.java | 20 ++- .../center/mq/CorpusProcessKafkaProducer.java | 2 +- .../center/mq/NameplateKafkaProducer.java | 121 ++++++++++++++++++ 4 files changed, 229 insertions(+), 2 deletions(-) create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java create mode 100644 ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java new file mode 100644 index 0000000..da9a2be --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java @@ -0,0 +1,88 @@ +package com.volvo.ai.analytic.center.config; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.cloud.context.config.annotation.RefreshScope; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; + +import java.util.HashMap; +import java.util.Map; + +@Configuration +@RefreshScope +public class KafkaConfig { + @Value("${kafka.dcc.consumer.key-deserializer}") + String keySerializer; + // 第一个Kafka配置 + @Bean(name = "dccKafkaTemplate") + public KafkaTemplate dccProducerFactory( + @Value("${kafka.dcc.bootstrap-servers}") String bootstrapServers, + @Value("${kafka.dcc.producer.key-serializer}") String keySerializer, + @Value("${kafka.dcc.producer.value-serializer}") String valueSerializer) { + Map configProps = new HashMap<>(); + configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer); + configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer); + DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory<>(configProps); + + return new KafkaTemplate<>(factory); + } + + + // 第二个Kafka配置 + @Bean(name = "analyticCenterKafkaTemplate") + public KafkaTemplate analyticCenterProducerFactory( + @Value("${kafka.analyticCenter.bootstrap-servers}") String bootstrapServers, + @Value("${kafka.analyticCenter.producer.key-serializer}") String keySerializer, + @Value("${kafka.analyticCenter.producer.value-serializer}") String valueSerializer) { + Map configProps = new HashMap<>(); + configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer); + configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer); + DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory<>(configProps); + + return new KafkaTemplate<>(factory); + } + + @Bean(name = "dccConsumerFactory") + public ConcurrentKafkaListenerContainerFactory dccConsumerFactory( + @Value("${kafka.dcc.bootstrap-servers}") String bootstrapServers, + @Value("${kafka.dcc.consumer.group}") String groupId, + @Value("${kafka.dcc.consumer.key-deserializer}") String keyDeserializer, + @Value("${kafka.dcc.consumer.value-deserializer}") String valueDeserializer) { + Map props = new HashMap<>(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer); + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props)); + return factory; + } + + @Bean(name = "analyticCenterConsumerFactory") + public ConcurrentKafkaListenerContainerFactory analyticCenterConsumerFactory( + @Value("${kafka.analyticCenter.bootstrap-servers}") String bootstrapServers, + @Value("${kafka.analyticCenter.consumer.group}") String groupId, + @Value("${kafka.analyticCenter.consumer.key-deserializer}") String keyDeserializer, + @Value("${kafka.analyticCenter.consumer.value-deserializer}") String valueDeserializer) { + Map props = new HashMap<>(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer); + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props)); + return factory; + } + +} \ No newline at end of file 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 64c7396..0a18dbe 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 @@ -9,9 +9,13 @@ 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.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; 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; @@ -31,7 +35,9 @@ public class TestController { @Autowired private AiAnalysisRequestLogsService aiAnalysisRequestLogsService; - + @Autowired + @Qualifier("analyticCenterKafkaTemplate") + private KafkaTemplate kafkaProducer; @PostMapping("/mockMq") @ApiOperation(value = "补偿处理消息") @@ -41,5 +47,17 @@ public class TestController { return ResultMsg.ok("ok"); } + @PostMapping("/mockKafka") + @ApiOperation(value = "补偿处理消息") + public ResultMsg mockKafka(@RequestBody String message) { + JSONObject messageJson = JSONObject.parseObject(message); + String topic = messageJson.getString("topic"); + String msg = messageJson.getString("message"); + + kafkaProducer.send(topic,msg); + return ResultMsg.ok("ok"); + } + + } 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 e6cb9af..6fbbfd5 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 @@ -73,7 +73,7 @@ public class CorpusProcessKafkaProducer { // ), // groupId = "${spring.kafka.group}" // ) - @KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}") + @KafkaListener(topics = "${kafka.dcc.consumer.topic}", groupId = "${kafka.dcc.consumer.group}" ,containerFactory = "dccConsumerFactory") public void listen(List> recordMessages) { long startTime = System.currentTimeMillis(); try { diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java new file mode 100644 index 0000000..0405aa3 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java @@ -0,0 +1,121 @@ + +package com.volvo.ai.analytic.center.mq; + +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.kafka.clients.consumer.ConsumerRecord; +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.RestController; + +import javax.annotation.Resource; +import java.time.LocalDateTime; +import java.util.List; +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 +@RefreshScope +public class NameplateKafkaProducer { + + @Autowired + private TmTelephoneCorpusService tmTelephoneCorpusService; + + private final ObjectMapper objectMapper = new ObjectMapper(); + + @Value("${rocketmq.producer.corpus.dcctopic}") + private String dccMqTipic; + + @Resource + private RocketMQTemplate rocketMqTemplate; + + + @KafkaListener(topics = "${kafka.analyticCenter.consumer.topic}", groupId = "${kafka.analyticCenter.consumer.group}" ,containerFactory = "analyticCenterConsumerFactory") + public void listen(List> recordMessages) { + long startTime = System.currentTimeMillis(); + try { + log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages); + // 获取消息列表 + int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2; + log.info("获取的线程数：{}", optimalThreadPoolSize); + ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); + if(CollectionUtils.isNotEmpty(recordMessages)){ + log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size()); + for (ConsumerRecord record : recordMessages) { + log.info("CorpusProcessKafkaProducer message: {}",record); + 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()); + // 条件：只处理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); + long startTimeDify = System.currentTimeMillis(); + + 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); + + log.info(" dify处理耗时：{}", System.currentTimeMillis() - startTimeDify); + } + } catch (Exception e) { + log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}", e.getMessage()); + } + }); + + } + executor.shutdown(); + } + + log.info("Kafka 消息处理完成，耗时：{}", System.currentTimeMillis() - startTime); + // 在这里可以添加对解析后的对象的进一步处理逻辑 + } catch (Exception e) { + log.error("CorpusProcessKafkaProducer 电话语料 解析JSON出错: {}" , e.getMessage()); + } + } + + + +} +