refactor: 增加批量处理逻辑以优化设备状态更新和消息发送

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/SafetyProblemTracingGenServiceImpl.java

@@ -143,18 +143,22 @@ public class SafetyProblemTracingGenServiceImpl{
             return;
         }
         List<String> records = mapList.stream().map(m -> m.get("RECORD").toString()).collect(Collectors.toList());
-        // 更新设备状态为未处理（异常）
-        idxBizJgOtherInfoService.lambdaUpdate()
-                .in(IdxBizJgOtherInfo::getRecord, records)
-                .set(IdxBizJgOtherInfo::getStatus, SafetyProblemStatusEnum.UNHANDLED.getCode())
-                .update();
-        List<ESEquipmentCategoryDto> esEquipmentCategoryDto = Lists.newArrayList();
-        for (ESEquipmentCategoryDto equipmentCategoryDto : esEquipmentCategory.findAllById(records)) {
-            equipmentCategoryDto.setProblemStatus(SafetyProblemStatusEnum.UNHANDLED.getCode());
-            esEquipmentCategoryDto.add(equipmentCategoryDto);
+        int batchSize = 10000;
+        for (int i = 0; i < records.size(); i += batchSize) {
+            List<String> batch = records.subList(i, Math.min(i + batchSize, records.size()));
+            // 更新设备状态为未处理（异常）
+            idxBizJgOtherInfoService.lambdaUpdate()
+                    .in(IdxBizJgOtherInfo::getRecord, batch)
+                    .set(IdxBizJgOtherInfo::getStatus, SafetyProblemStatusEnum.UNHANDLED.getCode())
+                    .update();
+            List<ESEquipmentCategoryDto> esEquipmentCategoryDto = Lists.newArrayList();
+            for (ESEquipmentCategoryDto equipmentCategoryDto : esEquipmentCategory.findAllById(batch)) {
+                equipmentCategoryDto.setProblemStatus(SafetyProblemStatusEnum.UNHANDLED.getCode());
+                esEquipmentCategoryDto.add(equipmentCategoryDto);
+            }
+            esEquipmentCategory.saveAll(esEquipmentCategoryDto);
+            sendSafetyProblemMessage(mapList, safetyProblemTypeEnum);
         }
-        esEquipmentCategory.saveAll(esEquipmentCategoryDto);
-        sendSafetyProblemMessage(mapList, safetyProblemTypeEnum);
     }
 
     private void updateEnterpriseAndSendMessage(List<Map<String, Object>> mapList) {
@@ -191,21 +195,16 @@ public class SafetyProblemTracingGenServiceImpl{
      * @param safetyProblemTypeEnum
      */
     private void sendSafetyProblemMessage(List<Map<String, Object>> mapList, SafetyProblemTypeEnum safetyProblemTypeEnum) {
-        int batchSize = 10000;
-        for (int i = 0; i < mapList.size(); i += batchSize) {
-            List<Map<String, Object>> batch = mapList.subList(i, Math.min(i + batchSize, mapList.size()));
-            if (CollectionUtil.isNotEmpty(mapList)){
-                try {
-                    logger.info("发送安全追溯问题主题---->{}", safetyProblemTypeEnum.getTopic());
-                    emqKeeper.getMqttClient().publish(safetyProblemTypeEnum.getTopic(), JSON.toJSONBytes(batch), 2, false);
-                    logger.info("发送安全追溯问题消息成功---->");
-                } catch (MqttException e) {
-                    logger.error("发送安全追溯问题设备信息消息失败---->{}", e.getMessage());
-                    throw new RuntimeException(e);
-                }
+        if (CollectionUtil.isNotEmpty(mapList)){
+            try {
+                logger.info("发送安全追溯问题主题---->{}", safetyProblemTypeEnum.getTopic());
+                emqKeeper.getMqttClient().publish(safetyProblemTypeEnum.getTopic(), JSON.toJSONBytes(mapList), 2, false);
+                logger.info("发送安全追溯问题消息成功---->");
+            } catch (MqttException e) {
+                logger.error("发送安全追溯问题设备信息消息失败---->{}", e.getMessage());
+                throw new RuntimeException(e);
             }
         }
-
     }
 
 //    @Scheduled(cron = "0 0 1 * * ?")