修改回调

zoniot-water/zoniot-water-api/src/main/java/com/bz/zoneiot/water/api/dto/DeviceDataDto.java

@@ -2,6 +2,7 @@ package com.bz.zoneiot.water.api.dto;
 
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
+import lombok.ToString;
 
 import java.io.Serializable;
 import java.util.Date;
@@ -15,6 +16,7 @@ import java.util.List;
  * @Version 1.0
  */
 @Data
+@ToString
 public class DeviceDataDto implements Serializable {
 
     @ApiModelProperty(value = "设备ID")

zoniot-water/zoniot-water-core/src/main/java/com/bz/zoneiot/water/core/dao/MonitorDataMapper.java

@@ -24,6 +24,16 @@ public interface MonitorDataMapper {
     **/
     MonitorDataEntity getDeviceMonitorInfoByDeviceCode(@Param("deviceCode") String deviceCode);
 
+
+    /**
+    * 根据物联网设备ID查找
+    * @author Andy
+    * @date 15:33 2021/11/5
+    * @param deviceId:
+    * @return com.bz.zoneiot.water.core.entity.MonitorDataEntity
+    **/
+    MonitorDataEntity getDeviceMonitorInfoByIotDeviceId(@Param("iotDeviceId") Long deviceId);
+
     /**
     * 批量插入
     * @author Andy

zoniot-water/zoniot-water-core/src/main/java/com/bz/zoneiot/water/core/mapper/AlarmTypeDetailMapper.xml

@@ -120,7 +120,7 @@
     </select>
 
     <update id="updateStateByDeviceId">
-        update sms_alarm_details set state =#{state} where device_id = #{deviceId}
+        update sms_alarm_details set state =#{state}, ALARM_END_TIME = now(), DATE_UPDATE = now() where device_id = #{deviceId}
     </update>
 
 </mapper>

zoniot-water/zoniot-water-core/src/main/java/com/bz/zoneiot/water/core/mapper/DeviceMapper.xml

@@ -513,7 +513,7 @@
     </update>
 
     <select id="findDeviceIdByIotDeviceId" resultType="java.lang.Long">
-        select DEVICE_ID from sms_device where IOT_DEVICE_ID = #{deviceId}
+        select id  from sms_device where IOT_DEVICE_ID = #{deviceId}
     </select>
 
 

zoniot-water/zoniot-water-core/src/main/java/com/bz/zoneiot/water/core/mapper/MonitorDataMapper.xml

@@ -14,6 +14,7 @@
             <result property="unit" column="unit" jdbcType="INTEGER"/>
         </collection>
     </resultMap>
+
     <select id="getDeviceMonitorInfoByDeviceCode" resultMap="DeviceMonitorInfo">
 
         select
@@ -32,6 +33,22 @@
 
     </select>
 
+    <select id="getDeviceMonitorInfoByIotDeviceId" resultMap="DeviceMonitorInfo">
+        select
+             d.TENANT_ID as "tenantId"
+            ,d.id as "deviceId"
+            ,d.DEVICE_NAME as "deviceName"
+            ,d.DEVICE_CODE as "deviceCode"
+            ,da.id as "attributeId"
+            ,da.NAME as "attributeName"
+            ,da.IDENTIFIER as "identifier"
+            ,da.UNIT as "unit"
+        from sms_device d
+        left join sms_device_type dt on dt.id=d.DEVICE_TYPE_ID and dt.STATUS = 1
+        left join sms_device_attribute da on dt.id=da.DEVICE_TYPE_ID and da.STATUS = 1
+        where d.STATUS =1 and d.IOT_DEVICE_ID=#{iotDeviceId}
+    </select>
+
     <insert id="batchInsertDayReport">
         INSERT INTO sms_day_report(
         TENANT_ID, YEAR, MONTH, DAY, HOUR, DEVICE_ID, DEVICE_NAME, DEVICE_CODE, ATTRIBUTE_ID,

zoniot-water/zoniot-water-web/src/main/java/com/bz/zoneiot/water/web/controller/NotifyController.java

@@ -1,6 +1,5 @@
 package com.bz.zoneiot.water.web.controller;
 
-import com.bz.zoneiot.core.common.exception.BusinessException;
 import com.bz.zoneiot.core.common.pojo.AjaxMessage;
 import com.bz.zoneiot.water.api.dto.AlarmDetailsAddDto;
 import com.bz.zoneiot.water.api.dto.DeviceDataDto;

zoniot-water/zoniot-water-web/src/main/java/com/bz/zoneiot/water/web/service/handle/DispatchReportHandle.java

@@ -68,7 +68,7 @@ public class DispatchReportHandle {
     @Async
     public void handler(MonitorDataEntity monitorDataEntity, JSONObject receiveData, Date receiveDateTime) {
         String tenantId  = monitorDataEntity.getTenantId();
-        List<String> paramTypeList = dispatchPlanClient.getWaterWorksParamType();
+        List<String> paramTypeList  = dispatchPlanClient.getWaterWorksParamType();
         Map<String,String> titleMap = dispatchPlanClient.getWaterWorksTitle();
         List<Integer> paramTypes   = paramTypeList.stream().map(Integer::parseInt).collect(Collectors.toList());
         for (MonitorDataValueEntity valueEntity : monitorDataEntity.getDataValues()) {
@@ -76,7 +76,8 @@ public class DispatchReportHandle {
                 continue;
             }
             // 获取该设备属性的参数类型
-            List<DeviceParmEntity> list = deviceParmMapper.selectByDeviceIdByAttributeIdByParamType(monitorDataEntity.getTenantId(), monitorDataEntity.getDeviceId(), valueEntity.getAttributeId(), paramTypes);
+            List<DeviceParmEntity> list = deviceParmMapper.selectByDeviceIdByAttributeIdByParamType(monitorDataEntity.getTenantId(),
+                    monitorDataEntity.getDeviceId(), valueEntity.getAttributeId(), paramTypes);
             if (CollectionUtils.isEmpty(list)){
                 continue;
             }
@@ -114,7 +115,7 @@ public class DispatchReportHandle {
         }
         // 存储该设备当前小时最早的数据
         redisUtils.set(md5Key, dispatchPlanDataEntity.getDataValue());
-        redisUtils.setExpire(md5Key, HOURS_2 );
+        redisUtils.setExpire(md5Key, HOURS_2);
         DispatchPlanReportEntity dispatchPlanReportEntity = new DispatchPlanReportEntity();
         BeanCopyUtils.copyProperties(dispatchPlanDataEntity, dispatchPlanReportEntity, DispatchPlanReportEntity.class);
         dispatchPlanReportEntity.setStatDay(DateUtil.getDate(new Date()));

zoniot-water/zoniot-water-web/src/main/java/com/bz/zoneiot/water/web/service/handle/ReceiveDeviceDataService.java

@@ -1,9 +1,12 @@
 package com.bz.zoneiot.water.web.service.handle;
 
 import com.alibaba.fastjson.JSONObject;
+import com.bz.zoneiot.core.mongo.utils.StringUtil;
 import com.bz.zoneiot.core.utils.ByteArrayUtils;
 import com.bz.zoneiot.core.utils.DateUtil;
 import com.bz.zoneiot.core.utils.util.RedisUtils;
+import com.bz.zoneiot.water.api.dto.DeviceDataDto;
+import com.bz.zoneiot.water.api.dto.TimeValueDto;
 import com.bz.zoneiot.water.core.dao.DeviceMapper;
 import com.bz.zoneiot.water.core.dao.MonitorDataMapper;
 import com.bz.zoneiot.water.core.entity.MonitorDataEntity;
@@ -17,9 +20,7 @@ import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
 import java.math.BigDecimal;
-import java.util.Calendar;
-import java.util.Date;
-import java.util.List;
+import java.util.*;
 import java.util.stream.Collectors;
 
 
@@ -79,6 +80,16 @@ public class ReceiveDeviceDataService {
         return monitorDataMapper.getDeviceMonitorInfoByDeviceCode(deviceCode);
     }
 
+    public MonitorDataEntity getDeviceMonitorInfoByIotDeviceId(Long iotDeviceId){
+        // 先取缓存里的数据
+        byte[] bytes = redisUtils.get(("sms_water_" + iotDeviceId).getBytes());
+        if(bytes != null && bytes.length > 0 ){
+            return (MonitorDataEntity)ByteArrayUtils.bytesToObject(bytes).get();
+        }
+        return monitorDataMapper.getDeviceMonitorInfoByIotDeviceId(iotDeviceId);
+    }
+
+
     public MonitorDataEntity save(MonitorDataEntity monitorDataEntity) {
         // 缓存数据
         redisUtils.setExpire(("sms_water_"+monitorDataEntity.getDeviceCode()).getBytes(), ByteArrayUtils.objectToBytes(monitorDataEntity).get());
@@ -88,38 +99,57 @@ public class ReceiveDeviceDataService {
         return mongoTemplate.save(monitorDataEntity);
     }
 
-    /***
+    /**
     * manufacturer 厂家，mode型号，设备编号unitIdentifier，type设备类型
     * @author Andy
-    * @date 17:26 2021/10/19
-    * @param jsonObject:
+    * @date 16:40 2021/11/5
+    * @param deviceDataDto:
     * @return void
     **/
-    public void receivedDataHandle(JSONObject jsonObject){
-        String eventTime      = jsonObject.getString("eventTime");
-        String deviceCode     = jsonObject.getString("deviceNo");
-        Date receiveDateTime  = parseDate(eventTime);
-        log.info("rabbitMq接收消息处理，code:{}，上报时间:{}", deviceCode, eventTime);
-        if (!checkParameter(jsonObject, deviceCode, receiveDateTime)){
+    public void receivedDataHandle(DeviceDataDto deviceDataDto){
+        String deviceCode     = deviceDataDto.getDeviceNo();
+        Date receiveDateTime  = deviceDataDto.getEventTime();
+        log.info("rabbitMq接收消息处理，code:{}，上报时间:{}", deviceCode, DateUtil.format(receiveDateTime, "yyyy-MM-dd HH:mm:ss"));
+
+        if (receiveDateTime == null || StringUtil.isEmpty(deviceCode) || CollectionUtils.isEmpty(deviceDataDto.getDecodeData())){
             return;
         }
         // 查询不到设备或者设备属性为空
-        MonitorDataEntity monitorDataEntity = getDeviceMonitorInfoByDeviceCode(deviceCode);
+        MonitorDataEntity monitorDataEntity = this.getDeviceMonitorInfoByIotDeviceId(deviceDataDto.getDeviceId());
         if (monitorDataEntity == null || CollectionUtils.isEmpty(monitorDataEntity.getDataValues())){
             log.error("rabbitMq接收消息处理，code:{}，查询不到设备或者设备属性为空，退出",deviceCode);
             return;
         }
-        JSONObject receiveData  = JSONObject.parseObject(jsonObject.getString("decodeData"));
-        List<MonitorDataValueEntity> monitorDataValueEntities = monitorDataEntity.getDataValues();
+        List<TimeValueDto> timeValueDtoList = deviceDataDto.getDecodeData();
+        timeValueDtoList.sort(Comparator.comparing(TimeValueDto::getTime));
+        Date lastTime = null;
+        for (TimeValueDto timeValueDto : timeValueDtoList) {
+            this.saveData(timeValueDto, monitorDataEntity);
+            lastTime = timeValueDto.getTime();
+        }
+        // 修改设备上报时间
+        deviceMapper.updateLastTime(monitorDataEntity.getDeviceId(), lastTime);
+    }
+
+    /**
+    * 保存数据
+    * @author Andy
+    * @date 16:36 2021/11/5
+    * @param timeValueDto:
+    * @param monitorDataEntity:
+    * @return void
+    **/
+    private void saveData(TimeValueDto timeValueDto, MonitorDataEntity monitorDataEntity){
         Integer number = 0;
-        for (MonitorDataValueEntity monitorDataValueEntity : monitorDataValueEntities) {
+        Map<String, Object> receiveData = timeValueDto.getValue();
+        for (MonitorDataValueEntity monitorDataValueEntity : monitorDataEntity.getDataValues()) {
             BigDecimal bigDecimal = null;
             try {
-                Double doubleValue = receiveData.getDouble(monitorDataValueEntity.getIdentifier());
-                if (doubleValue == null) {
+                Object value = receiveData.get(monitorDataValueEntity.getIdentifier());
+                if (value == null || !StringUtil.isNumber(value.toString())){
                     continue;
                 }
-                bigDecimal = new BigDecimal(doubleValue);
+                bigDecimal = new BigDecimal(value.toString());
             } catch (Exception e) {
                 log.error("double 转换 error ->", e);
                 continue;
@@ -136,52 +166,25 @@ public class ReceiveDeviceDataService {
         }
         // 没有匹配到属性，视为垃圾数据忽略
         if (number == 0) {
-            log.error("rabbitMq接收消息处理，code:{}，上报时间:{}，没有匹配到属性，视为垃圾数据忽略，退出",deviceCode, eventTime);
+            log.error("rabbitMq接收消息处理，code:{}，上报时间:{}，没有匹配到属性，视为垃圾数据忽略，退出",
+                    monitorDataEntity.getId(), DateUtil.format(timeValueDto.getTime(), "yyyy-MM-dd HH:mm:ss"));
             return;
         }
         // 保存到缓存和mongodb数据库
-        saveToCacheAndMongodb(monitorDataEntity, receiveDateTime);
-        // 修改设备上报时间
-        deviceMapper.updateLastTime(monitorDataEntity.getDeviceId(),receiveDateTime);
+        saveToCacheAndMongodb(monitorDataEntity, timeValueDto.getTime());
         // 调度预案处理
-        dispatchReportHandler.handler(monitorDataEntity, receiveData, receiveDateTime);
+        // dispatchReportHandler.handler(monitorDataEntity, receiveData, timeValueDto.getTime());
         // 水泵运行状态报表业务处理
-        reportWaterPumpStateHandler.handler(monitorDataEntity, receiveData, receiveDateTime);
-    }
-
-
-    private Date parseDate(String eventTime){
-        return StringUtils.isEmpty(eventTime) ? null : DateUtil.parseDate(eventTime, "yyyy-MM-dd HH:mm:ss");
+        reportWaterPumpStateHandler.handler(monitorDataEntity, timeValueDto, timeValueDto.getTime());
     }
 
 
-    /***
-     * 对象有时间、有设备编码、有数据则解析，任何一个位空直接退出
-     * @author lihui
-     * @date 10:53 2021/5/28
-     * @param jsonObject :
-     * @param deviceCode :
-     * @return boolean
-     **/
-    private boolean checkParameter(JSONObject jsonObject, String deviceCode, Date receiveDateTime){
-        if (!jsonObject.containsKey(EVENT_TIME_KEY) || !jsonObject.containsKey(UNIT_IDENTIFIER_KEY) ||
-                !jsonObject.containsKey(PARSED_DATA_KEY)) {
-            log.error("rabbitMq接收消息处理，code:{}，上报时间:{}，退出",deviceCode);
-            return false;
-        }
-        if (receiveDateTime == null) {
-            log.error("rabbitMq接收消息处理，code:{}，报警信息查询时间转换错误，退出", deviceCode);
-            return false;
-        }
-        return true;
-    }
-
     private void saveToCacheAndMongodb(MonitorDataEntity monitorDataEntity, Date receiveDateTime){
         monitorDataEntity.setCollectDate(receiveDateTime);
         Calendar cal = Calendar.getInstance();
         cal.setTime(receiveDateTime);
         monitorDataEntity.setYear(cal.get(Calendar.YEAR));
-        monitorDataEntity.setMonth(cal.get(Calendar.MONTH)+1);
+        monitorDataEntity.setMonth(cal.get(Calendar.MONTH) + 1);
         monitorDataEntity.setDay(cal.get(Calendar.DAY_OF_MONTH));
         monitorDataEntity.setHour(cal.get(Calendar.HOUR_OF_DAY));
         save(monitorDataEntity);

zoniot-water/zoniot-water-web/src/main/java/com/bz/zoneiot/water/web/service/handle/ReportWaterPumpStateHandle.java

@@ -1,8 +1,8 @@
 package com.bz.zoneiot.water.web.service.handle;
 
 import cn.hutool.core.collection.CollectionUtil;
-import com.alibaba.fastjson.JSONObject;
 import com.bz.zoneiot.core.utils.DateUtil;
+import com.bz.zoneiot.water.api.dto.TimeValueDto;
 import com.bz.zoneiot.water.api.dto.watepump.ReprotWaterPumpQueryDto;
 import com.bz.zoneiot.water.core.dao.DeviceAttributeSpecsMapper;
 import com.bz.zoneiot.water.core.dao.ReportWaterPumpStateMapper;
@@ -21,6 +21,7 @@ import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 /**
  * @ClassName ReportWaterPumpStateHandler
@@ -46,8 +47,8 @@ public class ReportWaterPumpStateHandle {
     private DeviceAttributeSpecsMapper deviceAttributeSpecsMapper;
 
     @Async
-    public void handler(MonitorDataEntity monitorDataEntity, JSONObject receiveData, Date receiveDateTime){
-
+    public void handler(MonitorDataEntity monitorDataEntity, TimeValueDto timeValueDto, Date receiveDateTime){
+        Map<String, Object> receiveData = timeValueDto.getValue();
         log.info("【rabbitMq处理报警水泵运行状态报表：{}】开始处理", monitorDataEntity.getDeviceCode());
         // 0 运行， 1停止
         int state = 0;
@@ -60,7 +61,8 @@ public class ReportWaterPumpStateHandle {
         // TODO 需要采用分布式锁解决，避免部署多台服务器
         synchronized (monitorDataEntity.getDeviceCode().intern()) {
             for (MonitorDataValueEntity valueEntity : monitorDataEntity.getDataValues()) {
-                if (!receiveData.containsKey(valueEntity.getIdentifier())){
+                Object value = receiveData.get(valueEntity.getIdentifier());
+                if (value == null){
                     continue;
                 }
                 entity.setAttributeId(valueEntity.getAttributeId());
@@ -73,7 +75,7 @@ public class ReportWaterPumpStateHandle {
                 if (CollectionUtil.isEmpty(listSate)) {
                     continue;
                 }
-                BigDecimal stateValue  = new BigDecimal(receiveData.getDouble(valueEntity.getIdentifier()));
+                BigDecimal stateValue  = new BigDecimal(value.toString());
                 state = listSate.contains(stateValue.toString()) ? 0 : 1;
                 queryDto.setAttributeId(valueEntity.getAttributeId());
                 addOrUpdate(queryDto.getMd5Query(), monitorDataEntity, valueEntity, insertList, state, receiveDateTime);

zoniot-water/zoniot-water-web/src/main/java/com/bz/zoneiot/water/web/service/impl/NotifyServiceImpl.java

@@ -39,25 +39,30 @@ public class NotifyServiceImpl implements NotifyService {
 
     @Override
     public Integer insertAlarmInfo(AlarmDetailsAddDto dto) {
-        log.info("【接收通知】物联告警信息：{}", dto.toString());
-        return alarmTypeDetailsService.insert(dto);
+        log.info("【接收通知】物联告警信息begin：{}", dto.toString());
+        Integer result = alarmTypeDetailsService.insert(dto);
+        log.info("【接收通知】物联告警信息end：{}", dto.toString());
+        return result;
     }
 
     @Override
     public void receiveDeviceData(DeviceDataDto deviceData) {
-        receiveDeviceDataService.receivedDataHandle(JSONObject.parseObject(JSON.toJSONString(deviceData)));
+        log.info("【接收通知】物联设备信息begin：{}", deviceData.toString());
+        receiveDeviceDataService.receivedDataHandle(deviceData);
+        log.info("【接收通知】物联设备信息end：{}", deviceData.toString());
     }
 
     @Override
     public void updateDeviceSate(DeviceStateDto dto) {
         // 修改设备状态
-        log.info("【接收通知】物联修改设备信息：{}", dto.toString());
+        log.info("【接收通知】物联修改设备状态信息begin：{}", dto.toString());
         deviceMapper.updateStateByIotDeviceId(dto.getDeviceId(), dto.getDeviceStatus());
         // 如果状态正常就要变更报警为正常
         if (!DeviceStatusEnum.offline(dto.getDeviceStatus())) {
             Long deviceId = deviceMapper.findDeviceIdByIotDeviceId(dto.getDeviceId());
             alarmTypeDetailsService.updateStateByDeviceId(deviceId, AlarmStateEnum.HISTORY_ALARM.getCode());
         }
+        log.info("【接收通知】物联修改设备状态信息end：{}", dto.toString());
     }
 
 }