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-16 14:25:36 +08:00
parent 3865912527
commit 22dc5a8026
2 changed files with 32 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -68,7 +68,7 @@ public class KafkaConfig {
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000); // 会话1分钟 props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000); // 会话1分钟
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600000); // 可选：允许更长的消费间隔 props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600000); // 可选：允许更长的消费间隔
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000); props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20); // 单次最多拉取的消息数 /** props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20); // 单次最多拉取的消息数 **/
ConcurrentKafkaListenerContainerFactory<String, String> factory = ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>(); new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props)); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));

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

@@ -19,6 +19,7 @@ import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.util.concurrent.*;
/** /**
* @ClassName NameplateKafkaConsumer * @ClassName NameplateKafkaConsumer
@@ -45,40 +46,50 @@ public class NameplateKafkaConsumer {
@Resource @Resource
private RocketMQTemplate rocketMqTemplate; private RocketMQTemplate rocketMqTemplate;
private int messageCount = 0;
@KafkaListener(topics = "${analyticCenterKafka.consumer.topic}", @KafkaListener(topics = "${analyticCenterKafka.consumer.topic}",
groupId = "${analyticCenterKafka.consumer.group}" , groupId = "${analyticCenterKafka.consumer.group}" ,
containerFactory = "analyticCenterConsumerFactory", containerFactory = "analyticCenterConsumerFactory")
concurrency = "3")
public void listen(String recordMessages, Acknowledgment ack, public void listen(String recordMessages, Acknowledgment ack,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partitionId, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partitionId,
@Header(KafkaHeaders.OFFSET) Long offset) { @Header(KafkaHeaders.OFFSET) Long offset) throws InterruptedException {
long startTime = System.currentTimeMillis(); long startTime = System.currentTimeMillis();
log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId()); messageCount++;
log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId(),messageCount);
log.info("nameplateKafkaConsumerMessage: {}", recordMessages); log.info("nameplateKafkaConsumerMessage: {}", recordMessages);
if(StringUtils.isNotEmpty(recordMessages)){ if(StringUtils.isNotEmpty(recordMessages)){
try {
NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class); int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1;
log.info("nameplateKafkaConsumerParseType: {}", tmNameplateCorpus.getType()); log.info("获取的线程数：{}",optimalThreadPoolSize);
if(tmNameplateCorpus.getType().equals("INSERT")){ // 创建线程池
tmNameplateCorpus.getData().forEach(nameplate -> { ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小
log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId()); NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
tmNameplateCorpusService.processItem(nameplate); log.info("nameplateKafkaConsumerParseType: {},size:{}", tmNameplateCorpus.getType(), tmNameplateCorpus.getData().size());
}); if(tmNameplateCorpus.getType().equals("INSERT")){
} CountDownLatch latch = new CountDownLatch(tmNameplateCorpus.getData().size());
} catch (Exception e) { tmNameplateCorpus.getData().forEach(nameplate -> {
log.error("nameplateKafkaConsumer铭牌解析处理出错 {}:{}，ex:{}", CompletableFuture.runAsync(() -> {
partitionId, offset, e.getMessage()); try {
}finally { log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId());
log.info("nameplateKafkaConsumerack.acknowledge"); tmNameplateCorpusService.processItem(nameplate);
} catch (Exception e) {
log.error("nameplateKafkaConsumer铭牌解析处理出错 {}:{}，ex:{}",
partitionId, offset, e.getMessage());
}finally {
log.info("nameplateKafkaConsumerack.acknowledge");
latch.countDown();
}
}, executor);
});
boolean allDone = latch.await(2, TimeUnit.MINUTES);
log.info("nameplateKafkaConsumerack.acknowledge:{}",allDone);
try { try {
ack.acknowledge(); // 确保无论成功与否都提交 offset ack.acknowledge(); // 确保无论成功与否都提交 offset
} catch (IllegalStateException e) { } catch (IllegalStateException e) {
log.warn("Offset 已提交，跳过重复提交"); log.warn("Offset 已提交，跳过重复提交");
} }
} }
}else { }else {
ack.acknowledge(); // 空消息直接跳过 ack.acknowledge(); // 空消息直接跳过
} }
@@ -86,7 +97,5 @@ public class NameplateKafkaConsumer {
log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime); log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);
} }
} }