diff --git a/emis-biz/src/main/java/com/xdadan/erp/emis/service/impl/EmisTmsSiteBatchMissServiceImpl.java b/emis-biz/src/main/java/com/xdadan/erp/emis/service/impl/EmisTmsSiteBatchMissServiceImpl.java index 6a63426cc..32ea753d7 100644 --- a/emis-biz/src/main/java/com/xdadan/erp/emis/service/impl/EmisTmsSiteBatchMissServiceImpl.java +++ b/emis-biz/src/main/java/com/xdadan/erp/emis/service/impl/EmisTmsSiteBatchMissServiceImpl.java @@ -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>> futures = new ArrayList<>(); // 先查询所有数据 @@ -299,77 +315,49 @@ public class EmisTmsSiteBatchMissServiceImpl extends EmisBaseService implements } } - // 关闭线程池 - executorService.shutdown(); - - // // 构建运单号到EmisWaybill的映射 - // Map 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> 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> processFutures = new ArrayList<>(); + log.info("Processing {} batches with batch size: {}, total waybills: {}", + batches.size(), batchSize, allWaybills.size()); + + // 复用现有线程池 + List> processFutures = new ArrayList<>(); for (List batch : batches) { processFutures.add(executorService.submit(() -> processWaybillBatch(batch, transPlanRulesByLineCode))); } - // 收集处理结果 + // 添加超时机制，避免长时间阻塞 List allBatchStatus = new ArrayList<>(); List allMissRecords = new ArrayList<>(); - for (Future future : processFutures) { + int timeoutMinutes = 10; // 10分钟超时 + + for (Future 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 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 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 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(); + } + } } }