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-05-15 12:17:05 +08:00
parent 602090f87a
commit 6cfcade1f5
7 changed files with 23 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,10 @@ import java.util.Map;
@Configuration
@RefreshScope
public class KafkaConfig {
/**
*
// 第一个Kafka配置
@Bean(name = "dccKafkaTemplate")
public KafkaTemplate<String, String> dccKafkaTemplate(
@@ -31,7 +35,7 @@ public class KafkaConfig {
return new KafkaTemplate<>(factory);
}
*/
// 第二个Kafka配置
@Bean(name = "analyticCenterKafkaTemplate")
@@ -47,7 +51,7 @@ public class KafkaConfig {
return new KafkaTemplate<>(factory);
}
/**
@Bean(name = "dccConsumerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> dccConsumerFactory(
@Value("${kafka.dcc.bootstrap-servers}") String bootstrapServers,
@@ -65,7 +69,7 @@ public class KafkaConfig {
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
return factory;
}
*/
@Bean(name = "analyticCenterConsumerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> analyticCenterConsumerFactory(
@Value("${kafka.analyticCenter.bootstrap-servers}") String bootstrapServers,
@@ -78,6 +82,7 @@ public class KafkaConfig {
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/CorpusFailJob.java
View File

@@ -146,7 +146,7 @@ public class CorpusFailJob {
List<TmNameplateCorpus> nameplateCorpusList = tmNameplateCorpusService.queryTelephoneCorpusByCustomerFlowId( Arrays.asList(corpusReportDTO.getCustomerFlowId()));
if(CollectionUtils.isNotEmpty(nameplateCorpusList)) {
TmNameplateCorpus nameplateCorpus = nameplateCorpusList.get(0);
tmNameplateCorpusService.sendNameplateLto(execDifyFlow, nameplateCorpus);
tmNameplateCorpusService.sendNameplateLto(execDifyFlow,oldAiAnalysisRequestLogs.getAiAnalysisRequestId(), nameplateCorpus);
}
}else {
List<AicorpusTelephoneDTO> dccDtoList = tmTelephoneCorpusMapper.queryTelephoneCorpusBySourceIds( Arrays.asList(corpusReportDTO.getRecordId()));

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 = "${kafka.dcc.consumer.topic}", groupId = "${kafka.dcc.consumer.group}" ,containerFactory = "dccConsumerFactory")
@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
public void listen(List<ConsumerRecord<String, Object>> recordMessages) {
long startTime = System.currentTimeMillis();
try {

9
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java
View File

@@ -12,6 +12,7 @@ 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.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController;
@@ -47,7 +48,7 @@ public class NameplateKafkaConsumer {
groupId = "${kafka.analyticCenter.consumer.group}" ,
containerFactory = "analyticCenterConsumerFactory",
concurrency = "3")
public void listen(String recordMessages) {
public void listen(String recordMessages, Acknowledgment ack) {
long startTime = System.currentTimeMillis();
log.info("nameplateKafkaConsumer message: {}", recordMessages);
@@ -64,6 +65,12 @@ public class NameplateKafkaConsumer {
} catch (Exception e) {
log.error("nameplateKafkaConsumer铭牌解析处理出错: {}", e.getMessage());
}
// 手动提交 offset
if (ack != null) {
ack.acknowledge();
}
}else {
ack.acknowledge(); // 空消息直接跳过
}
log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/TmNameplateCorpusService.java
View File

@@ -19,7 +19,7 @@ public interface TmNameplateCorpusService extends IService<TmNameplateCorpus> {
void processItem(TmNameplateCorpus item);
void sendNameplateLto(JSONObject execDifyFlow, TmNameplateCorpus tmNameplateCorpus);
void sendNameplateLto(JSONObject execDifyFlow,String aiAnalysisRequestId, TmNameplateCorpus tmNameplateCorpus);
ResultMsg<Object> updateNameplate(String message);

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java
View File

@@ -81,7 +81,7 @@ public class DiFyServiceImpl implements DiFyService{
.aiAnalysisRequestId(aiAnalysisRequestId)
.aiAnalysisRequestType(businessType)
.aiAnalysisErrorHandlingStatus("0")
.aiAnalysisErrorMessage(data.getString("error"))
.aiAnalysisErrorMessage(data.getString("error").substring(0,1000))
.build());
}

6
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmNameplateCorpusServiceImpl.java
View File

@@ -229,7 +229,7 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
@Override
public void sendNameplateLto(JSONObject execDifyFlow, TmNameplateCorpus tmNameplateCorpus) {
public void sendNameplateLto(JSONObject execDifyFlow,String aiAnalysisRequestId, TmNameplateCorpus tmNameplateCorpus) {
log.info("铭牌sendNameplateLto getCustomerFlowId:{}", tmNameplateCorpus.getCustomerFlowId());
if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) {
String text = execDifyFlow.getString("outputs");
@@ -239,7 +239,7 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
try {
ttNameplateRecordMapper.insert(TtNameplateRecord.builder().nameplateCorpusId(tmNameplateCorpus.getId()).customerFlowId(tmNameplateCorpus.getCustomerFlowId()).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(text).build());
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(aiAnalysisRequestId).businessResponse(text).build());
} catch (Exception e) {
log.info(" 铭牌语料处理保存报告异常processItem：{} ", e);
}
@@ -267,7 +267,7 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
queryWrapper.eq(TmNameplateCorpus::getCustomerFlowId, customerFlowId);
List<TmNameplateCorpus> tmNameplateCorpusList = tmNameplateCorpusMapper.selectList(queryWrapper);
if (CollectionUtils.isNotEmpty(tmNameplateCorpusList)){
sendNameplateLto(difyJson, tmNameplateCorpusList.get(0));
sendNameplateLto(difyJson,aiAnalysisRequestId, tmNameplateCorpusList.get(0));
LambdaQueryWrapper<AiAnalysisErrors> errorQueryWrapper = new LambdaQueryWrapper<>();
errorQueryWrapper.eq(AiAnalysisErrors::getAiAnalysisRequestId, aiAnalysisRequestId);
errorQueryWrapper.eq(AiAnalysisErrors::getAiAnalysisErrorHandlingStatus, "0");