同步优化

airport-ledger/src/main/java/com/sundot/airport/ledger/service/impl/LedgerSyncServiceImpl.java

@@ -11,6 +11,7 @@ import com.sundot.airport.common.utils.DateUtils;
 import com.sundot.airport.ledger.domain.*;
 import com.sundot.airport.ledger.mapper.*;
 import com.sundot.airport.ledger.service.ILedgerSyncService;
+import com.sundot.airport.ledger.service.IScoreEventService;
 import com.sundot.airport.system.domain.BasePosition;
 import com.sundot.airport.system.mapper.SysDeptMapper;
 import com.sundot.airport.system.mapper.SysUserMapper;
@@ -60,7 +61,7 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
     @Autowired private LedgerExamScoreMapper examScoreMapper;
 
     // ── Score Mappers ─────────────────────────────────────────
-    @Autowired private ScoreEventMapper scoreEventMapper;
+    @Autowired private IScoreEventService scoreEventService;
     @Autowired private ScoreDimensionMapper dimensionMapper;
     @Autowired private ScoreIndicatorMapper indicatorMapper;
     @Autowired private SysDeptMapper sysDeptMapper;
@@ -199,11 +200,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("安全防控能力");
         ScoreIndicator lv2 = getIndicator(IND_SUPERVISION_L2);
         int ins = 0, skip = 0;
+
+        // 优化：收集需要更新的ID，批量更新sync_flag，避免逐条更新导致锁等待超时
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+
         for (LedgerSupervisionProblem row : list) {
-            LambdaUpdateWrapper<LedgerSupervisionProblem> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerSupervisionProblem::getId, row.getId())
-                    .set(LedgerSupervisionProblem::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            supervisionMapper.update(null, uw);
             String src = "supervision_problem:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getInspectedName())) { skip++; continue; }
@@ -221,9 +223,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     row.getLocation(), sv, ZERO,
                     row.getProblemDesc(), src, row.getEvidenceFile(),
                     row.getChannelId(),row.getChannelNo(),row.getAreaId(),row.getPositionId(),row.getPosition());
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerSupervisionProblem> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerSupervisionProblem::getId, subList)
+                    .set(LedgerSupervisionProblem::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            supervisionMapper.update(null, uw);
+        });
+
         return new SyncResult(ins, skip, "");
     }
 
@@ -235,11 +247,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("安全防控能力");
         ScoreIndicator lv2 = getIndicator(IND_INTERCEPT_L2);
         int ins = 0, skip = 0;
+
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+
         for (LedgerRealtimeInterception row : list) {
-            LambdaUpdateWrapper<LedgerRealtimeInterception> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerRealtimeInterception::getId, row.getId())
-                    .set(LedgerRealtimeInterception::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            interceptionMapper.update(null, uw);
             String src = "realtime_interception:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getInspectorName())) { skip++; continue; }
@@ -263,9 +276,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     row.getLocation(), sv, ZERO,
                     row.getItemName(), src, row.getEvidenceFile(),
                     row.getChannelId(),row.getChannelNo(),row.getAreaId(),row.getPositionId(),row.getPosition());
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerRealtimeInterception> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerRealtimeInterception::getId, subList)
+                    .set(LedgerRealtimeInterception::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            interceptionMapper.update(null, uw);
+        });
+
         return new SyncResult(ins, skip, "");
     }
 
@@ -277,11 +300,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("安全防控能力");
         ScoreIndicator lv2 = getIndicator(IND_SECTEST_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerSecurityTest row : list) {
-            LambdaUpdateWrapper<LedgerSecurityTest> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerSecurityTest::getId, row.getId())
-                    .set(LedgerSecurityTest::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            securityTestMapper.update(null, uw);
             String src = "security_test:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!"未通过".equals(row.getTestResult())) { skip++; continue; }
@@ -299,9 +323,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     row.getRegion(), NEG_HALF, ZERO,
                     row.getTestItem(), src, row.getEvidenceFile(),
                     row.getChannelId(),row.getChannel(),row.getAreaId(),row.getPositionId(),row.getTestType());
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerSecurityTest> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerSecurityTest::getId, subList)
+                    .set(LedgerSecurityTest::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            securityTestMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -313,11 +347,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("安全防控能力");
         ScoreIndicator lv2 = getIndicator(IND_UNSAFE_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerUnsafeEvent row : list) {
-            LambdaUpdateWrapper<LedgerUnsafeEvent> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerUnsafeEvent::getId, row.getId())
-                    .set(LedgerUnsafeEvent::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            unsafeEventMapper.update(null, uw);
             String src = "unsafe_event:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getResponsibleName())) { skip++; continue; }
@@ -335,9 +370,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     row.getArea(), sv, ZERO,
                     row.getEventDesc(), src, row.getEvidenceFile(),
                     row.getChannelId(),row.getChannelNo(),row.getAreaId(),row.getPositionId(),row.getPosition());
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerUnsafeEvent> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerUnsafeEvent::getId, subList)
+                    .set(LedgerUnsafeEvent::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            unsafeEventMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -349,11 +394,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("安全防控能力");
         ScoreIndicator lv2 = getIndicator(IND_SEIZURE_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerSeizureStats row : list) {
-            LambdaUpdateWrapper<LedgerSeizureStats> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerSeizureStats::getId, row.getId())
-                    .set(LedgerSeizureStats::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            seizureMapper.update(null, uw);
             String src = "seizure_stats:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getInspectorName())) { skip++; continue; }
@@ -368,9 +414,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     row.getWorkArea(), sv, ZERO,
                     row.getItemName(), src, row.getEvidenceFile(),
                     null,null,row.getAreaId(),row.getPositionId(),row.getPosition());
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerSeizureStats> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerSeizureStats::getId, subList)
+                    .set(LedgerSeizureStats::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            seizureMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -386,11 +442,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("安全防控能力");
         ScoreIndicator lv2 = getIndicator(IND_SEIZURE_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerRewardApproval row : list) {
-            LambdaUpdateWrapper<LedgerRewardApproval> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerRewardApproval::getId, row.getId())
-                    .set(LedgerRewardApproval::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            rewardApprovalMapper.update(null, uw);
             String src = "reward_approval:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getPersonName())) { skip++; continue; }
@@ -405,9 +462,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     null, sv, ZERO,
                     row.getRewardReason(), src, null,
                     null,null,null,null,null);
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerRewardApproval> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerRewardApproval::getId, subList)
+                    .set(LedgerRewardApproval::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            rewardApprovalMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -419,11 +486,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("服务响应能力");
         ScoreIndicator lv2 = getIndicator(IND_SVCPATROL_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerServicePatrol row : list) {
-            LambdaUpdateWrapper<LedgerServicePatrol> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerServicePatrol::getId, row.getId())
-                    .set(LedgerServicePatrol::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            servicePatrolMapper.update(null, uw);
             String src = "service_patrol:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getInspectedName())) { skip++; continue; }
@@ -440,9 +508,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     row.getLocation(), sv, ZERO,
                     row.getProblemDesc(), src, row.getEvidenceFile(),
                     row.getChannelId(),row.getChannelNo(),row.getAreaId(),row.getPositionId(),row.getPosition());
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerServicePatrol> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerServicePatrol::getId, subList)
+                    .set(LedgerServicePatrol::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            servicePatrolMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -454,11 +532,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("服务响应能力");
         ScoreIndicator lv2 = getIndicator(IND_COMPLAINT_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerComplaint row : list) {
-            LambdaUpdateWrapper<LedgerComplaint> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerComplaint::getId, row.getId())
-                    .set(LedgerComplaint::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            complaintMapper.update(null, uw);
             String src = "complaint:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getResponsibleName())) { skip++; continue; }
@@ -468,9 +547,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     null, sv, ZERO,
                     row.getComplaintDesc(), src, row.getEvidenceFile(),
                     null,null,null,null,null);
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerComplaint> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerComplaint::getId, subList)
+                    .set(LedgerComplaint::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            complaintMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -483,11 +572,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         ScoreIndicator lv2 = getIndicator(IND_TYPCASE_L2);
         ScoreIndicator lv3 = getIndicator(IND_TYPCASE_L3);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerTerminalBonus row : list) {
-            LambdaUpdateWrapper<LedgerTerminalBonus> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerTerminalBonus::getId, row.getId())
-                    .set(LedgerTerminalBonus::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            terminalBonusMapper.update(null, uw);
             String src = "terminal_bonus:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getPersonName())) { skip++; continue; }
@@ -497,9 +587,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     null, sv, ZERO,
                     "航站楼加分", src, null,
                     null,null,null,null,null);
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerTerminalBonus> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerTerminalBonus::getId, subList)
+                    .set(LedgerTerminalBonus::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            terminalBonusMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -511,11 +611,12 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
         Long dimId = dimCache.get("业务实操能力");
         ScoreIndicator lv2 = getIndicator(IND_EXAM_L2);
         int ins = 0, skip = 0;
+        
+        // 优化：收集需要更新的ID，批量更新sync_flag
+        List<Long> idsToUpdate = new ArrayList<>();
+        List<ScoreEvent> eventsToInsert = new ArrayList<>();
+        
         for (LedgerExamScore row : list) {
-            LambdaUpdateWrapper<LedgerExamScore> uw = new LambdaUpdateWrapper<>();
-            uw.eq(LedgerExamScore::getId, row.getId())
-                    .set(LedgerExamScore::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
-            examScoreMapper.update(null, uw);
             String src = "exam_score:" + row.getId();
             if (existsBySrc(src)) { skip++; continue; }
             if (!hasName(row.getPersonName())) { skip++; continue; }
@@ -533,9 +634,19 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
                     null, POS_FIVE, ZERO,
                     row.getExamCategory() + " " + row.getExamPeriod(), src, null,
                     null,null,null,null,null);
-            scoreEventMapper.insert(e);
+            eventsToInsert.add(e);
+            idsToUpdate.add(row.getId());
             ins++;
         }
+        
+        // 使用通用批量处理方法
+        batchProcess(eventsToInsert, idsToUpdate, subList -> {
+            LambdaUpdateWrapper<LedgerExamScore> uw = new LambdaUpdateWrapper<>();
+            uw.in(LedgerExamScore::getId, subList)
+                    .set(LedgerExamScore::getSyncFlag, LedgerSyncFlagEnum.ONE.getCode());
+            examScoreMapper.update(null, uw);
+        });
+        
         return new SyncResult(ins, skip, "");
     }
 
@@ -543,6 +654,30 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
     //  Helpers
     // ═════════════════════════════════════════════════════════
 
+    /**
+     * 通用批量处理方法：批量插入ScoreEvent并更新sync_flag
+     * 
+     * @param eventsToInsert 待插入的事件列表
+     * @param idsToUpdate 待更新的ID列表
+     * @param updateFunction 更新函数，接收ID子列表，返回更新的SQL执行结果
+     */
+    private void batchProcess(List<ScoreEvent> eventsToInsert, List<Long> idsToUpdate,
+                              java.util.function.Consumer<List<Long>> updateFunction) {
+        // 批量插入ScoreEvent（每批500条）
+        if (!eventsToInsert.isEmpty()) {
+            scoreEventService.saveBatch(eventsToInsert, 500);
+        }
+
+        // 批量更新sync_flag（每批500条），减少锁持有时间
+        if (!idsToUpdate.isEmpty()) {
+            for (int i = 0; i < idsToUpdate.size(); i += 500) {
+                int end = Math.min(i + 500, idsToUpdate.size());
+                List<Long> subList = idsToUpdate.subList(i, end);
+                updateFunction.accept(subList);
+            }
+        }
+    }
+
     private void loadDimCache() {
         if (!dimCache.isEmpty()) return;
         dimensionMapper.selectList(new ScoreDimension()).forEach(d -> dimCache.put(d.getName(), d.getId()));
@@ -667,7 +802,7 @@ public class LedgerSyncServiceImpl implements ILedgerSyncService {
     }
 
     private boolean existsBySrc(String sourceLedger) {
-        return scoreEventMapper.selectCount(
+        return scoreEventService.count(
                 new LambdaQueryWrapper<ScoreEvent>()
                         .eq(ScoreEvent::getSourceLedger, sourceLedger)) > 0;
     }