You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
82 lines
3.0 KiB
Java
82 lines
3.0 KiB
Java
package com.xydl.service.impl;
|
|
|
|
|
|
import com.xydl.mapper.OperationDB;
|
|
import com.xydl.util.FormatUtil;
|
|
import com.xydl.util.MqttUtil;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
import org.springframework.stereotype.Service;
|
|
|
|
import java.util.*;
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
@Service
|
|
@Slf4j
|
|
public class MqttServiceImpl {
|
|
|
|
@Autowired
|
|
OperationDB operationDBMapper;
|
|
|
|
@Autowired
|
|
MqttUtil mqttUtil;
|
|
|
|
|
|
public void reportRecord() {
|
|
List<String> allTableNames = operationDBMapper.getAllTable();
|
|
for (String tableName : allTableNames) {
|
|
processOneTable(tableName);
|
|
}
|
|
}
|
|
|
|
//单个表数据发送
|
|
public void processOneTable(String tableName) {
|
|
log.info("tableName:{}", tableName);
|
|
Map<String, String> fieldMap = new HashMap<>();
|
|
List<Map<String, String>> fieldMaps = operationDBMapper.getFieldMap(tableName);
|
|
for (Map<String, String> map : fieldMaps) {
|
|
for (String key : map.keySet()) {
|
|
fieldMap.put(map.get("field_name"), map.get("dest_field_name"));
|
|
}
|
|
}
|
|
String sqlExecuting = operationDBMapper.getSQL(tableName);
|
|
List<Integer> dataEqmids = operationDBMapper.getDataEqmids(tableName);
|
|
Map<Integer, Object> devIDLastTimeMap = new HashMap<>();
|
|
List<Map<String, Object>> devIDLastTimeMaps = operationDBMapper.getDeviceIDAndtime(tableName);
|
|
for (Map<String, Object> map : devIDLastTimeMaps) {
|
|
for (String devId : map.keySet()) {
|
|
devIDLastTimeMap.put(Integer.parseInt((String) map.get("devid_val")), map.get("field_val2"));
|
|
}
|
|
}
|
|
for (int deviceID : dataEqmids) {
|
|
Object time = devIDLastTimeMap.get(deviceID);
|
|
if (time == null) {
|
|
time = "2000-01-01 01:00:00";
|
|
operationDBMapper.addEarliestTime("10", tableName, String.valueOf(deviceID), time.toString());
|
|
}
|
|
publishData(deviceID, time.toString(), sqlExecuting, tableName, fieldMap);
|
|
}
|
|
}
|
|
|
|
//推送单个设备数据
|
|
public void publishData(int deviceID, String time, String sqlExecuting, String tableName, Map<String, String> fieldMap) {
|
|
|
|
String devIdSQL = sqlExecuting.replace("%%DEVID%%", String.valueOf(deviceID));
|
|
String newSQL = devIdSQL.replace("%%KEYVALUE%%", "'" + time + "'");
|
|
List<Map<String, Object>> dataOfoneDeviceID = operationDBMapper.getData(newSQL);
|
|
String jsonStringData = FormatUtil.mqttFormatTransform(dataOfoneDeviceID, fieldMap);
|
|
try {
|
|
if (mqttUtil.publish2MQTT(jsonStringData)) {
|
|
operationDBMapper.updateSyncRecordsTable(tableName, deviceID, time);
|
|
log.debug("表{}设备{}推送成功:{}", tableName, deviceID, jsonStringData);
|
|
}
|
|
} catch (Exception e) {
|
|
log.error("表{}设备{}推送异常：{}", tableName, deviceID, e.getMessage());
|
|
}
|
|
}
|
|
|
|
|
|
}
|
|
|