diff --git a/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/parts/PartsServiceImpl.java b/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/parts/PartsServiceImpl.java index 61f0a15ef..b317b3197 100644 --- a/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/parts/PartsServiceImpl.java +++ b/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/parts/PartsServiceImpl.java @@ -38,6 +38,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -301,7 +302,7 @@ public class PartsServiceImpl implements PartsService { try { // 等待所有任务查询完成 - latch.await(); + latch.await(10, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new ServiceException(PARTS_IMPORT_INTERRUPT_ERROR); diff --git a/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/plate/PlateManageServiceImpl.java b/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/plate/PlateManageServiceImpl.java index 48e635290..1205142d2 100644 --- a/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/plate/PlateManageServiceImpl.java +++ b/cf-module-prod-executor/cf-module-prod-executor-biz/src/main/java/com/cf/imes/module/executor/service/manage/plate/PlateManageServiceImpl.java @@ -33,6 +33,7 @@ import java.util.Map; import java.util.Objects; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -244,7 +245,7 @@ public class PlateManageServiceImpl implements PlateManageService { try { // 等待所有任务查询完成 - latch.await(); + latch.await(10, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new ServiceException(PLATE_IMPORT_INTERRUPT_ERROR); diff --git a/cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/factory/WebCadOrderImportFactory.java b/cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/factory/WebCadOrderImportFactory.java index be4c09dad..a19df3a70 100644 --- a/cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/factory/WebCadOrderImportFactory.java +++ b/cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/factory/WebCadOrderImportFactory.java @@ -78,8 +78,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -1500,7 +1499,7 @@ public class WebCadOrderImportFactory { String[] specSplit = spec.split("-"); // 例:15mm-18mm if (specSplit.length == 2) { - return specSplit[0] + "-" + width + "mm-"; + return specSplit[0] + "-" + width + "mm"; } // 例:9mm-18mm-经典檀木 if (specSplit.length == 3) { @@ -1617,69 +1616,27 @@ public class WebCadOrderImportFactory { private void saveDataAsync() { TransactionStatus transactionStatus = transactionManager.getTransaction(TransactionDefinition.withDefaults()); AtomicBoolean hasError = new AtomicBoolean(false); - ExecutorService executorService = Executors.newFixedThreadPool(3); - CountDownLatch latch = new CountDownLatch(3); // 当前数据源data_code String peek = DynamicDataSourceContextHolder.peek(); + List> bodyList = batchProcess(bodyMap.values().stream().toList()); + List> groupList = batchProcess(orderGroupDOS); + List> itemList = batchProcess(orderItemDOS); + // 计数器 + CountDownLatch latch = new CountDownLatch(bodyList.size() + groupList.size() + itemList.size()); + // 保存 -// saveBodyAsync(latch, hasError, peek); -// saveGroupAsync(latch, hasError, peek); -// saveItemAsync(latch, hasError, peek); - Runnable bodyTask = () -> { - try { - // 子线程同步主线程的多数据源 - DynamicDataSourceContextHolder.push(peek); - orderInfoBatchInsertWithinThreshold(orderBodyMapper, bodyMap.values().stream().toList(), true); - } catch (Exception e) { - hasError.set(true); - log.error("Error saving order bodies", e); - } finally { - latch.countDown(); - } - }; - - Runnable itemTask = () -> { - try { - // 子线程同步主线程的多数据源 - DynamicDataSourceContextHolder.push(peek); - orderInfoBatchInsertWithinThreshold(orderItemMapper, orderItemDOS, true); - } catch (Exception e) { - hasError.set(true); - log.error("Error saving order items", e); - } finally { - latch.countDown(); - } - }; - - Runnable groupTask = () -> { - try { - // 子线程同步主线程的多数据源 - DynamicDataSourceContextHolder.push(peek); - orderInfoBatchInsertWithinThreshold(orderGroupMapper, orderGroupDOS, true); - } catch (Exception e) { - hasError.set(true); - log.error("Error saving order groups", e); - } finally { - latch.countDown(); - } - }; - - // 执行子线程 - executorService.submit(bodyTask); - executorService.submit(itemTask); - executorService.submit(groupTask); + saveBodyAsync(bodyList,latch, hasError, peek); + saveGroupAsync(groupList, latch, hasError, peek); + saveItemAsync(itemList, latch, hasError, peek); try { - // 等待所有任务完成 - latch.await(); + // 等待所有任务完成,30s没有完成就主动终止 + latch.await(30, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); hasError.set(true); log.error("WEBCAD import saveData Thread interrupted", e); - } finally { - // 中止所有任务 - executorService.shutdown(); } if (hasError.get()) { @@ -1700,9 +1657,7 @@ public class WebCadOrderImportFactory { * @param hasError * @param peek */ - private void saveBodyAsync(CountDownLatch latch, AtomicBoolean hasError, String peek) { - List orderBodyDOS = bodyMap.values().stream().toList(); - List> bodyList = batchProcess(orderBodyDOS); + private void saveBodyAsync(List> bodyList, CountDownLatch latch, AtomicBoolean hasError, String peek) { for (List bodyDOS : bodyList) { threadPoolTaskExecutor.submit(() -> { try { @@ -1729,8 +1684,7 @@ public class WebCadOrderImportFactory { * @param hasError * @param peek */ - private void saveGroupAsync(CountDownLatch latch, AtomicBoolean hasError, String peek) { - List> groupList = batchProcess(orderGroupDOS); + private void saveGroupAsync(List> groupList, CountDownLatch latch, AtomicBoolean hasError, String peek) { for (List orderGroupDOS : groupList) { threadPoolTaskExecutor.submit(() -> { try { @@ -1757,8 +1711,7 @@ public class WebCadOrderImportFactory { * @param hasError * @param peek */ - private void saveItemAsync(CountDownLatch latch, AtomicBoolean hasError, String peek) { - List> itemList = batchProcess(orderItemDOS); + private void saveItemAsync(List> itemList, CountDownLatch latch, AtomicBoolean hasError, String peek) { for (List itemDOS : itemList) { threadPoolTaskExecutor.submit(() -> { try {