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 14:04:42 +08:00
parent 69298735f7
commit 1ce61d4bfc
1 changed files with 6 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -43,22 +43,19 @@ public class CorpusProcessKafkaProducer {
@PostMapping("corpusProcessKafkaConsumer") @PostMapping("corpusProcessKafkaConsumer")
@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}") @KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
public void listen(List<ConsumerRecord<String, Object>> recordMessage) { public void listen(List<String> recordMessages) {
try { try {
log.info("CorpusProcessKafkaProducer Received message: {}", recordMessage); log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages);
// 获取消息列表 // 获取消息列表
int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2; int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2;
log.info("获取的线程数：{}", optimalThreadPoolSize); log.info("获取的线程数：{}", optimalThreadPoolSize);
ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize);
if(CollectionUtils.isNotEmpty(recordMessage)){ if(CollectionUtils.isNotEmpty(recordMessages)){
log.info("CorpusProcessKafkaProducer List size: {}", recordMessage.size()); log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size());
for (ConsumerRecord<String, Object> record : recordMessage) { for (String message : recordMessages) {
log.info("CorpusProcessKafkaProducer List record.value: {}",record.value()); log.info("CorpusProcessKafkaProducer message: {}",message);
log.info("CorpusProcessKafkaProducer List record.key: {}",record.key());
executor.submit(() -> { executor.submit(() -> {
try { try {
String message = (String) record.value();
AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class); AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay()); log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay());
DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class); DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);