refactor: merge previous session changes

Resolve merge state from prior work session.

iot-platform/src/main/java/com/iot/platform/service/SysrealtimeService.java

@@ -2,20 +2,58 @@ package com.iot.platform.service;
 
 import com.iot.platform.mapper.SysrealtimeMapper;
 import org.apache.ibatis.annotations.Param;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
 import java.util.List;
 
 @Service
 public class SysrealtimeService {
+    private static final Logger log = LoggerFactory.getLogger(SysrealtimeService.class);
 
     @Resource
     public SysrealtimeMapper sysrealtimeMapper;
 
-    public void createrealtime(String tableName){
-        sysrealtimeMapper.createrealtime(tableName);
+    public void createrealtime(String tableName) {
+        log.info("准备创建表: [{}]", tableName);
+        log.info("执行建表SQL前，先验证数据库连接...");
+
+        // 直接用JDBC测试连接和建表
+        try (Connection conn = getConnection()) {
+            String sql = "CREATE TABLE IF NOT EXISTS `" + tableName + "` (" +
+                    "create_time VARCHAR(255) NOT NULL COMMENT '时间戳'," +
+                    "device_id VARCHAR(255) NOT NULL COMMENT '设备id'," +
+                    "timestamp VARCHAR(255) NOT NULL COMMENT '时间戳'," +
+                    "k VARCHAR(255) NOT NULL COMMENT 'key'," +
+                    "v VARCHAR(255) NOT NULL COMMENT '值'," +
+                    "INDEX idx_device_id (device_id)," +
+                    "INDEX idx_device_create_time (device_id, create_time)," +
+                    "INDEX idx_k (k)" +
+                    ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时数据信息'";
+            log.info("SQL: {}", sql);
+            conn.createStatement().executeUpdate(sql);
+            log.info("表[{}]创建成功", tableName);
+        } catch (Exception e) {
+            log.error("JDBC建表[{}]失败: {}", tableName, e.getMessage());
+            Throwable cause = e;
+            while (cause.getCause() != null) cause = cause.getCause();
+            log.error("根因: {}", cause.getMessage());
+            throw new RuntimeException("建表失败", e);
+        }
+    }
+
+    private Connection getConnection() throws SQLException {
+        String url = "jdbc:mysql://47.104.204.180:3306/data?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8";
+        String user = "root";
+        String password = "Zhu059300()__";
+        return DriverManager.getConnection(url, user, password);
     }
+
     public List<String> selecttables(){
         return sysrealtimeMapper.selecttables();
     }

iot-platform/src/main/java/com/iot/platform/service/TDengineService.java

@@ -8,9 +8,13 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.io.ByteArrayOutputStream;
 import java.sql.*;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.zip.*;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 @Service
 public class TDengineService {
@@ -37,6 +41,12 @@ public class TDengineService {
     // === 新增：缓存超级表结构 (key = dbName.stableName) ===
     private final Map<String, Set<String>> stableColumnCache = new ConcurrentHashMap<>();
 
+    // JSON 列名，用于存储所有动态字段
+    private static final String JSON_COLUMN_NAME = "ext_data";
+
+    // ObjectMapper 线程安全，可复用
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
     public TDengineService() {
         // 延迟初始化：不在构造器中创建连接池，避免 TDengine 本地库缺失时阻断启动
     }
@@ -162,70 +172,6 @@ public class TDengineService {
     }
 
     // ==========================================
-    // 新增列（成功后更新缓存）
-    // ==========================================
-    private boolean addColumnToStable(String dbName, String stableName, String columnName) {
-        String sql = String.format("ALTER TABLE %s.%s ADD COLUMN %s VARCHAR(50)",
-                wrapName(dbName), wrapName(stableName), wrapName(columnName));
-
-        Connection conn = null;
-        Statement stmt = null;
-        try {
-            conn = getConnection();
-            stmt = conn.createStatement();
-            stmt.setQueryTimeout(10);
-            stmt.executeUpdate(sql);
-            log.info("✅ 新增列成功: {}.{}.{}", dbName, stableName, columnName);
-
-            // ✅ 更新缓存
-            String key = getStableKey(dbName, stableName);
-            stableColumnCache.computeIfPresent(key, (k, existingCols) -> {
-                Set<String> newCols = new HashSet<>(existingCols);
-                newCols.add(columnName);
-                return newCols;
-            });
-
-            return true;
-        } catch (SQLException e) {
-            if (e.getMessage().contains("already exists")) {
-                // 列已存在，确保缓存包含它（防御性）
-                String key = getStableKey(dbName, stableName);
-                stableColumnCache.computeIfPresent(key, (k, cols) -> {
-                    if (!cols.contains(columnName)) {
-                        Set<String> updated = new HashSet<>(cols);
-                        updated.add(columnName);
-                        return updated;
-                    }
-                    return cols;
-                });
-                return true;
-            }
-            if (e.getMessage().contains("Row length exceeds") || e.getMessage().contains("too many columns")) {
-                log.warn("⚠️ 跳过新增列 (超限): {}", columnName);
-                return false;
-            }
-            log.error("❌ 新增列失败: {}", e.getMessage());
-            return false;
-        } finally {
-            if (stmt != null) try { stmt.close(); } catch (SQLException ignored) {}
-            closeConnection(conn);
-        }
-    }
-
-//    private Map<String, String> filterEmptyFields(Map<String, Object> fieldValues) {
-//        Map<String, String> filtered = new HashMap<>();
-//        for (Map.Entry<String, Object> entry : fieldValues.entrySet()) {
-//            if (entry.getValue() != null) {
-//                String val = entry.getValue().toString().trim();
-//                if (!val.isEmpty()) {
-//                    filtered.put(entry.getKey(), escapeValue(val));
-//                }
-//            }
-//        }
-//        return filtered;
-//    }
-
-    // ==========================================
     // 初始化表结构
     // ==========================================
     private boolean initTableStructure(String dbName, String supertablename, String table, Set<String> fieldNames) {
@@ -238,20 +184,22 @@ public class TDengineService {
 
             stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS " + wrapName(dbName));
 
+            // 创建超级表：固定 ts + surfacename + ext_data(VARCHAR)
+            // 注：TDengine 2.x JSON 类型只能用于 TAGS，不能用于普通列
             String stableSql = String.format(
-                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64)) TAGS (location BINARY(64))",
+                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64), ext_data VARCHAR(8192)) TAGS (location BINARY(64))",
                     wrapName(dbName),
                     wrapName(supertablename)
             );
             stmt.executeUpdate(stableSql);
 
-            Set<String> existing = getStableColumns(dbName, supertablename);
-            for (String field : fieldNames) {
-                if ("ts".equalsIgnoreCase(field) || "surfacename".equalsIgnoreCase(field)) continue;
-                if (!existing.contains(field)) {
-                    addColumnToStable(dbName, supertablename, field);
-                }
-            }
+            // 更新缓存：固定列
+            String key = getStableKey(dbName, supertablename);
+            Set<String> fixedCols = new HashSet<>();
+            fixedCols.add("ts");
+            fixedCols.add("surfacename");
+            fixedCols.add("ext_data");
+            stableColumnCache.put(key, fixedCols);
 
             String tableSql = String.format(
                     "CREATE TABLE IF NOT EXISTS %s.%s USING %s.%s TAGS ('%s')",
@@ -282,44 +230,94 @@ public class TDengineService {
 
         String supertablename = table.contains("_") ? table.substring(0, table.lastIndexOf('_')) : table;
 
-        Set<String> allFields = new HashSet<>();
-        for (Map<String, Object> data : dataList) {
-            if (data != null) {
-                for (String key : data.keySet()) {
-                    if (!"topic".equalsIgnoreCase(key)) allFields.add(key);
-                }
+        // 初始化表结构（固定为 ts + surfacename + ext_data）
+        initTableStructure(dbName, supertablename, table, Collections.emptySet());
+
+        // 分批处理，每批最多 50 条记录（避免 SQL 过长）
+        int batchSize = 50;
+        int totalBatches = (dataList.size() + batchSize - 1) / batchSize;
+
+        for (int i = 0; i < totalBatches; i++) {
+            int start = i * batchSize;
+            int end = Math.min(start + batchSize, dataList.size());
+            List<Map<String, Object>> batch = dataList.subList(start, end);
+
+            if (!insertBatchInternal(dbName, supertablename, table, batch)) {
+                return false;
             }
         }
-        allFields.remove("ts");
-        allFields.remove("surfacename");
 
-        initTableStructure(dbName, supertablename, table, allFields);
+        log.info("✅ 批量写入成功: {} | 条数: {}", table, dataList.size());
+        return true;
+    }
+
+    /**
+     * 内部方法：插入一批数据
+     */
+    private boolean insertBatchInternal(String dbName, String supertablename, String table, List<Map<String, Object>> dataList) {
+        // 确保表存在（可能有竞态条件）
+        ensureTableExists(dbName, supertablename, table);
 
         StringBuilder sqlBuilder = new StringBuilder();
         sqlBuilder.append("INSERT INTO ").append(wrapName(dbName)).append(".").append(wrapName(table))
-                .append(" (ts, surfacename");
-
-        List<String> colList = new ArrayList<>();
-        for (String col : allFields) {
-            if (isValidFieldName(col)) {
-                colList.add(col);
-                sqlBuilder.append(", ").append(wrapName(col));
-            } else {
-                log.warn("跳过非法字段名: {}", col);
+                .append(" (ts, surfacename, ext_data) VALUES ");
+
+        boolean hasData = false;
+        for (Map<String, Object> data : dataList) {
+            if (data == null) continue;
+
+            // 将动态字段转为 JSON 并 GZIP 压缩
+            String extJson = buildDynamicJson(data);
+            String compressed = compressToBase64(extJson);
+
+            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("', '")
+                    .append(escapeValue(compressed)).append("'),");
+            hasData = true;
+        }
+
+        if (!hasData) return true;
+
+        sqlBuilder.setLength(sqlBuilder.length() - 1);
+        String finalSql = sqlBuilder.toString();
+
+        Connection conn = null;
+        Statement stmt = null;
+        try {
+            conn = getConnection();
+            stmt = conn.createStatement();
+            stmt.setQueryTimeout(30);
+            stmt.executeUpdate(finalSql);
+            return true;
+        } catch (SQLException e) {
+            // 表不存在时尝试重建表后重试
+            if (e.getMessage().contains("Table does not exist")) {
+                log.warn("⚠️ 表不存在，重建表: {}", table);
+                initTableStructure(dbName, supertablename, table, Collections.emptySet());
+                return insertBatchRetry(dbName, supertablename, table, dataList);
             }
+            log.error("❌ 批量写入 SQL 失败: {} | 错误: {}", table, e.getMessage());
+            return false;
+        } finally {
+            if (stmt != null) try { stmt.close(); } catch (SQLException ignored) {}
+            closeConnection(conn);
         }
-        sqlBuilder.append(") VALUES ");
+    }
+
+    /**
+     * 重试插入（表重建后）
+     */
+    private boolean insertBatchRetry(String dbName, String supertablename, String table, List<Map<String, Object>> dataList) {
+        StringBuilder sqlBuilder = new StringBuilder();
+        sqlBuilder.append("INSERT INTO ").append(wrapName(dbName)).append(".").append(wrapName(table))
+                .append(" (ts, surfacename, ext_data) VALUES ");
 
         boolean hasData = false;
         for (Map<String, Object> data : dataList) {
             if (data == null) continue;
-            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("'");
-            for (String col : colList) {
-                Object v = data.get(col);
-                String val = (v == null) ? "" : escapeValue(v.toString());
-                sqlBuilder.append(", '").append(val).append("'");
-            }
-            sqlBuilder.append("),");
+            String extJson = buildDynamicJson(data);
+            String compressed = compressToBase64(extJson);
+            sqlBuilder.append("(NOW(), '").append(escapeValue(supertablename)).append("', '")
+                    .append(escapeValue(compressed)).append("'),");
             hasData = true;
         }
 
@@ -335,17 +333,112 @@ public class TDengineService {
             stmt = conn.createStatement();
             stmt.setQueryTimeout(30);
             stmt.executeUpdate(finalSql);
-            log.info("✅ 批量写入成功: {} | 条数: {}", table, dataList.size());
             return true;
         } catch (SQLException e) {
-            log.error("❌ 批量写入 SQL 失败: {} | 错误: {}", table, e.getMessage());
-            throw e;
+            log.error("❌ 重试插入失败: {} | 错误: {}", table, e.getMessage());
+            return false;
         } finally {
             if (stmt != null) try { stmt.close(); } catch (SQLException ignored) {}
             closeConnection(conn);
         }
     }
 
+    /**
+     * 确保表存在（检查+创建）
+     */
+    private void ensureTableExists(String dbName, String supertablename, String table) {
+        Connection conn = null;
+        Statement stmt = null;
+        try {
+            conn = getConnection();
+            stmt = conn.createStatement();
+            stmt.setQueryTimeout(5);
+
+            // 检查超级表是否存在（使用 information_schema）
+            String checkStableSql = String.format(
+                    "SELECT * FROM information_schema.ins_tables WHERE table_name = '%s' AND db_name = '%s'",
+                    escapeValue(supertablename), escapeValue(dbName));
+            ResultSet rs = stmt.executeQuery(checkStableSql);
+            boolean stableExists = rs.next();
+            rs.close();
+
+            if (!stableExists) {
+                log.info("超级表不存在，创建: {}.{}", dbName, supertablename);
+                initTableStructure(dbName, supertablename, table, Collections.emptySet());
+                return;
+            }
+
+            // 检查子表是否存在
+            String checkTableSql = String.format(
+                    "SELECT * FROM information_schema.ins_tables WHERE table_name = '%s' AND db_name = '%s' AND table_type = 'CHILD_TABLE'",
+                    escapeValue(table), escapeValue(dbName));
+            rs = stmt.executeQuery(checkTableSql);
+            boolean tableExists = rs.next();
+            rs.close();
+
+            if (!tableExists) {
+                String tableSql = String.format(
+                        "CREATE TABLE IF NOT EXISTS %s.%s USING %s.%s TAGS ('%s')",
+                        wrapName(dbName),
+                        wrapName(table),
+                        wrapName(dbName),
+                        wrapName(supertablename),
+                        escapeValue(supertablename)
+                );
+                stmt.executeUpdate(tableSql);
+                log.info("✅ 子表创建成功: {}", table);
+            }
+        } catch (SQLException e) {
+            log.warn("检查表存在性失败，继续尝试插入: {}", e.getMessage());
+        } finally {
+            if (stmt != null) try { stmt.close(); } catch (SQLException ignored) {}
+            closeConnection(conn);
+        }
+    }
+
+    /**
+     * 将动态字段构建为 JSON 字符串（排除 topic、ts、surfacename）
+     */
+    private String buildDynamicJson(Map<String, Object> data) {
+        if (data == null) return "{}";
+        Map<String, Object> dynamic = new LinkedHashMap<>();
+        for (Map.Entry<String, Object> entry : data.entrySet()) {
+            String key = entry.getKey();
+            if ("topic".equalsIgnoreCase(key) || "ts".equalsIgnoreCase(key)
+                    || "surfacename".equalsIgnoreCase(key)) {
+                continue;
+            }
+            Object value = entry.getValue();
+            if (value != null && !value.toString().trim().isEmpty()) {
+                dynamic.put(key, value);
+            }
+        }
+        try {
+            return objectMapper.writeValueAsString(dynamic);
+        } catch (Exception e) {
+            log.warn("JSON 序列化失败，返回空对象: {}", e.getMessage());
+            return "{}";
+        }
+    }
+
+    /**
+     * GZIP 压缩字符串
+     */
+    private String compressToBase64(String data) {
+        if (data == null || data.isEmpty()) return "";
+        try {
+            ByteArrayOutputStream bos = new ByteArrayOutputStream();
+            GZIPOutputStream gzip = new GZIPOutputStream(bos);
+            gzip.write(data.getBytes("UTF-8"));
+            gzip.close();
+            byte[] compressed = Base64.getEncoder().encode(bos.toByteArray());
+            return new String(compressed);
+        } catch (Exception e) {
+            log.warn("GZIP 压缩失败，使用原始数据: {}", e.getMessage());
+            return data;
+        }
+    }
+
     @Deprecated
     public boolean addToBatch(String dbName, String table, Map<String, Object> dataMap) {
         List<Map<String, Object>> list = new ArrayList<>();

iot-platform/src/main/java/com/iot/platform/task/VehicleSyncTask.java

@@ -178,6 +178,15 @@ public class VehicleSyncTask {
                     continue;
                 }
                 String controllerId = parts[1];
+                try {
+                    sysrealtimeService.createrealtime(controllerId);
+                } catch (Exception e) {
+                    Throwable cause = e;
+                    while (cause.getCause() != null) cause = cause.getCause();
+                    log.error("创建表失败: {} | {} | {}", controllerId, e.getMessage(), cause.getMessage());
+                    e.printStackTrace();
+                    continue;
+                }
                 String createTime = getStringValue(dataMap, "createTime");
                 String timestamp = getStringValue(dataMap, "timestamp");
                 String deviceId = getStringValue(dataMap, "device_id");

iot-platform/src/main/resources/application.yml

@@ -14,6 +14,7 @@ server:
 logging:
   level:
     com.iot.platform: debug
+    com.iot.platform.mapper: trace
     org.springframework: warn
 
 # Spring配置

iot-platform/src/main/resources/mapper/SysrealtimeMapper.xml

@@ -18,12 +18,10 @@
             timestamp VARCHAR(255) NOT NULL COMMENT '时间戳',
             k VARCHAR(255) NOT NULL COMMENT 'key',
             v VARCHAR(255) NOT NULL COMMENT '值',
-            -- 单独索引
-            INDEX device_id (device_id),
-            -- 联合索引（适合同时根据device_id和create_time查询的场景）
-            INDEX device_id_create_time (device_id, create_time)
-            INDEX k (k)
-            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时数据信息';
+            INDEX idx_device_id (device_id),
+            INDEX idx_device_create_time (device_id, create_time),
+            INDEX idx_k (k)
+        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时数据信息';
     </update>
 
     <select id="selecttables" resultType="String">
@@ -39,4 +37,4 @@
     <select id="selectkey" resultType="Integer">
         select COUNT(*) from `${tableName}` where k=#{k}
     </select>
-</mapper>
+</mapper>