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 d00ed9ef1..be4c09dad 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,6 +78,8 @@ 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.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -99,7 +101,7 @@ public class WebCadOrderImportFactory { /** * 批量操作阈值 */ - private static final int BATCH_THRESHOLD_NUMBER = 300; + private static final int BATCH_THRESHOLD_NUMBER = 200; private OrderMapper orderMapper; @@ -1615,14 +1617,58 @@ 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(); // 保存 - saveBodyAsync(latch, hasError, peek); - saveGroupAsync(latch, hasError, peek); - saveItemAsync(latch, hasError, peek); +// 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); try { // 等待所有任务完成 @@ -1631,6 +1677,9 @@ public class WebCadOrderImportFactory { Thread.currentThread().interrupt(); hasError.set(true); log.error("WEBCAD import saveData Thread interrupted", e); + } finally { + // 中止所有任务 + executorService.shutdown(); } if (hasError.get()) { @@ -1767,4 +1816,8 @@ public class WebCadOrderImportFactory { sealEdgeConfigList.clear(); processGroupMap.clear(); } + + public static void main(String[] args) { + System.out.println(Math.ceil(205/100)); + } }