pms sap库存逻辑优化 并行查询数据不完整

yudao-module-pms/yudao-module-pms-biz/src/main/java/cn/iocoder/yudao/module/pms/job/sap/SyncSapStockJob.java

@@ -320,13 +320,13 @@ public class SyncSapStockJob implements JobHandler {
             context.destination.ping(); // 测试连接
 
             // 查询基础数据
-            context.existStocks = TenantUtils.execute(1L, () ->
-                    iotSapStockMapper.selectList(new IotSapStockPageReqVO()));
-            context.storageLocations = TenantUtils.execute(1L, () ->
+            context.existStocks = TenantUtils.executeIgnore(() ->
+                    iotSapStockMapper.selectList());
+            context.storageLocations = TenantUtils.executeIgnore(() ->
                     sapOrgApi.getSapOrgByType(3));
-            context.factories = TenantUtils.execute(1L, () ->
+            context.factories = TenantUtils.executeIgnore(() ->
                     sapOrgApi.getSapOrgByType(1));
-            context.depts = TenantUtils.execute(1L, () ->
+            context.depts = TenantUtils.executeIgnore(() ->
                     deptApi.getDeptList());
 
             // 构建映射关系
@@ -338,6 +338,9 @@ public class SyncSapStockJob implements JobHandler {
                     CollUtil.size(context.depts),
                     CollUtil.size(context.existStocks));
 
+            System.out.println(String.format("工厂=%s, 库存地点=%s, 部门=%s, 现有库存=%s",
+                    CollUtil.size(context.factories), CollUtil.size(context.storageLocations), CollUtil.size(context.depts), CollUtil.size(context.existStocks)));
+
             return context;
         } catch (Exception e) {
             log.error("预加载同步上下文失败", e);

yudao-module-pms/yudao-module-pms-biz/src/main/java/cn/iocoder/yudao/module/pms/sap/service/IotSapServiceImpl.java

@@ -768,37 +768,6 @@ public class IotSapServiceImpl implements IotSapService {
                     // 记录错误但继续处理其他工厂
                 }
             });
-            // 统一新增或修改数据
-            // 准备SAP库存日志数据
-            /* List<IotSapStockLogDO> tobeAddedSapStockLogs = new ArrayList<>();
-            if (CollUtil.isNotEmpty(allStockData)) {
-                allStockData.forEach(stock -> {
-                    IotSapStockLogDO sapStockLog = new IotSapStockLogDO();
-                    sapStockLog.setFactoryCode(stock.getWERKS());
-                    sapStockLog.setStorageLocationCode(stock.getLGORT());
-                    sapStockLog.setMaterialCode(stock.getMATNR().replaceFirst("^0+", ""));
-                    sapStockLog.setMaterialName(stock.getMAKTX());
-                    sapStockLog.setQuantity(stock.getLABST());
-                    sapStockLog.setUnitPrice(stock.getJIAGE());
-                    sapStockLog.setUnit(stock.getMEINS());
-                    sapStockLog.setSyncTime(LocalDateTime.now());
-                    tobeAddedSapStockLogs.add(sapStockLog);
-                });
-            }
-
-            TenantUtils.execute(1L, () -> iotSapStockLogMapper.deleteSapStockLogs(null));
-            if (CollUtil.isNotEmpty(tobeAddedSapStockLogs)) {
-                TenantUtils.execute(1L, () -> iotSapStockLogMapper.insertBatch(tobeAddedSapStockLogs));
-            }
-            // 初始化 批量插入SAP 库存 工厂 库存地点
-            if (CollUtil.isNotEmpty(totalTobeAddedStocks)) {
-                TenantUtils.execute(1L, () -> iotSapStockMapper.insertBatch(totalTobeAddedStocks));
-            }
-
-            // pms中存在而且SAP接口也返回的库存数据：更新
-            if (CollUtil.isNotEmpty(totalActualUpdatedStocks)) {
-                TenantUtils.execute(1L, () -> iotSapStockMapper.updateBatch(totalActualUpdatedStocks));
-            } */
 
             // 分批处理库存日志数据
             processStockLogsInBatches(allStockData);
@@ -866,7 +835,7 @@ public class IotSapServiceImpl implements IotSapService {
                 iotSapStockLogMapper.deleteSapStockLogs(null);
                 return null;
             });
-            log.info("成功删除所有库存日志记录");
+            log.error("成功删除所有库存日志记录");
         } catch (Exception e) {
             log.error("删除库存日志记录时发生错误", e);
             throw new RuntimeException("删除库存日志失败", e);
@@ -881,7 +850,7 @@ public class IotSapServiceImpl implements IotSapService {
             return;
         }
 
-        log.info("开始分批插入库存日志，总计 {} 条记录", stockLogs.size());
+        log.error("开始分批插入库存日志，总计 {} 条记录", stockLogs.size());
 
         // 分批处理
         List<List<IotSapStockLogDO>> batches = CollUtil.split(stockLogs, LOG_BATCH_SIZE);
@@ -893,7 +862,7 @@ public class IotSapServiceImpl implements IotSapService {
                     TenantUtils.execute(1L, () -> {
                         try {
                             iotSapStockLogMapper.insertBatch(batch);
-                            log.debug("成功插入一批库存日志，数量: {}", batch.size());
+                            log.error("成功插入一批库存日志，数量: {}", batch.size());
                         } catch (Exception e) {
                             log.error("插入库存日志批次时发生错误，批次大小: {}", batch.size(), e);
                             // 记录错误但继续处理其他批次
@@ -910,7 +879,7 @@ public class IotSapServiceImpl implements IotSapService {
 
         try {
             latch.await();
-            log.info("所有库存日志批次插入完成");
+            log.error("所有库存日志批次插入完成");
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             log.error("等待库存日志插入完成时被中断", e);
@@ -922,11 +891,11 @@ public class IotSapServiceImpl implements IotSapService {
      */
     private void processNewStocksInBatches(List<IotSapStockDO> newStocks) {
         if (CollUtil.isEmpty(newStocks)) {
-            log.info("没有需要新增的库存数据");
+            log.error("没有需要新增的库存数据");
             return;
         }
 
-        log.info("开始分批插入新增库存数据，总计 {} 条记录", newStocks.size());
+        log.error("开始分批插入新增库存数据，总计 {} 条记录", newStocks.size());
 
         List<List<IotSapStockDO>> batches = CollUtil.split(newStocks, BATCH_SIZE);
         CountDownLatch latch = new CountDownLatch(batches.size());
@@ -939,7 +908,7 @@ public class IotSapServiceImpl implements IotSapService {
                         try {
                             iotSapStockMapper.insertBatch(batch);
                             successCount[0] += batch.size();
-                            log.debug("成功插入一批新增库存，数量: {}", batch.size());
+                            log.error("成功插入一批新增库存，数量: {}", batch.size());
                         } catch (Exception e) {
                             log.error("插入新增库存批次时发生错误，批次大小: {}", batch.size(), e);
                         }
@@ -955,7 +924,7 @@ public class IotSapServiceImpl implements IotSapService {
 
         try {
             latch.await();
-            log.info("新增库存数据插入完成，成功插入 {} 条记录", successCount[0]);
+            log.error("新增库存数据插入完成，成功插入 {} 条记录", successCount[0]);
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             log.error("等待新增库存插入完成时被中断", e);
@@ -967,11 +936,11 @@ public class IotSapServiceImpl implements IotSapService {
      */
     private void processUpdatedStocksInBatches(List<IotSapStockDO> updatedStocks) {
         if (CollUtil.isEmpty(updatedStocks)) {
-            log.info("没有需要更新的库存数据");
+            log.error("没有需要更新的库存数据");
             return;
         }
 
-        log.info("开始分批更新库存数据，总计 {} 条记录", updatedStocks.size());
+        log.error("开始分批更新库存数据，总计 {} 条记录", updatedStocks.size());
 
         List<List<IotSapStockDO>> batches = CollUtil.split(updatedStocks, BATCH_SIZE);
         CountDownLatch latch = new CountDownLatch(batches.size());
@@ -984,7 +953,7 @@ public class IotSapServiceImpl implements IotSapService {
                         try {
                             iotSapStockMapper.updateBatch(batch);
                             successCount[0] += batch.size();
-                            log.debug("成功更新一批库存，数量: {}", batch.size());
+                            log.error("成功更新一批库存，数量: {}", batch.size());
                         } catch (Exception e) {
                             log.error("更新库存批次时发生错误，批次大小: {}", batch.size(), e);
                         }
@@ -1000,7 +969,7 @@ public class IotSapServiceImpl implements IotSapService {
 
         try {
             latch.await();
-            log.info("库存数据更新完成，成功更新 {} 条记录", successCount[0]);
+            log.error("库存数据更新完成，成功更新 {} 条记录", successCount[0]);
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             log.error("等待库存更新完成时被中断", e);
@@ -1127,17 +1096,17 @@ public class IotSapServiceImpl implements IotSapService {
         });
 
         if (CollUtil.isNotEmpty(notExistStorageLocations)) {
-            log.warn("工厂 {} 发现 {} 个未维护的库存地点: {}", factoryCode,
+            log.error("工厂 {} 发现 {} 个未维护的库存地点: {}", factoryCode,
                     notExistStorageLocations.size(), notExistStorageLocations);
             notExistStorageLocations.forEach(location ->
-                    log.warn("系统中未维护的工厂编码-库存地点编码: {}", location));
+                    log.error("系统中未维护的工厂编码-库存地点编码: {}", location));
         }
 
         // 查询当前工厂的现有库存数据
         Set<String> existStockKeys = new HashSet<>();
         IotSapStockPageReqVO reqVO = new IotSapStockPageReqVO();
         reqVO.setFactoryCode(factoryCode);
-
+        log.error("当前工厂 {} 已经存在的SAP库存数量：{}", factoryCode, context.existStocks.size());
         // 从预加载的数据中过滤出当前工厂的库存
         List<IotSapStockDO> factoryExistStocks = context.existStocks.stream()
                 .filter(stock -> factoryCode.equals(stock.getFactoryCode()))