铭牌的多线程控制， 修改企微画像的tag
This commit is contained in:
ZLI263
@@ -20,14 +20,14 @@ public class ExecutorConfig {
|
||||
@Value("${task.pool.corePoolSize}")
|
||||
private int corePoolSize;
|
||||
|
||||
@Value("${task.pool.maxPoolSize}")
|
||||
private int maxPoolSize;
|
||||
// @Value("${task.pool.maxPoolSize}")
|
||||
private int maxPoolSize=12;
|
||||
|
||||
@Value("${task.pool.keepAliveSeconds}")
|
||||
private int keepAliveSeconds;
|
||||
// @Value("${task.pool.keepAliveSeconds}")
|
||||
private int keepAliveSeconds=12000;
|
||||
|
||||
@Value("${task.pool.queueCapacity}")
|
||||
private int queueCapacity;
|
||||
// @Value("${task.pool.queueCapacity}")
|
||||
private int queueCapacity=20;
|
||||
|
||||
private ThreadPoolTaskExecutor executor;
|
||||
private ThreadPoolExecutor blockingExecutor;
|
||||
@@ -38,7 +38,7 @@ public class ExecutorConfig {
|
||||
executor = new ThreadPoolTaskExecutor();
|
||||
|
||||
log.info("Thread-process-pool-Initializing: corePoolSize: {}, maxPoolSize: {}, queueCapacity: {}, keepAliveSeconds: {}",corePoolSize, maxPoolSize, queueCapacity, keepAliveSeconds);
|
||||
// 设置核心线程数
|
||||
// 设置核心线程数 Thread-process-pool-Initializing: corePoolSize: 10, maxPoolSize: 300, queueCapacity: 1000, keepAliveSeconds: 60
|
||||
executor.setCorePoolSize(corePoolSize);
|
||||
// 设置最大线程数为5
|
||||
executor.setMaxPoolSize(maxPoolSize);
|
||||
|
||||
@@ -19,6 +19,7 @@ import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Semaphore;
|
||||
|
||||
/**
|
||||
* @ClassName NameplateKafkaConsumer
|
||||
@@ -40,7 +41,7 @@ public class NameplateKafkaConsumer {
|
||||
@Autowired
|
||||
@Resource(name = "threadPoolTaskExecutor")
|
||||
private ThreadPoolTaskExecutor executor;
|
||||
|
||||
private Semaphore semaphore = new Semaphore(10); // 限制并发数
|
||||
|
||||
@KafkaListener(topics = "${analyticCenterKafka.consumer.topic}", // = smart_assistant_nameplate_topic
|
||||
groupId = "${analyticCenterKafka.consumer.group}" , //smart_assistant_nameplate_topic_group
|
||||
@@ -51,12 +52,13 @@ public class NameplateKafkaConsumer {
|
||||
@Header(KafkaHeaders.OFFSET) Long offset) {
|
||||
long startTime = System.currentTimeMillis();
|
||||
log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId());
|
||||
log.info("nameplateKafkaConsumerMessage总数：{}，消息：{}", recordMessages);
|
||||
log.info("nameplateKafkaConsumerMessage，消息：{}", recordMessages);
|
||||
// 初始化绑定 Consumer
|
||||
if (StringUtils.isEmpty(recordMessages)) {
|
||||
ack.acknowledge();
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
|
||||
if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) {
|
||||
@@ -66,9 +68,13 @@ public class NameplateKafkaConsumer {
|
||||
tmNameplateCorpus.getData().forEach(nameplate ->
|
||||
CompletableFuture.runAsync(() -> {
|
||||
try {
|
||||
semaphore.acquire(); // 获取许可
|
||||
tmNameplateCorpusService.processItem(nameplate);
|
||||
} catch (Exception e) {
|
||||
log.error("corpusPortrait画像铭牌异步任务执行失败", e);
|
||||
} finally {
|
||||
log.info("corpusPortrait 释放锁，availablePermits {}", semaphore.availablePermits());
|
||||
semaphore.release(); // 释放许可
|
||||
}
|
||||
}, executor));
|
||||
|
||||
|
||||
@@ -22,8 +22,10 @@ import org.apache.commons.lang3.StringUtils;
|
||||
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.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.time.LocalDate;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
@@ -73,7 +75,9 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
|
||||
private String aiAnalysisRequestIdFinal = "aiAnalysisRequestId";
|
||||
|
||||
// 使用多线程并行执行两个业务场景
|
||||
ExecutorService executorService = Executors.newFixedThreadPool(6);
|
||||
@Autowired
|
||||
@Resource(name = "threadPoolTaskExecutor")
|
||||
private ThreadPoolTaskExecutor executor;
|
||||
|
||||
@Override
|
||||
public void runNameplateCorpusDifyRetry(String paramJson) {
|
||||
@@ -169,18 +173,18 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
|
||||
|
||||
log.info("铭牌语料第1个业务场景， 优先执行。 ");
|
||||
long startTime = System.currentTimeMillis();
|
||||
CompletableFuture.runAsync(() -> {
|
||||
CompletableFuture<Void> summaryTask = CompletableFuture.runAsync(() -> {
|
||||
//第一个业务场景 开始： 总结和分类业务场景
|
||||
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_NAMEPLATE.getCode(),
|
||||
JSONObject.toJSONString(corpusReportDTO), null);
|
||||
log.info("runDify execDifyFlow ,铭牌语料 ，总结和分类业务，返回： {}", execDifyFlow);
|
||||
}, executorService);
|
||||
}, executor);
|
||||
long endTime = System.currentTimeMillis();
|
||||
log.info("第一个业务场景（总结和分类）执行时间: {} ms", (endTime - startTime));
|
||||
//第一个业务场景， 结束
|
||||
|
||||
long startTime2 = System.currentTimeMillis();
|
||||
CompletableFuture.runAsync(() -> {
|
||||
CompletableFuture<Void> portraitTask = CompletableFuture.runAsync(() -> {
|
||||
//"analysisScene": "分析类型", // 1、企微会话 2、AI通话录音 3、AI铭牌(客流) 4、AI铭牌(试驾)
|
||||
DiFyReq diFyImageReq2 = new DiFyReq();
|
||||
diFyImageReq2.setUser(ConstantStr.corpus_user);
|
||||
@@ -199,9 +203,12 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
|
||||
updateNameplate(execDifyFlowForPortrait.toJSONString());
|
||||
log.info("runDify execDifyFlow ,铭牌语料 ,客户画像场景，返回： {}", execDifyFlowForPortrait);
|
||||
|
||||
}, executorService);
|
||||
}, executor);
|
||||
|
||||
long endTime2 = System.currentTimeMillis();
|
||||
log.info("第二个业务场景（用户画像）执行时间: {} ms", (endTime2 - startTime2));
|
||||
// 等待两个任务完成
|
||||
CompletableFuture.allOf(summaryTask, portraitTask).join();
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("nameplate processItem error {}", e);
|
||||
|
||||
@@ -210,10 +210,8 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
|
||||
String corpusChat = dataMaskingRuleService.runMaskingRule(runMaskingRuleInput);
|
||||
finalChatList.append(corpusChat).append("***");
|
||||
chatList2.append(corpusChat).append("####");
|
||||
chatList3.append(corpusChat).append("\n");
|
||||
chatList4.append(corpusChat).append("\r");
|
||||
log.info("第几循环次数{} ，掩码前：{} ,掩码后：{},合并后的总语料数据 ：{}", externalcontactCount.get(), chat,corpusChat, finalChatList);
|
||||
log.info("chatList2: {} ,chatList3：{} ,chatList4：{},", chatList2, chatList3,chatList4 );
|
||||
|
||||
});
|
||||
if (externalcontactCount.get() < 1) {
|
||||
log.info("没有客户回复的语料，无需解析");
|
||||
@@ -329,7 +327,8 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
|
||||
DiFyReq diFyImageReq2 = new DiFyReq();
|
||||
diFyImageReq2.setUser(ConstantStr.corpus_user);
|
||||
Map<String, Object> inputMap2 = new HashMap<>();
|
||||
inputMap2.put("businessId", unionId);
|
||||
inputMap2.put("businessId", unionId+","+corpusReportDTO.getUserId());
|
||||
inputMap2.put("consultantId", corpusReportDTO.getUserId());//顾问id
|
||||
inputMap2.put("communicateDate", DateUtil.format(maxMsgTimeItem.getMsgTime(), DatePattern.NORM_DATETIME_PATTERN));
|
||||
inputMap2.put("analysisScene", "1");
|
||||
inputMap2.put("version", 2);
|
||||
|
||||
Reference in New Issue
Block a user