admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

删除无用代码

Browse Source
This commit is contained in:
zren25
2025-05-28 15:32:57 +08:00
parent 2240a8d477
commit d7965e45b9
4 changed files with 10 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/NameplateCorpusJob.java
View File

@@ -36,29 +36,6 @@ public class NameplateCorpusJob {
private TmTelephoneCorpusMapper tmTelephoneCorpusMapper;
@Value("${dify.corpus.nameplate.isUse}")
private boolean isUse;
/**
* 铭牌语料处理
*/
@XxlJob("nameplateCorpusTask")
@PostMapping("nameplateCorpusTask")
public ResultMsg nameplateCorpusTask(@RequestBody String paramJson) {
try {
// 获取任务参数
String param = XxlJobHelper.getJobParam();
if(StringUtils.isEmpty(param)){
param = paramJson;
}
// 执行业务逻辑
XxlJobHelper.log("任务参数: {}", param);
log.info("nameplateCorpusTask 铭牌语料查询处理:{}",param);
tmNameplateCorpusService.runNameplateCorpusDify(param);
} catch (Exception e) {
log.error("nameplateCorpusTask 定时任务补偿处理消息异常",e.getMessage());
}
return ResultMsg.ok();
}
/**
* 铭牌语料处理失败重试
* @param paramJson

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

@@ -13,8 +13,6 @@ import java.util.List;
*/
public interface TmNameplateCorpusService extends IService<TmNameplateCorpus> {
void runNameplateCorpusDify(String paramJson);
void runNameplateCorpusDifyRetry(String paramJson);
void processItem(TmNameplateCorpus item);

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

@@ -25,14 +25,13 @@ import org.apache.rocketmq.spring.core.RocketMQTemplate;
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;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
@@ -77,58 +76,10 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
@Autowired
private DataMaskingRuleService dataMaskingRuleService;
@Autowired
@Resource(name = "threadPoolTaskExecutor")
private ThreadPoolTaskExecutor executor;
@Override
public void runNameplateCorpusDify(String paramJson) {
long startTime = System.currentTimeMillis();
log.info("runNameplateCorpusDify paramJson {}", paramJson);
// 获取当前日期
LocalDate today = LocalDate.now();
// 获取前一天日期
LocalDate yesterday = today.minusDays(1);
// 格式化输出
String formattedDate = yesterday.toString(); // 默认格式为 yyyy-MM-dd
String statTime = formattedDate.concat(" 00:00:00");
String endTime = formattedDate.concat(" 23:59:59");
if (StringUtils.isNotBlank(paramJson)) {
JSONObject paramJsonObj = JSONObject.parseObject(paramJson);
if (null != paramJsonObj && paramJsonObj.containsKey(ConstantStr.statTime) && paramJsonObj.containsKey(ConstantStr.endTime)) {
statTime = paramJsonObj.getString(ConstantStr.statTime);
endTime = paramJsonObj.getString(ConstantStr.endTime);
}
}
Integer total = tmNameplateCorpusMapper.countTmNameplateCorpusByData(statTime, endTime);
int totalPages = PageDto.getTotalPages(total, pageSize);
// 获取消息列表
int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1;
log.info("获取的线程数：{}",optimalThreadPoolSize);
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小
for (int i = 1; i <= totalPages; i++) {
int offset = (i - 1) * pageSize;
List<TmNameplateCorpus> messageList = tmNameplateCorpusMapper.queryTmNameplateCorpusByData(statTime, endTime, offset, pageSize);
// 处理查询到的数据
// 使用 CompletableFuture 并行处理
CompletableFuture<?>[] futures = messageList.stream()
.map(item -> CompletableFuture.runAsync(() -> {
try {
processItem(item);
} catch (Exception e) {
log.error("铭牌语料失败: customerFlowId={}, AcceptUserId={}, 异常: {}",
item.getCustomerFlowId(), e.getMessage(), e);
}
}, executor))
.toArray(CompletableFuture[]::new);
// 等待所有任务完成
CompletableFuture.allOf(futures).join();
}
// 关闭线程池
executor.shutdown();
log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
}
@Override
public void runNameplateCorpusDifyRetry(String paramJson) {
long startTime = System.currentTimeMillis();
@@ -156,13 +107,6 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
Integer total = tmNameplateCorpusMapper.countQueryTmNameplateCorpusRetry(statTime, endTime, customerFlowIds ,retry);
int totalPages = PageDto.getTotalPages(total, pageSize);
// 获取消息列表
int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1;
log.info("获取的线程数：{}",optimalThreadPoolSize);
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小
for (int i = 1; i <= totalPages; i++) {
int offset = (i - 1) * pageSize;
List<TmNameplateCorpus> messageList = tmNameplateCorpusMapper.queryTmNameplateCorpusRetry(statTime, endTime, offset, pageSize, customerFlowIds, retry);
@@ -178,11 +122,7 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
}
}, executor))
.toArray(CompletableFuture[]::new);
// 等待所有任务完成
CompletableFuture.allOf(futures).join();
}
// 关闭线程池
executor.shutdown();
log.info("重跑铭牌语料铭牌数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
}

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

@@ -27,6 +27,7 @@ import org.apache.rocketmq.spring.core.RocketMQTemplate;
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;
@@ -36,8 +37,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
@@ -88,7 +87,9 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
@Autowired
private TmTelephoneCorpusService tmTelephoneCorpusService;
@Autowired
@Resource(name = "threadPoolTaskExecutor")
private ThreadPoolTaskExecutor executor;
@Override
@@ -118,12 +119,6 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
Integer total = tmOdsVdqwMessagearchivingMapper.countOdsVdqwMessageByData(statTime, endTime,retry);
int totalPages = PageDto.getTotalPages(total, pageSize);
// 获取消息列表
int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 1;
log.info("企微处理总数据量：{},总页数：{}，获取的线程数：{}",totalPages,totalPages,optimalThreadPoolSize);
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize); // 根据需求调整线程池大小
try {
for (int i = 1; i <= totalPages; i++) {
int offset = (i - 1) * pageSize;
@@ -143,15 +138,12 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
// 等待所有任务完成
CompletableFuture.allOf(futures).join();
}
// 关闭线程池
executor.shutdown();
} catch (Exception e) {
log.error("企微数据跑批异常",e);
} finally {
// 关闭线程池
executor.shutdown();
log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
}
log.info("企微数据跑批结束 耗时：{}",System.currentTimeMillis()-startTime);
}
private void processItem(OdsVdqwMessageOTD item, String statTime, String endTime) {