demand: TMS系统 - 运输管理 - 漏组批次全量扫描数据量过大会造成CPU飙升问题优化
committer: heyu
This commit is contained in:
parent
2c52db5cca
commit
f655d26ded
@ -4,7 +4,10 @@ import java.util.*;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
import cn.hutool.core.collection.CollectionUtil;
|
||||
import com.xdadan.erp.emis.domain.*;
|
||||
@ -248,6 +251,7 @@ public class EmisTmsSiteBatchMissServiceImpl extends EmisBaseService implements
|
||||
@Async
|
||||
@Override
|
||||
public void autoScanSiteBatch(String lineCode) {
|
||||
ExecutorService executorService = null;
|
||||
try {
|
||||
emisTmsSiteBatchMissMapper.deleteReally(lineCode);
|
||||
long startTime = System.currentTimeMillis();
|
||||
@ -272,8 +276,20 @@ public class EmisTmsSiteBatchMissServiceImpl extends EmisBaseService implements
|
||||
.collect(Collectors.groupingBy(EmisTransPlanRule::getLineCode));
|
||||
|
||||
// 创建线程池
|
||||
int processors = Runtime.getRuntime().availableProcessors();
|
||||
ExecutorService executorService = Executors.newFixedThreadPool(processors);
|
||||
int processors = Runtime.getRuntime().availableProcessors() ;
|
||||
executorService = new ThreadPoolExecutor(
|
||||
processors/2, // 核心线程数
|
||||
processors, // 最大线程数
|
||||
60L, // 空闲线程存活时间
|
||||
TimeUnit.SECONDS, // 时间单位
|
||||
new LinkedBlockingQueue<>(), // 工作队列
|
||||
r -> {
|
||||
Thread t = new Thread(r);
|
||||
t.setName("emis-scan-" + t.getId());
|
||||
t.setDaemon(false);
|
||||
return t;
|
||||
} // 线程工厂
|
||||
);
|
||||
List<Future<List<EmisWaybill>>> futures = new ArrayList<>();
|
||||
|
||||
// 先查询所有数据
|
||||
@ -299,77 +315,49 @@ public class EmisTmsSiteBatchMissServiceImpl extends EmisBaseService implements
|
||||
}
|
||||
}
|
||||
|
||||
// 关闭线程池
|
||||
executorService.shutdown();
|
||||
|
||||
// // 构建运单号到EmisWaybill的映射
|
||||
// Map<String, EmisWaybill> waybillMap = allWaybills.stream()
|
||||
// .collect(Collectors.toMap(EmisWaybill::getBillCode, Function.identity(), (a,
|
||||
// b) -> a));
|
||||
//
|
||||
// allWaybills = allWaybills.stream()
|
||||
// .filter(wb -> {
|
||||
// String virtualRemark = wb.getVirtualRemark();
|
||||
// if (StringUtils.isBlank(virtualRemark)) {
|
||||
// return true;
|
||||
// }
|
||||
// // 解析virtualRemark中的运单号
|
||||
// String[] relatedNos = virtualRemark.split(",");
|
||||
// for (String no : relatedNos) {
|
||||
// EmisWaybill related = waybillMap.get(no.trim());
|
||||
// if (related != null && Objects.equals(related.getProblemType(), 173)) {
|
||||
// return false; // 只要有一个关联运单problemType=173，则过滤掉
|
||||
// }
|
||||
// }
|
||||
// return true;
|
||||
// })
|
||||
// .collect(Collectors.toList());
|
||||
//
|
||||
// allWaybills = allWaybills.stream()
|
||||
// .filter(wb -> !Objects.equals(wb.getProblemType(), 170))
|
||||
//// .filter(wb -> !Objects.equals(wb.getSiteBatchStatus(), "1"))
|
||||
// .collect(Collectors.toList());
|
||||
//
|
||||
// totalProcessed = allWaybills.size();
|
||||
|
||||
// 按1000条数据分组处理
|
||||
int batchSize = 1000;
|
||||
List<List<EmisWaybill>> batches = new ArrayList<>();
|
||||
for (int i = 0; i < allWaybills.size(); i += pageSize) {
|
||||
int end = Math.min(i + pageSize, allWaybills.size());
|
||||
for (int i = 0; i < allWaybills.size(); i += batchSize) {
|
||||
int end = Math.min(i + batchSize, allWaybills.size());
|
||||
batches.add(allWaybills.subList(i, end));
|
||||
}
|
||||
|
||||
if (CollectionUtil.isEmpty(batches)) {
|
||||
log.info("No waybills to process, skip batch processing");
|
||||
return;
|
||||
}
|
||||
|
||||
// 创建新的线程池处理数据
|
||||
executorService = Executors.newFixedThreadPool(1);
|
||||
List<Future<EmisBaseService.ProcessResult>> processFutures = new ArrayList<>();
|
||||
log.info("Processing {} batches with batch size: {}, total waybills: {}",
|
||||
batches.size(), batchSize, allWaybills.size());
|
||||
|
||||
// 复用现有线程池
|
||||
List<Future<ProcessResult>> processFutures = new ArrayList<>();
|
||||
|
||||
for (List<EmisWaybill> batch : batches) {
|
||||
processFutures.add(executorService.submit(() -> processWaybillBatch(batch, transPlanRulesByLineCode)));
|
||||
}
|
||||
|
||||
// 收集处理结果
|
||||
// 添加超时机制，避免长时间阻塞
|
||||
List<EmisWaybillBatchStatus> allBatchStatus = new ArrayList<>();
|
||||
List<EmisTmsSiteBatchMiss> allMissRecords = new ArrayList<>();
|
||||
for (Future<EmisBaseService.ProcessResult> future : processFutures) {
|
||||
int timeoutMinutes = 10; // 10分钟超时
|
||||
|
||||
for (Future<ProcessResult> future : processFutures) {
|
||||
try {
|
||||
ProcessResult result = future.get();
|
||||
ProcessResult result = future.get(timeoutMinutes, TimeUnit.MINUTES);
|
||||
successCount += result.getSuccessCount();
|
||||
failCount += result.getFailCount();
|
||||
failedBillCodes.addAll(result.getFailedBillCodes());
|
||||
allBatchStatus.addAll(result.getBatchStatusList());
|
||||
allMissRecords.addAll(result.getMissRecords());
|
||||
} catch (TimeoutException e) {
|
||||
log.error("Batch processing timeout after {} minutes", timeoutMinutes);
|
||||
future.cancel(true); // 取消超时任务
|
||||
} catch (Exception e) {
|
||||
log.error("Error processing batch", e);
|
||||
}
|
||||
}
|
||||
|
||||
// 关闭线程池
|
||||
executorService.shutdown();
|
||||
|
||||
// 全局去重
|
||||
Map<String, EmisWaybillBatchStatus> batchStatusMap = allBatchStatus.stream()
|
||||
.collect(Collectors.toMap(
|
||||
@ -391,18 +379,18 @@ public class EmisTmsSiteBatchMissServiceImpl extends EmisBaseService implements
|
||||
|
||||
// 批量插入 - 每次插入5000条
|
||||
if (!uniqueBatchStatus.isEmpty()) {
|
||||
int batchSize = 5000;
|
||||
for (int i = 0; i < uniqueBatchStatus.size(); i += batchSize) {
|
||||
int end = Math.min(i + batchSize, uniqueBatchStatus.size());
|
||||
int insertBatchSize = 5000;
|
||||
for (int i = 0; i < uniqueBatchStatus.size(); i += insertBatchSize) {
|
||||
int end = Math.min(i + insertBatchSize, uniqueBatchStatus.size());
|
||||
List<EmisWaybillBatchStatus> batch = uniqueBatchStatus.subList(i, end);
|
||||
emisWaybillBatchStatusMapper.batchInsertEmisWaybillBatchStatus(batch);
|
||||
totalBatchStatusRecords += batch.size();
|
||||
}
|
||||
}
|
||||
if (!uniqueMissRecords.isEmpty()) {
|
||||
int batchSize = 5000;
|
||||
for (int i = 0; i < uniqueMissRecords.size(); i += batchSize) {
|
||||
int end = Math.min(i + batchSize, uniqueMissRecords.size());
|
||||
int insertBatchSize = 5000;
|
||||
for (int i = 0; i < uniqueMissRecords.size(); i += insertBatchSize) {
|
||||
int end = Math.min(i + insertBatchSize, uniqueMissRecords.size());
|
||||
List<EmisTmsSiteBatchMiss> batch = uniqueMissRecords.subList(i, end);
|
||||
emisTmsSiteBatchMissMapper.batchInsertEmisTmsSiteBatchMiss(batch);
|
||||
totalMissRecords += batch.size();
|
||||
@ -425,6 +413,21 @@ public class EmisTmsSiteBatchMissServiceImpl extends EmisBaseService implements
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("Error in autoScanSiteBatch", e);
|
||||
} finally {
|
||||
// 确保线程池资源释放
|
||||
if (executorService != null && !executorService.isShutdown()) {
|
||||
executorService.shutdown();
|
||||
try {
|
||||
if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
|
||||
log.warn("Thread pool did not terminate gracefully, forcing shutdown");
|
||||
executorService.shutdownNow();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Thread pool shutdown interrupted", e);
|
||||
executorService.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user