diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java index 49d00d4..3247e21 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/ExecutorConfig.java @@ -17,17 +17,17 @@ import java.util.concurrent.TimeUnit; @Configuration public class ExecutorConfig { -// @Value("${task.pool.corePoolSize}") - private int corePoolSize=3; + @Value("${task.pool.corePoolSize}") + private int corePoolSize; -// @Value("${task.pool.maxPoolSize}") - private int maxPoolSize=5; + @Value("${task.pool.maxPoolSize}") + private int maxPoolSize; -// @Value("${task.pool.keepAliveSeconds}") + @Value("${task.pool.keepAliveSeconds}") private int keepAliveSeconds=6000000; -// @Value("${task.pool.queueCapacity}") - private int queueCapacity=10; + @Value("${task.pool.queueCapacity}") + private int queueCapacity; private ThreadPoolTaskExecutor executor; private ThreadPoolExecutor blockingExecutor; diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java index 02661ab..4dc76c0 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java @@ -65,45 +65,14 @@ public class NameplateKafkaConsumer { ack.acknowledge(); return; } - - // 使用CountDownLatch等待所有异步任务完成 - CountDownLatch latch = new CountDownLatch(tmNameplateCorpus.getData().size()); - AtomicBoolean hasError = new AtomicBoolean(false); - tmNameplateCorpus.getData().forEach(nameplate -> - CompletableFuture.runAsync(() -> { - try { - if (!semaphore.tryAcquire(5, TimeUnit.SECONDS)) { - log.warn("获取许可超时,跳过处理"); - hasError.set(true); - return; - } + tmNameplateCorpusService.processItem(nameplate)); - tmNameplateCorpusService.processItem(nameplate); - } catch (Exception e) { - log.error("处理铭牌数据失败", e); - hasError.set(true); - } finally { - semaphore.release(); - latch.countDown(); - } - }, executor)); - - // 等待所有任务完成 - if (latch.await(30, TimeUnit.SECONDS)) { - if (!hasError.get()) { - ack.acknowledge(); // 只有所有任务成功才提交offset - } else { - log.error("部分任务失败,不提交offset"); - } - } else { - log.error("任务执行超时,不提交offset"); - } } catch (Exception e) { log.error("处理消息失败", e); } - } + } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java index e1bb296..e4f0c45 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java @@ -325,10 +325,11 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl inputMap2 = new HashMap<>(); - inputMap2.put("businessId", unionId+","+corpusReportDTO.getUserId()); + inputMap2.put("businessId", unionId); inputMap2.put("consultantId", corpusReportDTO.getUserId());//顾问id inputMap2.put("communicateDate", DateUtil.format(maxMsgTimeItem.getMsgTime(), DatePattern.NORM_DATETIME_PATTERN)); inputMap2.put("analysisScene", "1"); + inputMap2.put("consultantId", corpusReportDTO.getUserId()); inputMap2.put("version", 2); inputMap2.put("chat", chatList.toString()); diFyImageReq2.setInputs(inputMap2);