cac
/
mqtt
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

.

Browse Source
main
wenhua.zhou 2 years ago
parent 17d5f8b42b
commit 99f7f51be2
3 changed files with 57 additions and 44 deletions
Show Stats Download Patch File Download Diff File

39
src/main/java/com/xydl/service/impl/MqttServiceImpl.java
Unescape Escape View File

@ -1,6 +1,7 @@
package com.xydl.service.impl;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.xydl.mapper.OperationDB;
import com.xydl.util.DataSourceUtils;
import com.xydl.util.FormatUtil;
@ -14,6 +15,7 @@ import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;
import com.fasterxml.jackson.databind.ObjectMapper;
@Service
@ -149,7 +151,7 @@ public class MqttServiceImpl {
public List<Map<String,Object>> getData(String sqlExecuting, String deviceId, String timeStamp) {
public List<Map<String,Object>> getData(String sqlExecuting, String deviceId, String time) {
Connection conn = null;
PreparedStatement pstmt = null;
ResultSet rs = null;
@ -159,14 +161,14 @@ public class MqttServiceImpl {
String sql = sqlExecuting;
pstmt = conn.prepareStatement(sql);
pstmt.setString(1, deviceId);
pstmt.setString(2,timeStamp);
pstmt.setString(2, time);
rs = pstmt.executeQuery();
int columnCount = rs.getMetaData().getColumnCount(); //获取列的数量
while(rs.next()){
Map<String,Object> record = new HashMap<>();
for (int col = 0; col < columnCount; col++) {
String columnName = rs.getMetaData().getColumnName(col + 1);
String columnValue = rs.getString(columnName);
Object columnValue = rs.getString(columnName);
record.put(columnName,columnValue);
}
records.add(record);
@ -249,27 +251,6 @@ public class MqttServiceImpl {
}
private List<Map<String,Object>> transformList(Map<String,String> fieldMap, List<Map<String,Object>> deviceIDData) {
List<Map<String,Object>> newDeviceIDData = new ArrayList<>();
for(Map<String,Object> fieldValueMap : deviceIDData){
newDeviceIDData.add(transformOneRecord(fieldMap,fieldValueMap));
}
return newDeviceIDData;
}
private Map<String,Object> transformOneRecord(Map<String,String> fieldMap, Map<String,Object> fieldValueMap) {
Map<String,Object> newFieldValueMap = new HashMap<>();
for(String field : fieldMap.keySet()){
if(fieldValueMap.containsKey(field)){
newFieldValueMap.put(fieldMap.get(field),fieldValueMap.get(field) );
}
}
return newFieldValueMap;
}
public boolean updateSyncRecordsTable(String tableName, String deviceID, String time) {
Connection conn = null;
PreparedStatement pstmt = null;
@ -297,16 +278,14 @@ public class MqttServiceImpl {
logger.info("开始执行");
List<String> allTableNames = getAllTableNameFromSyncTable();
for(String tableName : allTableNames){
Map<String,String> fieldMap = getFieldMap(tableName);
String sqlExecuting = getSQL(tableName);
Map<String,Object> devIDLastTimeMap = getDeviceIDAndtime(tableName);
for(String deviceID : devIDLastTimeMap.keySet()){
List<Map<String,Object>> dataOfoneDeviceID = getData(sqlExecuting,deviceID, (String) devIDLastTimeMap.get(deviceID));
List<Map<String,Object>> newDataOfoneDeviceID = transformList(fieldMap,dataOfoneDeviceID);
String jsonStringData = FormatUtil.list2Json(newDataOfoneDeviceID);
String jsonStringData = FormatUtil.mqttFormatTransform(dataOfoneDeviceID,fieldMap);
logger.info("表{}设备{}推送数据：{}",tableName,deviceID,jsonStringData);
if(MqttUtil.publish2MQTT(jsonStringData)){
updateSyncRecordsTable(tableName,deviceID, (String) devIDLastTimeMap.get(deviceID));
@ -340,9 +319,7 @@ public class MqttServiceImpl {
}
}
}
}

57
src/main/java/com/xydl/util/FormatUtil.java
Unescape Escape View File

@ -3,21 +3,19 @@ package com.xydl.util;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.sql.Timestamp;
import java.util.*;
import java.util.stream.Collectors;
public class FormatUtil {
public static String list2Json(List list){
public static String mqttFormatTransform(List<Map<String,Object>> list, Map<String,String> fieldMap){
ObjectMapper objectMapper = new ObjectMapper();
List<Map<String,Object>> assetList = new ArrayList<>();
assetList.add(new HashMap<String,Object>(){{
put("AssetCode","ironCore");
put("AttributeList",list);
put("Timestamp", new Date().getTime());
}});
Map<String,Object> resultMap = new HashMap<String,Object>(){{
put("AssetList",assetList);
put("AssetList",list.stream().map(e -> recordTransform(e,fieldMap)).collect(Collectors.toList()));
}};
String jsonString = null;
try {
jsonString = objectMapper.writeValueAsString(resultMap);
@ -27,9 +25,48 @@ public class FormatUtil {
return jsonString;
}
public static void main(String[] args) {
System.out.println(new Date().getTime());
public static Map<String,Object> recordTransform(Map<String, Object> record, Map<String, String> fieldMap){
Map<String,Object> mqttRecord = new TreeMap<>();
List<Map<String,Object>> attribuiteList = new ArrayList<>();
for(String key : record.keySet()){
if(fieldMap.containsKey(key)){
attribuiteList.add(new HashMap<String,Object>(){{
put("AttributeCode", fieldMap.get(key));
put("DataValue",record.get(key));
}});
}
}
mqttRecord.put("AssetCode",record.get("sensorid"));
mqttRecord.put("AttributeList",attribuiteList);
String captureTime = record.get("d_time") !=null ? (String) record.get("d_time") : (String) record.get("capturetime");
mqttRecord.put("Timestamp",Timestamp.valueOf(captureTime).getTime());
return mqttRecord;
}
// private static List<Map<String,Object>> transformFieldForList(Map<String, String> fieldMap, List<Map<String, Object>> oriFieldList) {
// List<Map<String,Object>> newDeviceIDData = new ArrayList<>();
// for(Map<String,Object> fieldValueMap : oriFieldList){
// newDeviceIDData.add(transformOneRecord(fieldMap,fieldValueMap));
// }
// return newDeviceIDData;
// }
//
// private static Map<String,Object> transformOneRecord(Map<String, String> fieldMap, Map<String, Object> fieldValueMap) {
// Map<String,Object> newFieldValueMap = new HashMap<>();
// for(String field : fieldMap.keySet()){
// if(fieldValueMap.containsKey(field)){
// newFieldValueMap.put(fieldMap.get(field),fieldValueMap.get(field) );
// }
// }
// return newFieldValueMap;
// }
// public static void main(String[] args) {
// System.out.println(Timestamp.valueOf("2022-05-01 12:11:00").getTime());
// System.out.println(new Timestamp(new Date(Timestamp.valueOf("2022-05-01 12:11:00").getTime()).getTime()));
// }
}

5
src/main/resources/sql/mqtt-schema.sql
Unescape Escape View File

@ -20,7 +20,6 @@ SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for sync_tables_info
-- ----------------------------
DROP TABLE IF EXISTS `sync_tables_info`;
CREATE TABLE `sync_tables_info` (
`client_id` int UNSIGNED NOT NULL,
`table_name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL DEFAULT '',
@ -33,7 +32,7 @@ CREATE TABLE `sync_tables_info` (
PRIMARY KEY (`client_id`, `table_name`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
DROP TABLE IF EXISTS `sync_records`;
CREATE TABLE `sync_records` (
`client_id` int UNSIGNED NOT NULL,
`table_name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL DEFAULT '',
@ -44,7 +43,7 @@ CREATE TABLE `sync_records` (
PRIMARY KEY (`client_id`, `table_name`, `devid_val`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
DROP TABLE IF EXISTS `sync_fields_info`;
CREATE TABLE `sync_fields_info` (
`id` int UNSIGNED NOT NULL AUTO_INCREMENT,
`client_id` int UNSIGNED NOT NULL,

Write Preview
Loading…
Cancel
Save