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

修改kafka消费

Browse Source
This commit is contained in:
zren25
2025-03-31 13:02:02 +08:00
parent 70d0a96483
commit 79acc23dde
2 changed files with 76 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -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<ConsumerRecord<String, Object>> 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<String, Object> 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();
}
// 在这里可以添加对解析后的对象的进一步处理逻辑