调整设备实时消息处理代码

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/IEquipmentSpecificSerivce.java

@@ -181,7 +181,7 @@ public interface IEquipmentSpecificSerivce extends IService<EquipmentSpecific> {
      *
      * @param indexs
      */
-    void updateEquipmentSpecIndexRealtimeData(List<EquipmentSpecificIndex> indexs);
+    void updateEquipmentSpecIndexRealtimeData(EquipmentSpecificIndex index);
 
 
     List<EquipmentSpecificVo> getEquipAndCarIotcodeByIotcode(String iotCode);

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/EquipmentSpecificSerivceImpl.java

@@ -1615,10 +1615,9 @@ public class EquipmentSpecificSerivceImpl extends ServiceImpl<EquipmentSpecificM
     }
 
     @Override
-    public void updateEquipmentSpecIndexRealtimeData(List<EquipmentSpecificIndex> indexs) {
+    public void updateEquipmentSpecIndexRealtimeData(EquipmentSpecificIndex index) {
         // TODO Auto-generated method stub
-        if (!ObjectUtils.isEmpty(indexs)) {
-            EquipmentSpecificIndex index = indexs.get(0);
+        if (!ObjectUtils.isEmpty(index)) {
             EquipmentSpecific es = equipmentSpecificMapper.selectById(index.getEquipmentSpecificId());
             es.setRealtimeIotEsIndexId(index.getId());
             es.setRealtimeIotIndexKey(index.getNameKey());

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -77,7 +77,6 @@ import com.yeejoin.equipmanage.service.IEquipmentIndexService;
 import com.yeejoin.equipmanage.service.IEquipmentService;
 import com.yeejoin.equipmanage.service.IEquipmentSpecificAlarmLogService;
 import com.yeejoin.equipmanage.service.IEquipmentSpecificAlarmService;
-import com.yeejoin.equipmanage.service.IEquipmentSpecificIndexSerivce;
 import com.yeejoin.equipmanage.service.IEquipmentSpecificIndexService;
 import com.yeejoin.equipmanage.service.IEquipmentSpecificSerivce;
 import com.yeejoin.equipmanage.service.IFireFightingSystemService;
@@ -94,9 +93,11 @@ import lombok.extern.slf4j.Slf4j;
 /**
  * @author keyong
  * @title: MqttReceiveServiceImpl
+ * 
  *         <pre>
- * @description: 增量数据处理
+ * &#64;description: 增量数据处理
  *         </pre>
+ * 
  * @date 2020/11/3 13:39
  */
 @Slf4j
@@ -152,9 +153,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	IMainIotMonitorSerivce iMainIotMonitorSerivce;
 
 	@Autowired
-    private IEquipmentSpecificIndexSerivce equipmentSpecificIndexSerivce;
-
-    @Autowired
 	private ISyncDataService syncDataService;
 
 	@Autowired
@@ -202,22 +200,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	@Value("${isSendApp}")
 	private Boolean isSendApp;
 
-   
-
-    private final static Map staticMap = new HashMap();
-
 	private static Boolean bool = Boolean.FALSE;
 
-    static {
-        staticMap.put("FireCar_GDLongitude", "0");
-        staticMap.put("FireCar_GDLatitude", "0");
-        staticMap.put("FireCar_Speed", "0");
-        staticMap.put("FireCar_Power", "0");
-        staticMap.put("FireCar_CourseOverGround", "0");
-        staticMap.put("time", System.currentTimeMillis());
-        staticMap.put("FireCar_Start", "false");
-    }
-
 	@Override
 	@Transactional(rollbackFor = Exception.class)
 	public void handlerMqttIncrementMessage(String topic, String message) {
@@ -259,21 +243,23 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 	/**
 	 * 物联数据处理
+	 * 
 	 * @param topicEntity
 	 * @param iotDatalist
 	 */
-    public void realTimeDateProcessing(TopicEntityVo topicEntity, List<IotDataVO> iotDatalist){
+	public void realTimeDateProcessing(TopicEntityVo topicEntity, List<IotDataVO> iotDatalist) {
 		String iotCode = topicEntity.getIotCode();
 
-  	  if(EquipAndCarEnum.equip.type.equals(topicEntity.getType())){
-  		  List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService.getEquipmentSpeIndexBySpeIotCode(iotCode);
-  		  if (ObjectUtils.isEmpty(indexList)){
+		if (EquipAndCarEnum.equip.type.equals(topicEntity.getType())) {
+			List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService
+					.getEquipmentSpeIndexBySpeIotCode(iotCode);
+			if (ObjectUtils.isEmpty(indexList)) {
 				return;
 			}
 			equipRealTimeDate(iotDatalist, indexList, topicEntity);
-  	  }else{
+		} else {
 			List<CarProperty> carProperties = carPropertyService.getCarPropListByIotCode(iotCode);
-  		  if (ObjectUtils.isEmpty(carProperties)){
+			if (ObjectUtils.isEmpty(carProperties)) {
 				return;
 			}
 			carRealTimeDate(iotDatalist, carProperties);
@@ -281,13 +267,128 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 	}
 
+	/**
+	 * 装备实时数据处理
+	 * 
+	 * @param iotDatalist
+	 * @param indexList
+	 * @param topicEntity
+	 */
+	public void equipRealTimeDate(List<IotDataVO> iotDatalist, List<EquipmentSpecificIndex> indexList,
+			TopicEntityVo topicEntity) {
+		List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
+		List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
+		List<IndexStateVo> indexStateList = new ArrayList<>();
+		// 存储温度数据至内存中（火眼）
+		iotDataListToCacheMap(iotDatalist);
+		iotDatalist.forEach(iotDataVO -> {
+			for (EquipmentSpecificIndex equipmentSpecificIndex : indexList) {
+				if (!ObjectUtils.isEmpty(equipmentSpecificIndex.getNameKey())
+						&& equipmentSpecificIndex.getNameKey().toLowerCase().equals(iotDataVO.getKey().toLowerCase())) {
+					EquipmentSpecificIndex equipmentSpeIndex = new EquipmentSpecificIndex();
+					BeanUtils.copyProperties(equipmentSpecificIndex, equipmentSpeIndex);
+					String value = iotDataVO.getValue().toString();
+					equipmentSpeIndex.setValue(value);
+					equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpecificIndex.getValueEnum()));
+					equipmentSpeIndex.setUpdateDate(new Date());
+					equipmentSpecificIndexService.updateById(equipmentSpeIndex);
+
+					// 更新设备表指标状态
+					iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);
+
+					equipmentSpecificIndexList.add(equipmentSpeIndex);
+
+					indexStateList.add(createIndexStateVo(equipmentSpeIndex));
+
+					// 添加指标报告
+					saveEquipmentAlarmReportDay(equipmentSpeIndex);
+
+					// 火眼数据构造告警指标逻辑
+					equipmentSpecificIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
+
+					// 指标告警处理
+					if (equipmentSpecificIndex.getIsAlarm() != null && 1 == equipmentSpecificIndex.getIsAlarm()) {
+						equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpecificIndex));
+					}
+				}
+			}
+		});
+
+		// 报警数据保存
+		List<EquipmentSpecificAlarmLog> alarmLogs = saveOrUpdateEquipAlarm(equipmentSpecificAlarms);
 
-    public static List<EquipmentSpecificAlarmLog> upAlarmLogStatus(String iotCode, String equipmentSpecificIndexKey, IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService) {
+		// 需要在事务提交之后，否侧事务隔离查询不出数据
+		TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+			@Override
+			public void afterCommit() {
+				// 向预控系统发送消息
+				sendEquipSpecIndexToAutosysTopic(equipmentSpecificIndexList);
+
+				// 首页性能指标数据订阅
+				mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));
+
+				// 组态大屏消息推送，设备表实时指标修改
+				intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);
+
+				// 数字换流站同步指标修改
+				syncSpecificIndexsToGS(equipmentSpecificIndexList);
+
+				// 则更新拓扑节点数据及告警状态
+				updateNodeDateByEquipId(equipmentSpecificIndexList);
+
+				// 向画布推送
+				publishDataToCanvas(equipmentSpecificIndexList);
+
+				// 向其他系统推送报警
+				equipmentAlarmLogsToOtherSystems(alarmLogs);
+			}
+		});
+	}
+
+	public void carRealTimeDate(List<IotDataVO> iotDatalist, List<CarProperty> carProperties) {
+		List<CarProperty> carIndexsList = new ArrayList<>();
+		iotDatalist.forEach(iotDataVO -> {
+			// 对指标key为labels的数据处理
+			if (EquipmentIndexLabelsEnum.labels.name.equals(iotDataVO.getKey())) {
+				StringBuilder sb = new StringBuilder("equipmentOnCar_");
+				EquipmentIndexLabelsVo labelsVo = new EquipmentIndexLabelsVo();
+				Object obj = iotDataVO.getValue();
+				if (obj instanceof JSONArray) {
+					List<String> labelList = (List<String>) obj;
+					labelList.forEach(code -> {
+						String key = sb.append(code).toString();
+						labelsVo.setEquipmentIotCode(code);
+						labelsVo.setTime(new Date());
+						redisUtils.set(key, com.alibaba.fastjson.JSONObject.toJSONString(labelsVo), redisExpireTime);
+					});
+				}
+			}
+			List<CarPropertyVo> carPropertyVos = new ArrayList<>();
+			carProperties.forEach(carProperty -> {
+				if (iotDataVO.getKey().equals(carProperty.getNameKey())) {
+					carProperty.setValue(iotDataVO.getValue().toString());
+					carProperty.setUpdateDate(new Date());
+					carPropertyVos.add(carPropertyToCarPropertyVo(carProperty));
+					carIndexsList.add(carProperty);
+				}
+			});
+
+			boolean updateBatchById = carPropertyService.updateBatchById(carIndexsList);
+
+			if (updateBatchById) {
+				carTransactionSynch(carProperties, carPropertyVos);
+			}
+		});
+	}
+
+	public static List<EquipmentSpecificAlarmLog> upAlarmLogStatus(String iotCode, String equipmentSpecificIndexKey,
+			IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService) {
 		LambdaQueryWrapper<EquipmentSpecificAlarmLog> queryWrapper = new LambdaQueryWrapper();
 		queryWrapper.eq(EquipmentSpecificAlarmLog::getIotCode, iotCode);
 		queryWrapper.eq(EquipmentSpecificAlarmLog::getEquipmentSpecificIndexKey, equipmentSpecificIndexKey);
 		queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
-        List<EquipmentSpecificAlarmLog> logs = equipmentSpecificAlarmLogService.getBaseMapper().selectList(queryWrapper);
+		List<EquipmentSpecificAlarmLog> logs = equipmentSpecificAlarmLogService.getBaseMapper()
+				.selectList(queryWrapper);
 		logs.forEach(x -> {
 			x.setCleanTime(new Date());
 			x.setStatus(AlarmStatusEnum.HF.getCode());
@@ -298,11 +399,13 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		return logs;
 	}
 
+	/**
+	 * 画布数据消息推送
+	 * 
+	 * @param indexList
+	 */
 	public void publishDataToCanvas(List<EquipmentSpecificIndex> indexList) {
 		if (!ObjectUtils.isEmpty(indexList)) {
-            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
-                @Override
-                public void afterCommit() {
 			EquipmentSpecificIndex index = indexList.stream().filter(x -> x.getUpdateDate() != null)
 					.sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
 					.collect(Collectors.toList()).get(0);
@@ -329,23 +432,27 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 			// 发送数据至画布
 			mqttSendGateway.sendToMqtt(canvasTopic, JSON.toJSONString(map));
 		}
-            });
-        }
 	}
 
+	/**
+	 * 报警消息推送amos平台
+	 * 
+	 * @param equipmentSpecificAlarmLog
+	 */
 	void syncSystemctlMsg(EquipmentSpecificAlarmLog equipmentSpecificAlarmLog) {
 		try {
 			MessageModel model = new MessageModel();
-            String alarmReason = ValidationUtil.isEmpty(equipmentSpecificAlarmLog.getAlarmReason()) ? "" : equipmentSpecificAlarmLog.getAlarmReason();
+			String alarmReason = ValidationUtil.isEmpty(equipmentSpecificAlarmLog.getAlarmReason()) ? ""
+					: equipmentSpecificAlarmLog.getAlarmReason();
 			if (alarmReason.contains("：")) {
 				String[] split = alarmReason.split("：");
 				alarmReason = split[1];
 			}
 			model.setTitle(equipmentSpecificAlarmLog.getEquipmentSpecificIndexName());
 			String body = String.format("警情类型：%s；报警设备：%s；报警位置：%s；报警原因：%s；报警时间：%s",
-                    equipmentSpecificAlarmLog.getEquipmentSpecificIndexName(), equipmentSpecificAlarmLog.getEquipmentSpecificName(),
-                    equipmentSpecificAlarmLog.getLocation(), alarmReason,
-                    DateUtils.date2LongStr(equipmentSpecificAlarmLog.getCreateDate()));
+					equipmentSpecificAlarmLog.getEquipmentSpecificIndexName(),
+					equipmentSpecificAlarmLog.getEquipmentSpecificName(), equipmentSpecificAlarmLog.getLocation(),
+					alarmReason, DateUtils.date2LongStr(equipmentSpecificAlarmLog.getCreateDate()));
 			model.setBody(body);
 			model.setMsgType("iotMonitor");
 			if (isSendApp) {
@@ -362,56 +469,11 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 			systemctlFeign.create(token.getAppKey(), token.getProduct(), token.getToke(), model);
 			log.info(String.format("调用平台消息服务成功：%s", JSON.toJSONString(model)));
 		} catch (Exception e) {
-//			e.printStackTrace();
+			// e.printStackTrace();
 			log.error("告警消息同步平台失败：syncSystemctlMsg，===>>>" + e.getMessage());
 		}
 	}
 
-    public List<EquipmentSpecificAlarm> addIndexAlarmRecord(EquipmentSpecificIndex equipmentSpcIndex, List<IotDataVO> iotDatalist) {
-        // 处理火眼视频异常
-        EquipmentSpecificIndex equipmentSpecificIndex = handleTemperatureAlarm(equipmentSpcIndex, iotDatalist);
-        List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
-        EquipmentSpecificAlarm equipmentSpecificAlarm = new EquipmentSpecificAlarm();
-        equipmentSpecificAlarm.setSystemIds(equipmentSpcIndex.getSystemId());
-        equipmentSpecificAlarm.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpcIndex.getSystemId()));
-        if (EquipmentRiskTypeEnum.GZ.getCode().equals(equipmentSpecificIndex.getTypeCode()) || EquipmentRiskTypeEnum.HZGJ.getCode().equals(equipmentSpecificIndex.getTypeCode())
-                || EquipmentRiskTypeEnum.PB.getCode().equals(equipmentSpecificIndex.getTypeCode())) {
-            List<EquipmentSpecificAlarm> indexAlarms = equipmentSpecificAlarmMapper.findEquipmentSpecificAlarmByEquipmentSpecificIdAndEquipmentIndexIdAndStatusIstrue(equipmentSpecificIndex.getEquipmentSpecificId(), equipmentSpecificIndex.getEquipmentIndexId());
-            // NB设备告警
-            if (verifyNB(equipmentSpecificIndex.getNameKey())) {
-                return getNbEquipAlarmList(indexAlarms, equipmentSpecificIndex, equipmentSpecificAlarm);
-            }
-            // 报警表新增信息
-            if (ObjectUtils.isEmpty(indexAlarms) && (TrueOrFalseEnum.real.value.equals(equipmentSpecificIndex.getValue()))) {
-                addEquipmentSpecificAlarm(equipmentSpecificAlarms, equipmentSpecificIndex, equipmentSpecificAlarm);
-            } else {
-                // 报警表更新信息
-                indexAlarms.forEach(action -> {
-                    if (TrueOrFalseEnum.real.value.equals(equipmentSpecificIndex.getValue())) {
-                        // 报警，修改发生频率
-                        action.setFrequency((action.getFrequency() + 1));
-                    } else {
-                        // 报警恢复，修改数据为恢复状态
-                        action.setRecoveryDate(new Date());
-                        action.setEquipmentSpecificIndexValue(TrueOrFalseEnum.fake.value);
-                        action.setStatus(AlarmStatusEnum.HF.getCode());
-                    }
-                    action.setUpdateDate(new Date());
-                    // 更新所在系统,设备可能编辑过,更新所在系统、装备名称、装备定义code
-                    action.setSystemIds(equipmentSpcIndex.getSystemId());
-                    action.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpcIndex.getSystemId()));
-                    action.setEquipmentSpecificName(equipmentSpcIndex.getEquipmentSpecificName());
-                    action.setEquipmentCode(equipmentSpcIndex.getEquipmentCode());
-                    // 冗余字段，alarm_log表更新时使用
-                    action.setEquipmentSpecificCode(equipmentSpcIndex.getEquipmentSpecificCode());
-                    equipmentSpecificAlarms.add(action);
-                });
-            }
-        }
-
-        return equipmentSpecificAlarms;
-    }
-
 	private String getSystemCodeBySpeId(String sysIds) {
 		if (StringUtil.isNotEmpty(sysIds)) {
 			List<FireFightingSystemEntity> sys = fireFightingSystemMapper.getFightingSysByIds(sysIds.split(","));
@@ -433,7 +495,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		equipmentAlarmReportDay.setEquipmentDetailId(equipmentSpecificIndex.getEquipmentDetailId());
 		equipmentAlarmReportDay.setEquipmentId(equipmentSpecificIndex.getEquipmentId());
 		equipmentAlarmReportDay.setEquipmentCode(equipmentSpecificIndex.getEquipmentCode());
-        equipmentAlarmReportDay.setIndexTrueNum(TrueOrFalseEnum.real.value.equals(equipmentSpecificIndex.getValue()) ? 1L : 0L);
+		equipmentAlarmReportDay
+				.setIndexTrueNum(TrueOrFalseEnum.real.value.equals(equipmentSpecificIndex.getValue()) ? 1L : 0L);
 		equipmentAlarmReportDay.setAlarmType(equipmentSpecificIndex.getTypeCode());
 		equipmentAlarmReportDay.setAlarmTypeName(equipmentSpecificIndex.getTypeName());
 		equipmentAlarmReportDay.setIndexId(equipmentSpecificIndex.getEquipmentIndexId());
@@ -453,7 +516,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		boolean bool = equipmentSpecificAlarmLogService.save(equipmentSpecificAlarmLog);
 		// 同步告警消息给平台
 		if (amosSwitch && bool) {
-            EquipmentSpecificAlarmLog alarmLog = equipmentSpecificAlarmLogService.getById(equipmentSpecificAlarmLog.getId());
+			EquipmentSpecificAlarmLog alarmLog = equipmentSpecificAlarmLogService
+					.getById(equipmentSpecificAlarmLog.getId());
 			new Thread(new Runnable() {
 				@Override
 				public void run() {
@@ -471,7 +535,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	 * @param equipmentSpecificIndex
 	 * @return
 	 */
-    private List<EquipmentSpecificAlarm> addEquipmentSpecificAlarm(List<EquipmentSpecificAlarm> equipmentSpecificAlarms, EquipmentSpecificIndex equipmentSpecificIndex, EquipmentSpecificAlarm equipmentSpecificAlarm) {
+	private List<EquipmentSpecificAlarm> addEquipmentSpecificAlarm(List<EquipmentSpecificAlarm> equipmentSpecificAlarms,
+			EquipmentSpecificIndex equipmentSpecificIndex, EquipmentSpecificAlarm equipmentSpecificAlarm) {
 		BeanUtils.copyProperties(equipmentSpecificIndex, equipmentSpecificAlarm);
 		equipmentSpecificAlarm.setId(null);
 		equipmentSpecificAlarm.setFrequency(1);
@@ -494,7 +559,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	}
 
 	// NB装备告警
-    private List<EquipmentSpecificAlarm> getNbEquipAlarmList(List<EquipmentSpecificAlarm> indexAlarms, EquipmentSpecificIndex equipmentSpecificIndex, EquipmentSpecificAlarm equipmentSpecificAlarm) {
+	private List<EquipmentSpecificAlarm> getNbEquipAlarmList(List<EquipmentSpecificAlarm> indexAlarms,
+			EquipmentSpecificIndex equipmentSpecificIndex, EquipmentSpecificAlarm equipmentSpecificAlarm) {
 		List<EquipmentSpecificAlarm> equipmentSpecificAlarmList = new ArrayList<>();
 		if (ValidationUtil.isEmpty(indexAlarms)) { // 告警表为空，新增告警数据
 			addEquipmentSpecificAlarm(equipmentSpecificAlarmList, equipmentSpecificIndex, equipmentSpecificAlarm);
@@ -507,7 +573,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 			indexAlarms.forEach(action -> {
 				// 状态为正常或报警解除
 				if (checkStateIsNormal(action, equipmentSpecificIndex)) {
-                    //修改报警数据为正常
+					// 修改报警数据为正常
 					action.setRecoveryDate(new Date());
 					action.setStatus(AlarmStatusEnum.HF.getCode());
 					// 修改之前数据为已处理
@@ -530,8 +596,10 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 	private boolean ifSendToGis(List<CarIndexGisVo> list) {
 		boolean flag = true;
-        List<CarIndexGisVo> list1 = list.stream().filter(x -> CarForGisEnum.JD.getNameKey().equals(x.getNameKey())).collect(Collectors.toList());
-        List<CarIndexGisVo> list2 = list.stream().filter(x -> CarForGisEnum.WD.getNameKey().equals(x.getNameKey())).collect(Collectors.toList());
+		List<CarIndexGisVo> list1 = list.stream().filter(x -> CarForGisEnum.JD.getNameKey().equals(x.getNameKey()))
+				.collect(Collectors.toList());
+		List<CarIndexGisVo> list2 = list.stream().filter(x -> CarForGisEnum.WD.getNameKey().equals(x.getNameKey()))
+				.collect(Collectors.toList());
 		for (CarIndexGisVo gisVo : list) {
 			if (CarForGisEnum.JD.getNameKey().equals(gisVo.getNameKey())) {
 				if (!StringUtil.isNotEmpty(gisVo.getValue()) || "0".equals(gisVo.getValue())) {
@@ -556,10 +624,14 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	 * 接收到的IOT数据为火眼存储到Map中
 	 */
 	private void iotDataListToCacheMap(List<IotDataVO> iotDatalist) {
-        List<IotDataVO> iotDataVOs = iotDatalist.stream().filter(x -> "alarmLevel".equals(x.getKey()) || "alarmType".equals(x.getKey()) || "temperature".equals(x.getKey())
-                || "ruleTemperature".equals(x.getKey()) || "thermometryUnit".equals(x.getKey()) || "alarmRule".equals(x.getKey())).collect(Collectors.toList());
+		List<IotDataVO> iotDataVOs = iotDatalist.stream()
+				.filter(x -> "alarmLevel".equals(x.getKey()) || "alarmType".equals(x.getKey())
+						|| "temperature".equals(x.getKey()) || "ruleTemperature".equals(x.getKey())
+						|| "thermometryUnit".equals(x.getKey()) || "alarmRule".equals(x.getKey()))
+				.collect(Collectors.toList());
 		if (iotDataVOs.size() > 0) {
-            Map<String, Object> map = iotDatalist.stream().collect(Collectors.toMap(IotDataVO::getKey, IotDataVO::getValue));
+			Map<String, Object> map = iotDatalist.stream()
+					.collect(Collectors.toMap(IotDataVO::getKey, IotDataVO::getValue));
 			putTemperatureMap(map.get("traceId").toString(), map);
 		}
 	}
@@ -567,14 +639,19 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	/**
 	 * 处理火眼逻辑合并为一条告警
 	 */
-    private EquipmentSpecificIndex handleTemperatureAlarm(EquipmentSpecificIndex equipmentSpecificIndex, List<IotDataVO> iotDatalist) {
-        List<IotDataVO> collect = iotDatalist.stream().filter(x -> "traceId".equals(x.getKey())).collect(Collectors.toList());
+	private EquipmentSpecificIndex handleTemperatureAlarm(EquipmentSpecificIndex equipmentSpecificIndex,
+			List<IotDataVO> iotDatalist) {
+		List<IotDataVO> collect = iotDatalist.stream().filter(x -> "traceId".equals(x.getKey()))
+				.collect(Collectors.toList());
 		if (collect.size() > 0 && temperatureMapIsEmpty(String.valueOf(collect.get(0).getValue()))) {
 			String traceId = String.valueOf(collect.get(0).getValue());
 			TemperatureAlarmDto dto = temperatureMap.get(traceId);
-            equipmentSpecificIndex.setEquipmentSpecificIndexName(AlarmTypeEnum.getTypeByCode(AlarmTypeEnum.GZGJ.getCode()));
+			equipmentSpecificIndex
+					.setEquipmentSpecificIndexName(AlarmTypeEnum.getTypeByCode(AlarmTypeEnum.GZGJ.getCode()));
 			equipmentSpecificIndex.setNameKey(AlarmTypeEnum.GZGJ.getCode());
-            equipmentSpecificIndex.setAlamReason(TemperatureAlarm.getAlarmContent(dto.getAlarmLevel(), dto.getAlarmType(), dto.getAlarmRule(), dto.getRuleTemperature(), dto.getTemperature(), dto.getThermometryUnit()));
+			equipmentSpecificIndex.setAlamReason(
+					TemperatureAlarm.getAlarmContent(dto.getAlarmLevel(), dto.getAlarmType(), dto.getAlarmRule(),
+							dto.getRuleTemperature(), dto.getTemperature(), dto.getThermometryUnit()));
 			equipmentSpecificIndex.setValue("true");
 			equipmentSpecificIndex.setIsAlarm(1);
 			temperatureMap.remove(traceId);
@@ -597,14 +674,15 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 	private boolean temperatureMapIsEmpty(String traceId) {
 		TemperatureAlarmDto dto = temperatureMap.get(traceId);
-        if (!ValidationUtil.isEmpty(dto) && !ValidationUtil.isEmpty(dto.getAlarmLevel()) && !ValidationUtil.isEmpty(dto.getAlarmType()) && !ValidationUtil.isEmpty(dto.getAlarmRule())
-                && !ValidationUtil.isEmpty(dto.getRuleTemperature()) && !ValidationUtil.isEmpty(dto.getTemperature()) && !ValidationUtil.isEmpty(dto.getThermometryUnit())) {
+		if (!ValidationUtil.isEmpty(dto) && !ValidationUtil.isEmpty(dto.getAlarmLevel())
+				&& !ValidationUtil.isEmpty(dto.getAlarmType()) && !ValidationUtil.isEmpty(dto.getAlarmRule())
+				&& !ValidationUtil.isEmpty(dto.getRuleTemperature()) && !ValidationUtil.isEmpty(dto.getTemperature())
+				&& !ValidationUtil.isEmpty(dto.getThermometryUnit())) {
 			return true;
 		}
 		return false;
 	}
 
-
 	/**
 	 * 判断是否为NB物联监测设备参数
 	 *
@@ -618,8 +696,10 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 	/**
 	 * 判断状态为正常或报警解除
 	 */
-    private boolean checkStateIsNormal(EquipmentSpecificAlarm equipmentSpecificAlarm, EquipmentSpecificIndex equipmentSpecificIndex) {
-        String enumKey = String.format("%s_%s", equipmentSpecificAlarm.getEquipmentSpecificIndexKey(), equipmentSpecificIndex.getValue());
+	private boolean checkStateIsNormal(EquipmentSpecificAlarm equipmentSpecificAlarm,
+			EquipmentSpecificIndex equipmentSpecificIndex) {
+		String enumKey = String.format("%s_%s", equipmentSpecificAlarm.getEquipmentSpecificIndexKey(),
+				equipmentSpecificIndex.getValue());
 		equipmentSpecificAlarm.setAlamReason(equipmentSpecificAlarm.getEquipmentSpecificIndexName().replace("NB_", "")
 				+ "：" + NBalarmEnum.getDescByKey(enumKey));
 		boolean flag = false;
@@ -638,82 +718,14 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		return flag;
 	}
 
-  
-    /**
-     * 装备实时数据处理
-     * @param iotDatalist
-     * @param indexList
-     * @param topicEntity
-     */
-  private void equipRealTimeDate(List<IotDataVO> iotDatalist, List<EquipmentSpecificIndex> indexList, TopicEntityVo topicEntity){
-	  List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
-	  List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
-	  List<IndexStateVo> indexStateList = new ArrayList<>();
-	// 存储温度数据至内存中（火眼）
-	  iotDataListToCacheMap(iotDatalist);
-    
-	  iotDatalist.forEach( iotDataVO -> {
-  		 for (EquipmentSpecificIndex equipmentSpecificIndex : indexList) {
-  			  if (!ObjectUtils.isEmpty(equipmentSpecificIndex.getNameKey())
-  					  && equipmentSpecificIndex.getNameKey().toLowerCase().equals(iotDataVO.getKey().toLowerCase())) {
-  				  EquipmentSpecificIndex equipmentSpeIndex = new EquipmentSpecificIndex();
-  				  BeanUtils.copyProperties(equipmentSpecificIndex, equipmentSpeIndex);
-                String value = iotDataVO.getValue().toString();
-                equipmentSpeIndex.setValue(value);
-                equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpecificIndex.getValueEnum()));
-                equipmentSpeIndex.setUpdateDate(new Date());
-                equipmentSpecificIndexService.updateById(equipmentSpeIndex);
-                
-                equipmentSpecificIndexList.add(equipmentSpeIndex);
-                
-                indexStateList.add(createIndexStateVo(equipmentSpeIndex));
-                
-                // 向预控系统发送消息
-                sendEquipSpecIndexToAutosysTopic(equipmentSpeIndex);
-                
-                                
-                // 添加指标报告
-                saveEquipmentAlarmReportDay(equipmentSpeIndex);
-                
-                //火眼数据构造告警指标逻辑
-                equipmentSpecificIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
-                
-                //指标告警处理
-                if(equipmentSpecificIndex.getIsAlarm() !=null &&  1 == equipmentSpecificIndex.getIsAlarm()){
-                	equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpecificIndex));
-                }
-                    
-  			  }
-  		 }
-	  });
-
-	  // 首页性能指标数据订阅
-      mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));
-      
-      //组态大屏消息推送，设备表实时指标修改
-      intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);
-
-      //数字换流站同步指标修改
-      syncSpecificIndexsToGS(equipmentSpecificIndexList);
-      
-      // 报警数据保存
-      saveOrUpdateEquipAlarm(equipmentSpecificAlarms);
-      
-      //则更新拓扑节点数据及告警状态
-      updateNodeDateByEquipId(equipmentSpecificIndexList);
-
-      //向画布推送
-      publishDataToCanvas(equipmentSpecificIndexList);
-  }
-  
-  
-  public void saveOrUpdateEquipAlarm(List<EquipmentSpecificAlarm> equipmentSpecificAlarms){
-	  if(ObjectUtils.isEmpty(equipmentSpecificAlarms)){
-		  return;
+	public List<EquipmentSpecificAlarmLog> saveOrUpdateEquipAlarm(
+			List<EquipmentSpecificAlarm> equipmentSpecificAlarms) {
+		List<EquipmentSpecificAlarmLog> equipmentAlarmLogs = new ArrayList<>();
+		if (ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
+			return equipmentAlarmLogs;
 		}
 		equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
-	  List<EquipmentSpecificAlarmLog> equipmentAlarmLogs = new ArrayList<>();
-	  equipmentSpecificAlarms.forEach(action->{
+		equipmentSpecificAlarms.forEach(action -> {
 			if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
 				equipmentAlarmLogs.add(addEquipAlarmLogRecord(action));
 				if (ValidationUtil.isEmpty(action.getAlamContent())) {
@@ -721,16 +733,24 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 				}
 				mqttSendGateway.sendToMqtt(TopicEnum.EQDQR.getTopic(), JSONArray.toJSON(action).toString());
 			} else {
-        	  equipmentAlarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), equipmentSpecificAlarmLogService));
+				equipmentAlarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(),
+						equipmentSpecificAlarmLogService));
 				mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
 				bool = Boolean.TRUE;
 			}
-		  
 		});
+		return equipmentAlarmLogs;
+	}
 
-	  TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
-          @Override
-          public void afterCommit() {
+	/**
+	 * 报警日志同步其他系统
+	 * 
+	 * @param equipmentAlarmLogs
+	 */
+	public void equipmentAlarmLogsToOtherSystems(List<EquipmentSpecificAlarmLog> equipmentAlarmLogs) {
+		if (ObjectUtils.isEmpty(equipmentAlarmLogs)) {
+			return;
+		}
 		JSONObject jsonObject = new JSONObject();
 		jsonObject.put("seqNo", UUID.randomUUID().toString().replace("-", "").toLowerCase());
 		mqttSendGateway.sendToMqtt(TopicEnum.ALARM_LOG_INSERT.getTopic(), jsonObject.toString());
@@ -739,7 +759,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		if (syncSwitch) {
 			List<FireEquipmentFireAlarm> alarmList = createFireEquipmentFireAlarmVo(equipmentAlarmLogs);
 			if (!CollectionUtils.isEmpty(alarmList)) {
-                      Map<String, List<FireEquipmentFireAlarm>> collect = alarmList.stream().collect(Collectors.groupingBy(FireEquipmentFireAlarm::getType));
+				Map<String, List<FireEquipmentFireAlarm>> collect = alarmList.stream()
+						.collect(Collectors.groupingBy(FireEquipmentFireAlarm::getType));
 				for (String key : collect.keySet()) {
 					List<FireEquipmentFireAlarm> list = collect.get(key);
 					if (!CollectionUtils.isEmpty(list)) {
@@ -765,24 +786,23 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 			}
 
 			// 向南瑞平台推送报警消息
-                  if(!bool){
+			if (!bool) {
 				syncDataService.syncCreatedSendAlarm(equipmentAlarmLogs);
 			}
 		}
 	}
-      });
-	 
-  }
 
 	/**
 	 * 组装数字换流站平台告警数据
+	 * 
 	 * @param
 	 * @return
 	 */
-  private List<FireEquipmentFireAlarm> createFireEquipmentFireAlarmVo(List<EquipmentSpecificAlarmLog> equipmentAlarmLogs){
-	  Map<String,String> stationInfo = equipmentSpecificMapper.getStationInfo().get(0);
+	private List<FireEquipmentFireAlarm> createFireEquipmentFireAlarmVo(
+			List<EquipmentSpecificAlarmLog> equipmentAlarmLogs) {
+		Map<String, String> stationInfo = equipmentSpecificMapper.getStationInfo().get(0);
 		List<FireEquipmentFireAlarm> alarmList = new ArrayList<>();
-	  equipmentAlarmLogs.forEach(action->{
+		equipmentAlarmLogs.forEach(action -> {
 			FireEquipmentFireAlarm alarm = new FireEquipmentFireAlarm();
 			BeanUtils.copyProperties(action, alarm);
 			alarm.setAliasname(StringUtil.toNotEmptyString(action.getEquipmentSpecificIndexName()));
@@ -808,13 +828,14 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 	/**
 	 * 高斯库同步指标修改
+	 * 
 	 * @param equipmentSpecificIndexList
 	 */
-  private void syncSpecificIndexsToGS(List<EquipmentSpecificIndex> equipmentSpecificIndexList){
+	private void syncSpecificIndexsToGS(List<EquipmentSpecificIndex> equipmentSpecificIndexList) {
 		if (!ObjectUtils.isEmpty(equipmentSpecificIndexList) && syncSwitch) {
 			// 数据同步
 			List<EquipmentIndexVO> fireEquipMeasurementCollect = new ArrayList<>();
-		  equipmentSpecificIndexList.forEach(action->{
+			equipmentSpecificIndexList.forEach(action -> {
 				EquipmentIndexVO equipmentIndexVO = new EquipmentIndexVO();
 				BeanUtils.copyProperties(action, equipmentIndexVO);
 				fireEquipMeasurementCollect.add(equipmentIndexVO);
@@ -825,8 +846,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		}
 	}
 
-  
-  private IndexStateVo createIndexStateVo(EquipmentSpecificIndex equipmentSpecificIndex){
+	private IndexStateVo createIndexStateVo(EquipmentSpecificIndex equipmentSpecificIndex) {
 		IndexStateVo indexStateVo = new IndexStateVo();
 		BeanUtils.copyProperties(equipmentSpecificIndex, indexStateVo);
 		indexStateVo.setId(equipmentSpecificIndex.getIotCode() + "_" + equipmentSpecificIndex.getNameKey());
@@ -835,15 +855,15 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		return indexStateVo;
 	}
 
-  
 	public List<EquipmentSpecificAlarm> createIndexAlarmRecord(EquipmentSpecificIndex equipmentSpcIndex) {
 		// 处理火眼视频异常
 		List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
 		EquipmentSpecificAlarm equipmentSpecificAlarm = new EquipmentSpecificAlarm();
 		equipmentSpecificAlarm.setSystemIds(equipmentSpcIndex.getSystemId());
 		equipmentSpecificAlarm.setSystemCodes(this.getSystemCodeBySpeId(equipmentSpcIndex.getSystemId()));
-          List<EquipmentSpecificAlarm> indexAlarms = equipmentSpecificAlarmMapper.findEquipmentSpecificAlarmByEquipmentSpecificIdAndEquipmentIndexIdAndStatusIstrue(equipmentSpcIndex.getEquipmentSpecificId()
-        		  , equipmentSpcIndex.getEquipmentIndexId());
+		List<EquipmentSpecificAlarm> indexAlarms = equipmentSpecificAlarmMapper
+				.findEquipmentSpecificAlarmByEquipmentSpecificIdAndEquipmentIndexIdAndStatusIstrue(
+						equipmentSpcIndex.getEquipmentSpecificId(), equipmentSpcIndex.getEquipmentIndexId());
 		// NB设备告警
 		if (verifyNB(equipmentSpcIndex.getNameKey())) {
 			return getNbEquipAlarmList(indexAlarms, equipmentSpcIndex, equipmentSpecificAlarm);
@@ -871,6 +891,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 				action.setEquipmentCode(equipmentSpcIndex.getEquipmentCode());
 				// 冗余字段，alarm_log表更新时使用
 				action.setEquipmentSpecificCode(equipmentSpcIndex.getEquipmentSpecificCode());
+				action.setBuildId(equipmentSpcIndex.getBuildId());
 				equipmentSpecificAlarms.add(action);
 			});
 		}
@@ -878,21 +899,20 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		return equipmentSpecificAlarms;
 	}
 
- 
 	/**
 	 * 发送数据至换流站
+	 * 
 	 * @param
 	 */
-  private void sendEquipSpecIndexToAutosysTopic(EquipmentSpecificIndex equipmentSpeIndex){
-	  
-      TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
-          @Override
-          public void afterCommit() {
+	public void sendEquipSpecIndexToAutosysTopic(List<EquipmentSpecificIndex> equipmentSpeIndexs) {
+		equipmentSpeIndexs.forEach(equipmentSpeIndex -> {
 			String topic = "";
-              if (TrueOrFalseEnum.real.value.equals(equipmentSpeIndex.getValue()) && EquipmentRiskTypeEnum.HZGJ.getCode().equals(equipmentSpeIndex.getTypeCode())) {
+			if (TrueOrFalseEnum.real.value.equals(equipmentSpeIndex.getValue())
+					&& EquipmentRiskTypeEnum.HZGJ.getCode().equals(equipmentSpeIndex.getTypeCode())) {
 				equipmentSpeIndex.setType(EquipmentRiskTypeEnum.HZGJ.getCode());
 				topic = String.format("%s.%s%s", serverName, "equipment/", RiskLeverForAutoSys.BJ.getCode());
-              } else if (TrueOrFalseEnum.real.value.equals(equipmentSpeIndex.getValue()) && EquipmentRiskTypeEnum.GZ.getCode().equals(equipmentSpeIndex.getTypeCode())) {
+			} else if (TrueOrFalseEnum.real.value.equals(equipmentSpeIndex.getValue())
+					&& EquipmentRiskTypeEnum.GZ.getCode().equals(equipmentSpeIndex.getTypeCode())) {
 				equipmentSpeIndex.setType(EquipmentRiskTypeEnum.GZ.getCode());
 				topic = String.format("%s.%s%s", serverName, "equipment/", RiskLeverForAutoSys.GZ.getCode());
 			} else {
@@ -904,37 +924,32 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 			topicEntityVo.setTopic(topic);
 			topicEntityVo.setMessage(JSON.toJSONString(equipmentSpeIndex));
 			mqttSendGateway.sendToMqtt(topic, JSON.toJSONString(topicEntityVo));
-          }
 		});
 	}
 
-  
 	/**
-   * 组态大屏消息推送，设备表实时指标修改
+	 * 组态大屏消息推送
+	 * 
 	 * @param equipmentSpecificIndexList
 	 * @param topicEntity
 	 */
-  public void intePageSysDataRefresh(List<EquipmentSpecificIndex> equipmentSpecificIndexList, TopicEntityVo topicEntity){
-	  //TODO 数字化换流站组态屏数据推送,需要在事务提交之后，否侧事务隔离查询不出数据
-      TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
-          @Override
-          public void afterCommit() {
+	public void intePageSysDataRefresh(List<EquipmentSpecificIndex> equipmentSpecificIndexList,
+			TopicEntityVo topicEntity) {
 		mqttSendGateway.sendToMqtt(TopicEnum.EQXXTJ.getTopic(), "");
 		iEquipmentSpecificSerivce.integrationPageSysDataRefresh(topicEntity.getCode());
-              iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpecificIndexList);
-          }
-      });
 	}
 
 	/**
 	 * 更新数据报表表
+	 * 
 	 * @param equipmentSpecificIndex
 	 */
-  private void saveEquipmentAlarmReportDay(EquipmentSpecificIndex equipmentSpecificIndex){
+	private void saveEquipmentAlarmReportDay(EquipmentSpecificIndex equipmentSpecificIndex) {
 		SimpleDateFormat sdf = new SimpleDateFormat(DateUtils.DATE_PATTERN);
 		EquipmentAlarmReportDay equipmentAlarmReportDay = addEquipAlarmReportRecord(equipmentSpecificIndex);
 		LambdaQueryWrapper<EquipmentAlarmReportDay> wrapper = new LambdaQueryWrapper<>();
-      wrapper.eq(EquipmentAlarmReportDay::getReportDate, sdf.format(equipmentAlarmReportDay.getReportDate())).eq(EquipmentAlarmReportDay::getIndexId, equipmentAlarmReportDay.getIndexId())
+		wrapper.eq(EquipmentAlarmReportDay::getReportDate, sdf.format(equipmentAlarmReportDay.getReportDate()))
+				.eq(EquipmentAlarmReportDay::getIndexId, equipmentAlarmReportDay.getIndexId())
 				.eq(EquipmentAlarmReportDay::getEquipmentSpecificId, equipmentAlarmReportDay.getEquipmentSpecificId());
 		List<EquipmentAlarmReportDay> reportDayList = iEquipmentAlarmReportDayService.list(wrapper);
 		if (reportDayList.isEmpty()) {
@@ -946,14 +961,15 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 			reportDay.setLastReportDate(new Date());
 			reportDay.setValue(equipmentAlarmReportDay.getValue());
 			reportDay.setFrequency(reportDay.getFrequency() + 1);
-          reportDay.setIndexTrueNum(reportDay.getIndexTrueNum() == null ? equipmentAlarmReportDay.getIndexTrueNum() : reportDay.getIndexTrueNum() + equipmentAlarmReportDay.getIndexTrueNum());
+			reportDay.setIndexTrueNum(reportDay.getIndexTrueNum() == null ? equipmentAlarmReportDay.getIndexTrueNum()
+					: reportDay.getIndexTrueNum() + equipmentAlarmReportDay.getIndexTrueNum());
 			iEquipmentAlarmReportDayService.updateById(reportDay);
 		}
 
 	}
 
-  private String valueTranslate(String value, String enumStr){
-  	if(ObjectUtils.isEmpty(enumStr)){
+	private String valueTranslate(String value, String enumStr) {
+		if (ObjectUtils.isEmpty(enumStr)) {
 			return "";
 		}
 		try {
@@ -970,51 +986,14 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		return "";
 	}
 
-  private void carRealTimeDate(List<IotDataVO> iotDatalist, List<CarProperty> carProperties){
-	  
-	  
-	  List<CarProperty> carIndexsList = new ArrayList<>();
-	  iotDatalist.forEach(iotDataVO->{
-		// 对指标key为labels的数据处理
-	      if (EquipmentIndexLabelsEnum.labels.name.equals(iotDataVO.getKey())) {
-	          StringBuilder sb = new StringBuilder("equipmentOnCar_");
-	          EquipmentIndexLabelsVo labelsVo = new EquipmentIndexLabelsVo();
-	          Object obj = iotDataVO.getValue();
-	          if (obj instanceof JSONArray) {
-	              List<String> labelList = (List<String>) obj;
-	              labelList.forEach(code -> {
-	                  String key = sb.append(code).toString();
-	                  labelsVo.setEquipmentIotCode(code);
-	                  labelsVo.setTime(new Date());
-	                  redisUtils.set(key, com.alibaba.fastjson.JSONObject.toJSONString(labelsVo), redisExpireTime);
-	              });
-	          }
-	      }
-	      List<CarPropertyVo> carPropertyVos = new ArrayList<>();
-	      carProperties.forEach(carProperty->{
-              if (iotDataVO.getKey().equals(carProperty.getNameKey())) {
-                  carProperty.setValue(iotDataVO.getValue().toString());
-                  carProperty.setUpdateDate(new Date());
-                  carPropertyVos.add(carPropertyToCarPropertyVo(carProperty));
-                  carIndexsList.add(carProperty);
-              }
-	      });
-	      
-	      boolean updateBatchById = carPropertyService.updateBatchById(carIndexsList);
-	      
-	      if(updateBatchById){
-	    	  carTransactionSynch(carProperties,carPropertyVos);
-	      }
-	  });
-  }
-  
 	/**
 	 * 车辆数据推送及同步
+	 * 
 	 * @param carProperties
 	 * @param carPropertyVos
 	 */
-  public void carTransactionSynch(List<CarProperty> carProperties, List<CarPropertyVo> carPropertyVos){
-	  //TODO 数字化换流站组态屏数据推送,需要在事务提交之后，否侧事务隔离查询不出数据
+	public void carTransactionSynch(List<CarProperty> carProperties, List<CarPropertyVo> carPropertyVos) {
+		// TODO 数字化换流站组态屏数据推送,需要在事务提交之后，否侧事务隔离查询不出数据
 		TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
 			@Override
 			public void afterCommit() {
@@ -1032,7 +1011,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		});
 	}
 
-  private CarPropertyVo carPropertyToCarPropertyVo(CarProperty property){
+	private CarPropertyVo carPropertyToCarPropertyVo(CarProperty property) {
 		CarPropertyVo carPropertyVo = new CarPropertyVo();
 		carPropertyVo.setCarId(property.getCarId());
 		carPropertyVo.setCreateDate(property.getCreateDate());
@@ -1047,18 +1026,18 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 		return carPropertyVo;
 	}
 
-  private List<CarIndexGisVo> createCarIndexGisVo(List<CarProperty> carProperties){
+	private List<CarIndexGisVo> createCarIndexGisVo(List<CarProperty> carProperties) {
 		List<CarIndexGisVo> list = new ArrayList<>();
-	  long id =0l;
+		long id = 0l;
 		String iotCode = "";
-	  for(CarProperty action : carProperties){
+		for (CarProperty action : carProperties) {
 			CarIndexGisVo v = new CarIndexGisVo();
 			id = action.getCarId();
 			iotCode = action.getIotCode();
 			v.setId(action.getCarId());
 			v.setIotCode(action.getIotCode());
 			v.setNameKey(action.getEquipmentIndexKey());
-		  v.setValue(ObjectUtils.isEmpty(action.getValue())?"0":action.getValue());
+			v.setValue(ObjectUtils.isEmpty(action.getValue()) ? "0" : action.getValue());
 			list.add(v);
 		}
 
@@ -1074,14 +1053,17 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 	/**
 	 * //若为物联设备，则更新拓扑节点数据及告警状态
+	 * 
 	 * @param indexList
 	 */
-  public void updateNodeDateByEquipId(List<EquipmentSpecificIndex> indexList){
+	public void updateNodeDateByEquipId(List<EquipmentSpecificIndex> indexList) {
 		if (!ObjectUtils.isEmpty(indexList)) {
 			EquipmentVo equipmentVo = equipmentService.getEquipBySpecific(indexList.get(0).getEquipmentSpecificId());
 			if (equipmentVo.getIsIot().equals("1")) {
-	        List<EquipmentSpecificAlarm> alarmList = equipmentSpecificAlarmService.getEquipListBySpecific(true, indexList.get(0).getEquipmentSpecificId());
-	        topographyService.updateNodeDateByEquipId(indexList.get(0).getEquipmentSpecificId(), indexList, alarmList);
+				List<EquipmentSpecificAlarm> alarmList = equipmentSpecificAlarmService.getEquipListBySpecific(true,
+						indexList.get(0).getEquipmentSpecificId());
+				topographyService.updateNodeDateByEquipId(indexList.get(0).getEquipmentSpecificId(), indexList,
+						alarmList);
 			}
 		}
 	}