admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

铭牌结构调整

Browse Source
This commit is contained in:
zren25
2025-05-08 19:02:56 +08:00
parent 81b8a69942
commit a587dd679a
4 changed files with 229 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java Normal file
View File

@@ -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<String, String> 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<String, Object> 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<String, String> factory = new DefaultKafkaProducerFactory<>(configProps);
return new KafkaTemplate<>(factory);
}
// 第二个Kafka配置
@Bean(name = "analyticCenterKafkaTemplate")
public KafkaTemplate<String, String> 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<String, Object> 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<String, String> factory = new DefaultKafkaProducerFactory<>(configProps);
return new KafkaTemplate<>(factory);
}
@Bean(name = "dccConsumerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> 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<String, Object> 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<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
return factory;
}
@Bean(name = "analyticCenterConsumerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> 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<String, Object> 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<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
return factory;
}
}

20
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/TestController.java
View File

@@ -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<String, String> kafkaProducer;
@PostMapping("/mockMq")
@ApiOperation(value = "补偿处理消息")
@@ -41,5 +47,17 @@ public class TestController {
return ResultMsg.ok("ok");
}
@PostMapping("/mockKafka")
@ApiOperation(value = "补偿处理消息")
public ResultMsg<Object> 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");
}
}

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java
View File

@@ -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<ConsumerRecord<String, Object>> recordMessages) {
long startTime = System.currentTimeMillis();
try {

121
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaProducer.java Normal file
View File

@@ -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<ConsumerRecord<String, Object>> 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<String, Object> 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());
}
}
}