refactor: P0-P2 security, quality, deployment overhaul

Security (P0):
- Remove hardcoded MySQL password from SysrealtimeService
- Add table name whitelist validation (regex + length limit)
- Validate ${tableName} in SysrealtimeMapper.xml
- Externalize all credentials to .env file
- Fix unbounded thread pools in MQTT consumers
- Fix mysqlWritePool leak in MqttFaultConsumer

Quality (P1):
- Extract AbstractMqttConsumer base class (-500+ duplicate lines)
- Refactor VehicleSyncTask: SCAN instead of KEYS, batch ops,
  RestTemplate timeout, loop-invariant extraction
- Add unit tests (RedisKeys, SysrealtimeService, MqttConsumer)

Architecture (P2):
- Add Spring Boot Actuator (/actuator/health)
- Create RedisKeys constants class for key namespace
- Add .env.example and start.sh for local development

Deployment:
- Add systemd service (iot-platform.service)
- Add deploy.sh with build, upload, backup, health check, rollback
- Add setup-server.sh for server initialization
- Add health-check.sh (Actuator + TCP port fallback)

Docs:
- Update CLAUDE.md for standalone iot-platform architecture
- Update README.md with build/deploy instructions
- Add deploy/README.md

.gitignore

@@ -48,3 +48,7 @@ nbdist/
 
 # Claude
 .claude/
+
+# Environment variables
+.env
+!.env.example

CLAUDE.md

@@ -4,162 +4,185 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ## Project Overview
 
-This is **RuoYi v3.9.0** (若依), a Chinese Spring Boot + Vue admin framework customized for IoT/Energy Storage Management (储能运营平台). It extends the standard RuoYi RBAC system with IoT device management, time-series data ingestion, MQTT messaging, and vehicle/equipment tracking.
+This is an **IoT Energy Storage Management Platform** (储能运营平台), originally based on RuoYi v3.9.0 but now stripped down to a standalone Spring Boot module focused on IoT device telemetry ingestion, MQTT messaging, time-series data storage, and vehicle/equipment tracking.
 
 - **Backend**: Spring Boot 2.5.15, Java 8, Spring Security 5.7.12, JWT authentication
 - **Frontend**: Vue 2 + Element UI (separate repository, not in this workspace)
-- **Documentation**: http://doc.ruoyi.vip
+- **Database**: MySQL 8 (dual datasource via Druid), TDengine 3.x for time-series
+- **Cache**: Redis (localhost:6379)
+- **Message Broker**: MQTT (EMQX on 47.104.204.180:1883)
 
 ## Build & Run Commands
 
-This is a Maven multi-module project. All commands should be run from the repository root.
-
 ```bash
-# Compile all modules
-mvn clean compile
-
-# Package into executable JAR (skips tests — there are no tests in this project)
-mvn clean package -Dmaven.test.skip=true
+# Build the iot-platform module
+mvn clean package -pl iot-platform -am -Dmaven.test.skip=true
 
-# Run the application (after packaging)
-cd ruoyi-admin/target
-java -jar ruoyi-admin.jar
+# Run locally with .env
+source .env && cd iot-platform/target && java -jar iot-platform.jar
 
-# Or use the provided script
-./ry.sh start    # start / stop / restart / status
+# Or use the convenience script
+./start.sh start
 ```
 
-The application starts on **port 8887**. The executable JAR is `ruoyi-admin/target/ruoyi-admin.jar`.
+The application starts on **port 8887**. The executable JAR is `iot-platform/target/iot-platform.jar`.
 
 ## Module Architecture
 
+This is now a **single-module Spring Boot application** (previously a multi-module RuoYi project).
+
 ```
-ruoyi-admin       → Web layer: REST controllers, Swagger config, application.yml entrypoint
-ruoyi-framework   → Infrastructure: Spring Security config, JWT filter, Druid datasource,
-                    Redis config, MyBatis config, AOP aspects (logging, rate limit, data scope)
-ruoyi-system      → Business layer: domain models, mappers (MyBatis), services,
-                    service implementations — this is where custom IoT logic lives
-ruoyi-common      → Shared utilities: constants, enums, exception types, utils (StringUtils,
-                    SecurityUtils, DateUtils, ServletUtils, etc.)
-ruoyi-quartz      → Scheduled job engine (standard RuoYi)
-ruoyi-generator   → Code generation templates (standard RuoYi)
+iot-platform/
+├── controller     → REST API endpoints, Swagger config
+├── service        → Business logic: IoT data sync, MQTT consumers, TDengine ops
+├── domain         → Entity classes, DTOs, VO objects
+├── mapper         → MyBatis mapper interfaces + XML in resources/mapper/
+├── mqtt           → MQTT consumers (AbstractMqttConsumer base class)
+├── task           → Scheduled tasks (VehicleSyncTask, etc.)
+├── config         → Spring configuration classes
+├── common         → Shared constants (RedisKeys), utilities
+└── datasource     → Dynamic datasource configuration
 ```
 
-Dependency flow: `admin → framework → system → common`. `quartz` and `generator` are optional add-ons.
-
 ## Data Architecture
 
 ### Dual MySQL Databases (Druid connection pool)
-- **Master** (`data`): RuoYi system tables (users, roles, menus, logs, custom business tables)
+- **Master** (`data`): System tables (users, roles, menus) + business tables
 - **Slave** (`cnc`): Secondary data — configured in `application-druid.yml`
 - Both hosted on `47.104.204.180:3306`
 - Dynamic datasource switching via `@DataSource` annotation
 
 ### TDengine (Time-Series Database)
 - Used for high-frequency IoT telemetry ingestion
-- Connection pool: HikariCP via `TDengineService` (`ruoyi-system/.../service/TDengineService.java`)
+- Connection pool: HikariCP via `TDengineService`
 - Connects to `jdbc:TAOS://localhost:6030/`
-- Super-table pattern: `CREATE STABLE ... (ts TIMESTAMP, surfacename VARCHAR(64)) TAGS (location BINARY(64))`
-- Column caching in memory (`stableColumnCache`) to avoid repeated `DESCRIBE` calls
+- Super-table pattern with column caching (`stableColumnCache`)
 
 ### Redis
 - Host: `localhost:6379`
-- Key patterns used by custom code:
-  - `DSB:active:devices` — Set of active IoT device Redis keys
-  - `DSB:<controllerId>:<metricName>` — Hash storing device telemetry fields
-  - `workorder:coordinate:<controllerId>` — Hash storing vehicle GPS (latitude, longitude)
-  - `<controllerId>:<topicName>` / `<controllerId>_cmd:<topicName>` / `<controllerId>_fault:<topicName>` — MQTT topic metadata
-
-## Custom Business Domains (Beyond Standard RuoYi)
-
-Standard RuoYi provides users, roles, menus, depts, posts, dicts, config, notices, operlog, logininfor. This project adds:
-
-| Domain | Purpose | Key Files |
-|--------|---------|-----------|
-| **SysDevice** | IoT device telemetry key-value storage | `domain/SysDevice.java`, `mapper/SysDeviceMapper.java`, `service/SysDeviceService.java` |
-| **SysDeviceControl** | Device configuration / metric whitelist | `domain/SysDeviceControl.java`, `service/SysDeviceControlService.java` |
-| **SysDeviceVo** | Aggregated device view for frontend | `domain/vo/SysDeviceVo.java`, `service/SysDeviceVoService.java` |
-| **SysCar** | Vehicle/equipment with GPS tracking | `domain/SysCar.java`, `service/SysCarService.java` |
-| **SysAlarm** | Alarm events from devices | `domain/SysAlarm.java`, `service/SysAlarmService.java` |
-| **SysFault** | Fault records | `domain/SysFault.java`, `service/SysFaultService.java` |
-| **SysWorkorder** | Work orders / service tickets | `domain/SysWorkorder.java`, `service/SysWorkorderService.java` |
-| **SysCompany** | Partner company management | `domain/SysCompany.java`, `service/SysCompanyService.java` |
-| **SysIndicators** | Daily business KPIs (order count, profit) | `domain/SysIndicators.java`, `service/SysIndicatorsService.java` |
-| **SysController** | MQTT topic registry per controller | `domain/SysController.java`, `service/SysControllerService.java` |
-| **Sysrealtime** | Real-time device data sync from Redis → MySQL | `domain/Sysrealtime.java`, `service/SysrealtimeService.java` |
-| **ControllerData / topics** | DTOs for MQTT JSON payload parsing | `domain/ControllerData.java`, `domain/topics.java` |
-| **MqttClientWrapper** | Wrapper around Eclipse Paho MQTT client | `domain/MqttClientWrapper.java` |
+- Key patterns (defined in `common/RedisKeys.java`):
+  - `DSB:active:devices` — Set of active IoT device keys
+  - `DSB:<controllerId>:<metricName>` — Hash storing device telemetry
+  - `workorder:coordinate:<controllerId>` — GPS coordinates
+  - `<controllerId>:<topicName>` — MQTT topic metadata
+
+## Key Configuration Files
+
+- `iot-platform/src/main/resources/application.yml` — Main config (port 8887, Redis, MyBatis, Actuator)
+- `iot-platform/src/main/resources/application-druid.yml` — Database connection (Druid pool, master/slave)
+- `iot-platform/src/main/resources/logback-spring.xml` — Logging config with rolling
+- `.env` — Environment variables (passwords, credentials) loaded by systemd or start.sh
+
+## Environment Variables
+
+All sensitive credentials are externalized to `.env`:
+
+```bash
+MYSQL_USERNAME=root
+MYSQL_PASSWORD="..."
+REDIS_PASSWORD=
+MQTT_USERNAME=...
+MQTT_PASSWORD=...
+TDENGINE_USERNAME=root
+TDENGINE_PASSWORD=taosdata
+DRUID_STAT_ENABLED=false
+DRUID_USERNAME=ruoyi
+DRUID_PASSWORD=...
+```
+
+Spring Boot config uses `${ENV_NAME:default}` syntax for all sensitive values.
 
 ## IoT Data Flow
 
 ```
 MQTT Broker (47.104.204.180:1883)
-    ↓  subscribes to "+/generics"
-TdengineController2 / TdegnineController3 (MQTT listeners)
-    ↓  parse JSON → ControllerData → topics
+    ↓  subscribes to "+/generics", "+/status", "+/fault"
+MqttGenericConsumer / MqttStatusConsumer / MqttFaultConsumer
+    ↓  parse JSON → ControllerData
     ├─→ Redis (DSB:active:devices, DSB:<id>:<metric> hashes)
-    ├─→ MySQL (sys_controller table — topic registry)
-    └─→ TDengine (time-series telemetry)
+    ├─→ MySQL (sys_controller, sys_device tables)
+    └─→ TDengine (time-series telemetry tables)
 
-SysCarController (scheduled tasks, every 30s)
+VehicleSyncTask (@Scheduled every 30s)
     ↓ reads Redis DSB:* keys
     ├─→ syncs to sysrealtime (MySQL)
     ├─→ updates vehicle GPS in sys_car (MySQL)
     ├─→ updates sys_indicators KPIs (MySQL)
-    └─→ triggers external webhook (https://esos-iot.com:9443/syscar/trigger)
+    └─→ triggers external webhook
 ```
 
-There are **4 `@Scheduled(fixedRate = 30000)` tasks** in `SysCarController`:
-1. `updatesyscar()` — sync vehicle GPS from Redis to MySQL, trigger webhook on change
-2. `insertdevice()` — sync device config from Redis to `sys_device_vo`
-3. `syncRedisToMySQL()` — sync telemetry from Redis to `sysrealtime`
-4. `insertindicators()` — aggregate daily KPIs per company into `sys_indicators`
+## MQTT Consumers
 
-## Key Configuration Files
+All MQTT consumers extend `AbstractMqttConsumer` which provides:
+- Connection management (reconnect, keepalive)
+- `checkServerAvailability()` with retry limits
+- Graceful shutdown via `@PreDestroy`
+
+Implementations:
+- `MqttGenericConsumer` — General telemetry ingestion
+- `MqttStatusConsumer` — Device status messages
+- `MqttFaultConsumer` — Fault/alarm messages
+
+## Deployment
+
+Production deployment uses systemd on online180 (47.104.204.180).
+
+```bash
+# Deploy from local
+cd deploy
+./deploy.sh --build          # Build and deploy
+./deploy.sh --jar <path>     # Deploy specific jar
+
+# Server management
+systemctl status iot-platform
+systemctl restart iot-platform
+journalctl -u iot-platform -f
+```
 
-- `ruoyi-admin/src/main/resources/application.yml` — Main config (port 8887, Redis, MyBatis, PageHelper, Swagger, token settings)
-- `ruoyi-admin/src/main/resources/application-druid.yml` — Database connection (master/slave MySQL, Druid pool)
-- `ruoyi-admin/src/main/resources/mybatis/mybatis-config.xml` — MyBatis global settings
-- `ruoyi-admin/src/main/resources/logback.xml` — Logging config
+See `deploy/README.md` for full deployment documentation.
 
 ## Authentication & Security
 
-- JWT-based stateless auth (token header: `Authorization`)
-- Spring Security config in `ruoyi-framework/.../config/SecurityConfig.java`
-- JWT filter: `JwtAuthenticationTokenFilter.java`
+- JWT-based stateless auth (header: `Authorization`)
+- Spring Security config in `config/SecurityConfig.java`
 - Password max retry: 5 attempts, lock time: 10 minutes
-- Swagger enabled at `/dev-api` path mapping
-- XSS filtering enabled for `/system/*`, `/monitor/*`, `/tool/*`
+- XSS filtering enabled
+- SQL injection prevention: field whitelist + parameterized queries in TDengineService
+- Table name validation: regex `^[a-zA-Z_][a-zA-Z0-9_]*$` in SysrealtimeService
 
 ## API Response Format
 
-All controllers return `AjaxResult` (from `ruoyi-common`), which wraps:
+All controllers return `AjaxResult`:
 ```java
 {
-  "code": 200,          // HttpStatus constant
-  "msg": "操作成功",     // message
-  "data": { ... }       // payload
+  "code": 200,
+  "msg": "操作成功",
+  "data": { ... }
 }
 ```
 
-Page queries use `TableDataInfo` which wraps:
+Page queries use `TableDataInfo`:
 ```java
 {
   "code": 200,
   "msg": "查询成功",
-  "rows": [ ... ],      // list data
-  "total": 100          // total count
+  "rows": [ ... ],
+  "total": 100
 }
 ```
 
+## Health Check
+
+Spring Boot Actuator is enabled at `/actuator/health`:
+```bash
+curl http://localhost:8887/actuator/health
+# {"status":"UP"}
+```
+
 ## Important Implementation Notes
 
-- **No unit tests exist** in this project. Test coverage is 0%.
-- **Field injection is used everywhere** (`@Autowired` on fields), not constructor injection.
 - **MyBatis mappers are Java interfaces** with XML mapping files in `src/main/resources/mapper/`
-- **Dynamic SQL provider**: `ruoyi-system/.../mapper/DynamicSqlProvider.java` provides programmatic SQL for some queries
-- **Service naming convention**: RuoYi standard services use `ISysXxxService` interface + `SysXxxServiceImpl` implementation. Custom services (added by this project) often omit the `I` prefix and use concrete classes directly.
-- **TDengine SQL is built with string concatenation** in `TDengineService.java` — be careful with SQL injection if user input reaches those paths.
-- **MQTT credentials are hardcoded** in `TdengineController2.java` and `TdegnineController3.java`.
-- **Database passwords are in plaintext** in `application-druid.yml`.
-- **The project uses Alibaba Maven mirror** (`https://maven.aliyun.com/repository/public`) configured in root `pom.xml`.
+- **Service naming**: Custom IoT services often use concrete classes directly (no `I` prefix)
+- **TDengine SQL**: Built with string concatenation but protected by field whitelist (`ALLOWED_COLUMNS`) and `escapeValue()`
+- **No distributed locks**: `VehicleSyncTask` scheduled tasks may duplicate in clustered environments
+- **Alibaba Maven mirror** (`https://maven.aliyun.com/repository/public`) configured in root `pom.xml`

README.md

@@ -3,6 +3,10 @@
 </p>
 <h1 align="center" style="margin: 30px 0 30px; font-weight: bold;">RuoYi v3.9.0</h1>
 <h4 align="center">基于SpringBoot+Vue前后端分离的Java快速开发框架</h4>
+
+> **本项目说明**：此仓库为 RuoYi v3.9.0 剥离后的 <b>IoT 储能运营平台</b> 独立模块。移除了 RuoYi 原生的 admin/framework/system/common/quartz/generator 等模块，仅保留 IoT 业务核心代码（设备管理、MQTT 消息接入、时序数据存储、车辆定位同步等）。
+>
+> 生产服务器：`online180` (47.104.204.180)，使用 systemd 管理服务，部署脚本位于 `deploy/` 目录。
 <p align="center">
 	<a href="https://gitee.com/y_project/RuoYi-Vue/stargazers"><img src="https://gitee.com/y_project/RuoYi-Vue/badge/star.svg?theme=dark"></a>
 	<a href="https://gitee.com/y_project/RuoYi-Vue"><img src="https://img.shields.io/badge/RuoYi-v3.9.0-brightgreen.svg"></a>
@@ -90,6 +94,64 @@
 </table>
 
 
+## IoT 平台构建与部署
+
+### 本地构建
+
+```bash
+# 构建 iot-platform 模块
+mvn clean package -pl iot-platform -am -Dmaven.test.skip=true
+
+# 本地运行（需先配置 .env）
+source .env && ./start.sh start
+```
+
+### 环境变量
+
+项目根目录 `.env` 文件包含所有敏感凭据（已加入 `.gitignore`）：
+
+```bash
+MYSQL_USERNAME=root
+MYSQL_PASSWORD="..."
+MQTT_USERNAME=...
+MQTT_PASSWORD=...
+TDENGINE_USERNAME=root
+TDENGINE_PASSWORD=taosdata
+```
+
+Spring Boot 配置使用 `${ENV_NAME:default}` 语法读取这些变量。
+
+### 生产部署
+
+```bash
+cd deploy
+
+# 一键构建并部署到 online180
+./deploy.sh --build
+
+# 或指定 jar 部署
+./deploy.sh --jar ../iot-platform/target/iot-platform.jar
+```
+
+部署流程：构建 → 上传 → 备份 → 停止 → 替换 → 启动 → 健康检查 → 失败自动回滚。
+
+详见 `deploy/README.md`。
+
+### 服务器管理
+
+```bash
+# 查看状态
+systemctl status iot-platform
+
+# 查看实时日志
+journalctl -u iot-platform -f
+
+# 重启服务
+systemctl restart iot-platform
+```
+
+---
+
 ## 若依前后端分离交流群
 
 QQ群： [![加入QQ群](https://img.shields.io/badge/已满-937441-blue.svg)](https://jq.qq.com/?_wv=1027&k=5bVB1og) [![加入QQ群](https://img.shields.io/badge/已满-887144332-blue.svg)](https://jq.qq.com/?_wv=1027&k=5eiA4DH) [![加入QQ群](https://img.shields.io/badge/已满-180251782-blue.svg)](https://jq.qq.com/?_wv=1027&k=5AxMKlC) [![加入QQ群](https://img.shields.io/badge/已满-104180207-blue.svg)](https://jq.qq.com/?_wv=1027&k=51G72yr) [![加入QQ群](https://img.shields.io/badge/已满-186866453-blue.svg)](https://jq.qq.com/?_wv=1027&k=VvjN2nvu) [![加入QQ群](https://img.shields.io/badge/已满-201396349-blue.svg)](https://jq.qq.com/?_wv=1027&k=5vYAqA05) [![加入QQ群](https://img.shields.io/badge/已满-101456076-blue.svg)](https://jq.qq.com/?_wv=1027&k=kOIINEb5) [![加入QQ群](https://img.shields.io/badge/已满-101539465-blue.svg)](https://jq.qq.com/?_wv=1027&k=UKtX5jhs) [![加入QQ群](https://img.shields.io/badge/已满-264312783-blue.svg)](https://jq.qq.com/?_wv=1027&k=EI9an8lJ) [![加入QQ群](https://img.shields.io/badge/已满-167385320-blue.svg)](https://jq.qq.com/?_wv=1027&k=SWCtLnMz) [![加入QQ群](https://img.shields.io/badge/已满-104748341-blue.svg)](https://jq.qq.com/?_wv=1027&k=96Dkdq0k) [![加入QQ群](https://img.shields.io/badge/已满-160110482-blue.svg)](https://jq.qq.com/?_wv=1027&k=0fsNiYZt) [![加入QQ群](https://img.shields.io/badge/已满-170801498-blue.svg)](https://jq.qq.com/?_wv=1027&k=7xw4xUG1) [![加入QQ群](https://img.shields.io/badge/已满-108482800-blue.svg)](https://jq.qq.com/?_wv=1027&k=eCx8eyoJ) [![加入QQ群](https://img.shields.io/badge/已满-101046199-blue.svg)](https://jq.qq.com/?_wv=1027&k=SpyH2875) [![加入QQ群](https://img.shields.io/badge/已满-136919097-blue.svg)](https://jq.qq.com/?_wv=1027&k=tKEt51dz) [![加入QQ群](https://img.shields.io/badge/已满-143961921-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=0vBbSb0ztbBgVtn3kJS-Q4HUNYwip89G&authKey=8irq5PhutrZmWIvsUsklBxhj57l%2F1nOZqjzigkXZVoZE451GG4JHPOqW7AW6cf0T&noverify=0&group_code=143961921) [![加入QQ群](https://img.shields.io/badge/已满-174951577-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=ZFAPAbp09S2ltvwrJzp7wGlbopsc0rwi&authKey=HB2cxpxP2yspk%2Bo3WKTBfktRCccVkU26cgi5B16u0KcAYrVu7sBaE7XSEqmMdFQp&noverify=0&group_code=174951577) [![加入QQ群](https://img.shields.io/badge/已满-161281055-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=Fn2aF5IHpwsy8j6VlalNJK6qbwFLFHat&authKey=uyIT%2B97x2AXj3odyXpsSpVaPMC%2Bidw0LxG5MAtEqlrcBcWJUA%2FeS43rsF1Tg7IRJ&noverify=0&group_code=161281055) [![加入QQ群](https://img.shields.io/badge/已满-138988063-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=XIzkm_mV2xTsUtFxo63bmicYoDBA6Ifm&authKey=dDW%2F4qsmw3x9govoZY9w%2FoWAoC4wbHqGal%2BbqLzoS6VBarU8EBptIgPKN%2FviyC8j&noverify=0&group_code=138988063) [![加入QQ群](https://img.shields.io/badge/已满-151450850-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=DkugnCg68PevlycJSKSwjhFqfIgrWWwR&authKey=pR1Pa5lPIeGF%2FFtIk6d%2FGB5qFi0EdvyErtpQXULzo03zbhopBHLWcuqdpwY241R%2F&noverify=0&group_code=151450850) [![加入QQ群](https://img.shields.io/badge/已满-224622315-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=F58bgRa-Dp-rsQJThiJqIYv8t4-lWfXh&authKey=UmUs4CVG5OPA1whvsa4uSespOvyd8%2FAr9olEGaWAfdLmfKQk%2FVBp2YU3u2xXXt76&noverify=0&group_code=224622315) [![加入QQ群](https://img.shields.io/badge/已满-287842588-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=Nxb2EQ5qozWa218Wbs7zgBnjLSNk_tVT&authKey=obBKXj6SBKgrFTJZx0AqQnIYbNOvBB2kmgwWvGhzxR67RoRr84%2Bus5OadzMcdJl5&noverify=0&group_code=287842588) [![加入QQ群](https://img.shields.io/badge/已满-187944233-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=numtK1M_I4eVd2Gvg8qtbuL8JgX42qNh&authKey=giV9XWMaFZTY%2FqPlmWbkB9g3fi0Ev5CwEtT9Tgei0oUlFFCQLDp4ozWRiVIzubIm&noverify=0&group_code=187944233) [![加入QQ群](https://img.shields.io/badge/已满-228578329-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=G6r5KGCaa3pqdbUSXNIgYloyb8e0_L0D&authKey=4w8tF1eGW7%2FedWn%2FHAypQksdrML%2BDHolQSx7094Agm7Luakj9EbfPnSTxSi2T1LQ&noverify=0&group_code=228578329) [![加入QQ群](https://img.shields.io/badge/191164766-blue.svg)](http://qm.qq.com/cgi-bin/qm/qr?_wv=1027&k=GsOo-OLz53J8y_9TPoO6XXSGNRTgbFxA&authKey=R7Uy%2Feq%2BZsoKNqHvRKhiXpypW7DAogoWapOawUGHokJSBIBIre2%2FoiAZeZBSLuBc&noverify=0&group_code=191164766) 点击按钮入群。

deploy/README.md

@@ -0,0 +1,158 @@
+# IoT Platform 部署文档
+
+## 概述
+
+本项目使用 systemd 管理进程，通过 `deploy.sh` 脚本实现一键构建、上传、备份、健康检查和自动回滚。
+
+## 服务器环境
+
+- **服务器**: online180 (47.104.204.180)
+- **OS**: Alibaba Cloud Linux 3
+- **Java**: OpenJDK 1.8.0_412
+- **安装目录**: `/opt/iot-platform/`
+- **服务名称**: `iot-platform`
+
+## 目录结构
+
+```
+/opt/iot-platform/
+├── iot-platform.jar          # 当前运行版本
+├── .env                      # 环境变量（密码、凭据）
+├── backup/                   # 版本备份（自动创建）
+├── bin/
+│   ├── deploy.sh             # 一键部署脚本（本地运行）
+│   └── health-check.sh       # 健康检查脚本
+├── config/                   # 线上专属配置
+└── logs/                     # 日志输出
+```
+
+## 快速开始
+
+### 首次部署（服务器初始化）
+
+如果服务器上还没有安装 systemd 服务，执行初始化：
+
+```bash
+cd deploy
+scp setup-server.sh root@47.104.204.180:/tmp/
+scp iot-platform.service root@47.104.204.180:/tmp/
+scp ../.env root@47.104.204.180:/tmp/iot-platform.env
+ssh root@47.104.204.180 "chmod +x /tmp/setup-server.sh && /tmp/setup-server.sh"
+```
+
+### 日常部署
+
+#### 方式一：本地构建后部署
+
+```bash
+cd deploy
+./deploy.sh --build
+```
+
+流程：
+1. 本地执行 `mvn clean package`
+2. 上传 jar 到服务器
+3. 备份当前版本（带时间戳）
+4. 停止 systemd 服务
+5. 替换 jar
+6. 启动服务
+7. 健康检查（`/actuator/health`）
+8. 失败则自动回滚到上一个版本
+
+#### 方式二：指定 jar 部署
+
+```bash
+cd deploy
+./deploy.sh --jar ../iot-platform/target/iot-platform.jar
+```
+
+#### 方式三：CI/CD 构建后部署
+
+```bash
+cd deploy
+./deploy.sh --jar /path/to/ci/artifacts/iot-platform.jar
+```
+
+## 环境变量
+
+`.env` 文件位于项目根目录和服务器 `/opt/iot-platform/.env`，包含所有敏感凭据：
+
+```bash
+# MySQL
+MYSQL_USERNAME=root
+MYSQL_PASSWORD="..."
+
+# Redis
+REDIS_PASSWORD=
+
+# MQTT
+MQTT_USERNAME=...
+MQTT_PASSWORD=...
+
+# TDengine
+TDENGINE_USERNAME=root
+TDENGINE_PASSWORD=taosdata
+
+# Druid 监控
+DRUID_STAT_ENABLED=false
+DRUID_USERNAME=ruoyi
+DRUID_PASSWORD=...
+```
+
+**注意**：`.env` 文件已加入 `.gitignore`，**切勿提交到 Git**。
+
+## 服务器管理
+
+```bash
+# 查看状态
+systemctl status iot-platform
+
+# 重启
+systemctl restart iot-platform
+
+# 停止
+systemctl stop iot-platform
+
+# 查看日志
+journalctl -u iot-platform -f
+
+# 查看历史日志
+journalctl -u iot-platform --since "1 hour ago"
+```
+
+## 健康检查
+
+```bash
+# 在服务器上执行
+bash /opt/iot-platform/bin/health-check.sh localhost 8887 30
+
+# 或从本地检查
+curl -s http://47.104.204.180:8887/actuator/health
+```
+
+健康检查策略：
+1. 优先检查 `/actuator/health`（Spring Boot Actuator）
+2. 如果 Actuator 未启用，降级为 TCP 端口监听检查
+
+## 回滚
+
+如果部署失败，`deploy.sh` 会自动回滚到上一个版本。
+
+手动回滚：
+
+```bash
+ssh root@47.104.204.180
+systemctl stop iot-platform
+ls /opt/iot-platform/backup/          # 查看可用备份
+cp /opt/iot-platform/backup/iot-platform-XXXX.jar /opt/iot-platform/iot-platform.jar
+systemctl start iot-platform
+```
+
+## 配置文件说明
+
+| 文件 | 说明 |
+|------|------|
+| `deploy.sh` | 一键部署脚本（本地执行） |
+| `setup-server.sh` | 服务器初始化脚本（服务器执行） |
+| `health-check.sh` | 健康检查脚本 |
+| `iot-platform.service` | systemd 服务定义 |

deploy/deploy.sh

@@ -0,0 +1,169 @@
+#!/bin/bash
+# IoT Platform 一键部署脚本
+# 用法:
+#   ./deploy.sh --build              # 本地构建后部署
+#   ./deploy.sh --jar path/to/jar    # 使用指定 jar 部署
+#   ./deploy.sh --build --jar path   # 构建并使用指定 jar（覆盖）
+#
+# 环境变量:
+#   SERVER_HOST    默认: 47.104.204.180
+#   SERVER_USER    默认: root
+#   REMOTE_DIR     默认: /opt/iot-platform
+
+set -e
+
+# 配置
+SERVER_HOST="${SERVER_HOST:-47.104.204.180}"
+SERVER_USER="${SERVER_USER:-root}"
+REMOTE_DIR="${REMOTE_DIR:-/opt/iot-platform}"
+APP_NAME="iot-platform"
+LOCAL_JAR=""
+DO_BUILD=false
+
+# 颜色
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+NC='\033[0m'
+
+# 解析参数
+while [[ $# -gt 0 ]]; do
+    case $1 in
+        --build)
+            DO_BUILD=true
+            shift
+            ;;
+        --jar)
+            LOCAL_JAR="$2"
+            shift 2
+            ;;
+        --help|-h)
+            echo "用法: ./deploy.sh [--build] [--jar <path>]"
+            echo "  --build     本地执行 mvn clean package"
+            echo "  --jar       指定要部署的 jar 文件路径"
+            exit 0
+            ;;
+        *)
+            echo "未知参数: $1"
+            exit 1
+            ;;
+    esac
+done
+
+# 确定 jar 路径
+if [ "$DO_BUILD" = true ]; then
+    echo -e "${YELLOW}[deploy] 开始本地构建...${NC}"
+    cd "$(dirname "$0")/.."
+    mvn clean package -pl iot-platform -am -Dmaven.test.skip=true
+    LOCAL_JAR="iot-platform/target/iot-platform.jar"
+    echo -e "${GREEN}[deploy] 构建完成: ${LOCAL_JAR}${NC}"
+fi
+
+if [ -z "$LOCAL_JAR" ]; then
+    # 默认使用本地 target 下的 jar
+    cd "$(dirname "$0")/.."
+    LOCAL_JAR="iot-platform/target/iot-platform.jar"
+fi
+
+if [ ! -f "$LOCAL_JAR" ]; then
+    echo -e "${RED}[deploy] 错误: 找不到 jar 文件: ${LOCAL_JAR}${NC}"
+    exit 1
+fi
+
+JAR_NAME=$(basename "$LOCAL_JAR")
+TIMESTAMP=$(date +%Y%m%d-%H%M%S)
+BACKUP_NAME="${APP_NAME}-${TIMESTAMP}.jar"
+
+echo "========================================"
+echo "IoT Platform 部署"
+echo "========================================"
+echo "目标服务器: ${SERVER_USER}@${SERVER_HOST}"
+echo "远程目录:   ${REMOTE_DIR}"
+echo "本地 JAR:   ${LOCAL_JAR}"
+echo "========================================"
+
+# 步骤 1: 上传 JAR 到临时目录
+echo -e "${YELLOW}[1/6] 上传 JAR 到服务器...${NC}"
+ssh "${SERVER_USER}@${SERVER_HOST}" "mkdir -p ${REMOTE_DIR}/tmp ${REMOTE_DIR}/backup"
+scp "${LOCAL_JAR}" "${SERVER_USER}@${SERVER_HOST}:${REMOTE_DIR}/tmp/${JAR_NAME}"
+echo -e "${GREEN}[1/6] 上传完成${NC}"
+
+# 步骤 2: 备份当前版本
+echo -e "${YELLOW}[2/6] 备份当前版本...${NC}"
+ssh "${SERVER_USER}@${SERVER_HOST}" "
+    if [ -f ${REMOTE_DIR}/${APP_NAME}.jar ]; then
+        cp -a ${REMOTE_DIR}/${APP_NAME}.jar ${REMOTE_DIR}/backup/${BACKUP_NAME}
+        echo '  已备份: ${BACKUP_NAME}'
+        ls -lh ${REMOTE_DIR}/backup/ | tail -5
+    else
+        echo '  无现有版本，跳过备份'
+    fi
+"
+echo -e "${GREEN}[2/6] 备份完成${NC}"
+
+# 步骤 3: 停止服务
+echo -e "${YELLOW}[3/6] 停止服务...${NC}"
+ssh "${SERVER_USER}@${SERVER_HOST}" "systemctl stop ${APP_NAME} || true"
+sleep 2
+echo -e "${GREEN}[3/6] 服务已停止${NC}"
+
+# 步骤 4: 替换 JAR
+echo -e "${YELLOW}[4/6] 替换 JAR...${NC}"
+ssh "${SERVER_USER}@${SERVER_HOST}" "
+    cp ${REMOTE_DIR}/tmp/${JAR_NAME} ${REMOTE_DIR}/${APP_NAME}.jar
+    chmod 644 ${REMOTE_DIR}/${APP_NAME}.jar
+    rm -f ${REMOTE_DIR}/tmp/${JAR_NAME}
+    echo '  新 JAR:'
+    ls -lh ${REMOTE_DIR}/${APP_NAME}.jar
+"
+echo -e "${GREEN}[4/6] 替换完成${NC}"
+
+# 步骤 5: 启动服务
+echo -e "${YELLOW}[5/6] 启动服务...${NC}"
+ssh "${SERVER_USER}@${SERVER_HOST}" "systemctl start ${APP_NAME}"
+sleep 3
+
+echo -e "${GREEN}[5/6] 服务已启动${NC}"
+
+# 步骤 6: 健康检查
+echo -e "${YELLOW}[6/6] 健康检查...${NC}"
+if ssh "${SERVER_USER}@${SERVER_HOST}" "bash ${REMOTE_DIR}/bin/health-check.sh localhost 8887 30"; then
+    echo -e "${GREEN}[6/6] 健康检查通过，部署成功!${NC}"
+    echo ""
+    echo "========================================"
+    echo "部署摘要"
+    echo "========================================"
+    echo "时间:     ${TIMESTAMP}"
+    echo "JAR:      ${LOCAL_JAR}"
+    echo "备份:     ${BACKUP_NAME}"
+    echo "状态:     成功"
+    echo ""
+    echo "服务状态:"
+    ssh "${SERVER_USER}@${SERVER_HOST}" "systemctl status ${APP_NAME} --no-pager"
+    exit 0
+else
+    echo -e "${RED}[6/6] 健康检查失败! 执行回滚...${NC}"
+
+    # 回滚
+    echo -e "${YELLOW}[rollback] 停止服务...${NC}"
+    ssh "${SERVER_USER}@${SERVER_HOST}" "systemctl stop ${APP_NAME} || true"
+
+    if ssh "${SERVER_USER}@${SERVER_HOST}" "test -f ${REMOTE_DIR}/backup/${BACKUP_NAME}"; then
+        echo -e "${YELLOW}[rollback] 恢复旧版本...${NC}"
+        ssh "${SERVER_USER}@${SERVER_HOST}" "
+            cp ${REMOTE_DIR}/backup/${BACKUP_NAME} ${REMOTE_DIR}/${APP_NAME}.jar
+            systemctl start ${APP_NAME}
+        "
+        sleep 3
+
+        if ssh "${SERVER_USER}@${SERVER_HOST}" "bash ${REMOTE_DIR}/bin/health-check.sh localhost 8887 30"; then
+            echo -e "${GREEN}[rollback] 回滚成功，旧版本已恢复${NC}"
+        else
+            echo -e "${RED}[rollback] 回滚后健康检查仍失败，请手动排查!${NC}"
+        fi
+    else
+        echo -e "${RED}[rollback] 无备份可回滚，请手动修复!${NC}"
+    fi
+
+    exit 1
+fi

deploy/health-check.sh

@@ -0,0 +1,48 @@
+#!/bin/bash
+# IoT Platform 健康检查脚本
+# 用法: ./health-check.sh [host] [port] [max_wait_seconds]
+# 返回: 0=健康, 1=不健康
+#
+# 检查策略:
+#   1. 优先检查 /actuator/health (Spring Boot Actuator)
+#   2. 如果 Actuator 返回 404，降级为检查 TCP 端口是否监听
+#   3. 如果连端口都没监听，则判定为不健康
+
+HOST=${1:-localhost}
+PORT=${2:-8887}
+MAX_WAIT=${3:-30}
+HEALTH_URL="http://${HOST}:${PORT}/actuator/health"
+
+INTERVAL=2
+ATTEMPTS=$((MAX_WAIT / INTERVAL))
+
+echo "[health-check] 开始检查: ${HOST}:${PORT}, 最多等待 ${MAX_WAIT} 秒"
+
+for i in $(seq 1 $ATTEMPTS); do
+    # 策略1: Actuator 健康端点
+    RESPONSE=$(curl -s -o /dev/null -w "%{http_code}" --connect-timeout 2 --max-time 3 "${HEALTH_URL}" 2>/dev/null)
+
+    if [ "$RESPONSE" = "200" ]; then
+        BODY=$(curl -s --connect-timeout 2 --max-time 3 "${HEALTH_URL}" 2>/dev/null)
+        if echo "$BODY" | grep -q '"status":"UP"'; then
+            echo "[health-check] Actuator 健康检查通过 (${i}/${ATTEMPTS})"
+            exit 0
+        fi
+    fi
+
+    # 策略2: 如果 Actuator 404，降级为端口监听检查
+    if [ "$RESPONSE" = "404" ]; then
+        if nc -z "${HOST}" "${PORT}" 2>/dev/null || \
+           ss -tlnp 2>/dev/null | grep -q ":${PORT} " || \
+           netstat -tlnp 2>/dev/null | grep -q ":${PORT} "; then
+            echo "[health-check] 端口监听检查通过 (${i}/${ATTEMPTS}) [Actuator 未启用，使用端口降级检查]"
+            exit 0
+        fi
+    fi
+
+    echo "[health-check] 第 ${i}/${ATTEMPTS} 次检查未通过 (HTTP ${RESPONSE}), ${INTERVAL} 秒后重试..."
+    sleep $INTERVAL
+done
+
+echo "[health-check] 健康检查失败: 服务未在 ${MAX_WAIT} 秒内就绪"
+exit 1

deploy/iot-platform.service

@@ -0,0 +1,32 @@
+[Unit]
+Description=IoT Platform Service
+After=network.target mysqld.service redis.service
+
+[Service]
+Type=simple
+User=root
+Group=root
+WorkingDirectory=/opt/iot-platform
+
+# JVM options
+Environment="JVM_OPTS=-server -Duser.timezone=Asia/Shanghai -Dfile.encoding=UTF-8 -Xms1g -Xmx2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=256m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/opt/iot-platform/logs/heapdump.hprof -XX:+UseG1GC -XX:MaxGCPauseMillis=200"
+
+# Load environment variables from .env (passwords, credentials)
+EnvironmentFile=/opt/iot-platform/.env
+
+ExecStart=/bin/bash -c 'exec java ${JVM_OPTS} -jar /opt/iot-platform/iot-platform.jar --spring.profiles.include=online'
+
+ExecStop=/bin/kill -SIGTERM $MAINPID
+ExecReload=/bin/kill -SIGUSR1 $MAINPID
+
+Restart=always
+RestartSec=10
+StartLimitInterval=60
+StartLimitBurst=3
+
+# Graceful shutdown: wait up to 60s for the app to stop
+TimeoutStopSec=60
+KillSignal=SIGTERM
+
+[Install]
+WantedBy=multi-user.target

deploy/setup-server.sh

@@ -0,0 +1,130 @@
+#!/bin/bash
+# IoT Platform 服务器初始化脚本
+# 在 online180 上执行，创建规范目录结构并迁移现有服务到 systemd
+
+set -e
+
+APP_NAME="iot-platform"
+INSTALL_DIR="/opt/${APP_NAME}"
+BACKUP_DIR="${INSTALL_DIR}/backup"
+LOG_DIR="${INSTALL_DIR}/logs"
+CONFIG_DIR="${INSTALL_DIR}/config"
+BIN_DIR="${INSTALL_DIR}/bin"
+
+OLD_JAR="/home/${APP_NAME}.jar"
+OLD_LOG_DIR="/home/logs"
+
+echo "========================================"
+echo "IoT Platform 服务器初始化"
+echo "========================================"
+
+# 1. 创建目录结构
+echo "[1/7] 创建目录结构..."
+mkdir -p "${INSTALL_DIR}" "${BACKUP_DIR}" "${LOG_DIR}" "${CONFIG_DIR}" "${BIN_DIR}"
+
+# 2. 迁移 JAR 文件
+echo "[2/7] 迁移 JAR 文件..."
+if [ -f "${OLD_JAR}" ]; then
+    cp -a "${OLD_JAR}" "${INSTALL_DIR}/${APP_NAME}.jar"
+    echo "  已复制: ${OLD_JAR} -> ${INSTALL_DIR}/${APP_NAME}.jar"
+else
+    echo "  警告: 未找到旧 JAR 文件 ${OLD_JAR}"
+fi
+
+# 3. 迁移日志
+echo "[3/7] 迁移日志..."
+if [ -d "${OLD_LOG_DIR}" ]; then
+    # 复制已有日志
+    cp -a "${OLD_LOG_DIR}"/* "${LOG_DIR}/" 2>/dev/null || true
+    echo "  已复制日志到 ${LOG_DIR}"
+    # 可选：备份旧日志目录，创建软链接
+    mv "${OLD_LOG_DIR}" "${OLD_LOG_DIR}.bak.$(date +%Y%m%d%H%M%S)" 2>/dev/null || true
+    ln -s "${LOG_DIR}" "${OLD_LOG_DIR}" 2>/dev/null || true
+    echo "  旧日志目录已备份并创建软链接"
+else
+    echo "  旧日志目录不存在，跳过迁移"
+fi
+
+# 4. 设置权限
+echo "[4/7] 设置权限..."
+chmod 755 "${INSTALL_DIR}" "${BACKUP_DIR}" "${LOG_DIR}" "${CONFIG_DIR}" "${BIN_DIR}"
+
+# 5. 安装 .env 文件（如果当前目录有）
+echo "[5/7] 安装环境变量文件..."
+if [ -f "/tmp/iot-platform.env" ]; then
+    cp "/tmp/iot-platform.env" "${INSTALL_DIR}/.env"
+    chmod 600 "${INSTALL_DIR}/.env"
+    echo "  已安装 .env"
+else
+    echo "  警告: 未找到 /tmp/iot-platform.env，请手动创建 ${INSTALL_DIR}/.env"
+fi
+
+# 6. 安装 systemd 服务
+echo "[6/7] 安装 systemd 服务..."
+if [ -f "/tmp/iot-platform.service" ]; then
+    cp "/tmp/iot-platform.service" /etc/systemd/system/iot-platform.service
+    chmod 644 /etc/systemd/system/iot-platform.service
+    systemctl daemon-reload
+    systemctl enable iot-platform
+    echo "  systemd 服务已安装并启用"
+else
+    echo "  错误: 未找到 /tmp/iot-platform.service"
+    exit 1
+fi
+
+# 7. 平滑切换：停止旧进程，启动 systemd
+echo "[7/7] 平滑切换到 systemd..."
+OLD_PID=$(ps -ef | grep java | grep "${APP_NAME}.jar" | grep -v grep | grep -v systemd | awk '{print $2}' | head -1)
+if [ -n "${OLD_PID}" ]; then
+    echo "  发现旧进程 PID=${OLD_PID}, 发送 SIGTERM..."
+    kill -TERM "${OLD_PID}" 2>/dev/null || true
+
+    # 等待最多 30 秒
+    for i in $(seq 1 30); do
+        if ! kill -0 "${OLD_PID}" 2>/dev/null; then
+            echo "  旧进程已停止"
+            break
+        fi
+        sleep 1
+    done
+
+    # 强制结束
+    if kill -0 "${OLD_PID}" 2>/dev/null; then
+        echo "  旧进程未响应，发送 SIGKILL..."
+        kill -KILL "${OLD_PID}" 2>/dev/null || true
+    fi
+else
+    echo "  未发现旧进程"
+fi
+
+echo ""
+echo "启动 systemd 服务..."
+systemctl start iot-platform
+
+sleep 3
+
+# 检查服务状态
+if systemctl is-active --quiet iot-platform; then
+    echo ""
+    echo "========================================"
+    echo "初始化成功!"
+    echo "========================================"
+    echo "服务状态:"
+    systemctl status iot-platform --no-pager
+    echo ""
+    echo "目录结构:"
+    ls -la "${INSTALL_DIR}"
+    echo ""
+    echo "常用命令:"
+    echo "  systemctl status iot-platform   # 查看状态"
+    echo "  journalctl -u iot-platform -f   # 实时日志"
+    echo "  systemctl stop iot-platform     # 停止服务"
+    echo "  systemctl restart iot-platform  # 重启服务"
+else
+    echo ""
+    echo "========================================"
+    echo "警告: 服务启动可能失败，请检查日志"
+    echo "========================================"
+    journalctl -u iot-platform --no-pager -n 50
+    exit 1
+fi

iot-platform/.env.example

@@ -0,0 +1,22 @@
+# IoT 平台环境变量配置示例
+# 生产环境部署前必须设置以下环境变量
+
+# MySQL 数据库（主从共用）
+export MYSQL_USERNAME=root
+export MYSQL_PASSWORD=your_mysql_password
+
+# Redis
+export REDIS_PASSWORD=your_redis_password
+
+# MQTT Broker
+export MQTT_USERNAME=your_mqtt_username
+export MQTT_PASSWORD=your_mqtt_password
+
+# TDengine 时序数据库
+export TDENGINE_USERNAME=root
+export TDENGINE_PASSWORD=your_tdengine_password
+
+# Druid 监控（可选，生产环境建议关闭）
+export DRUID_STAT_ENABLED=false
+export DRUID_USERNAME=admin
+export DRUID_PASSWORD=your_druid_password

iot-platform/pom.xml

@@ -125,6 +125,12 @@
             <optional>true</optional>
         </dependency>
 
+        <!-- Spring Boot Actuator -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+
         <!-- 测试 -->
         <dependency>
             <groupId>org.springframework.boot</groupId>

iot-platform/src/main/java/com/iot/platform/common/RedisKeys.java

@@ -0,0 +1,47 @@
+package com.iot.platform.common;
+
+/**
+ * Redis key 命名空间常量，统一前缀避免 key 冲突。
+ * 格式: iot:{domain}:{identifier}
+ */
+public final class RedisKeys {
+
+    private static final String PREFIX = "iot";
+
+    /** 活跃设备集合 */
+    public static final String ACTIVE_DEVICES = PREFIX + ":dsb:active:devices";
+
+    /** 设备遥测数据 hash */
+    public static String deviceTelemetry(String controllerId, String metricName) {
+        return PREFIX + ":dsb:" + controllerId + ":" + metricName;
+    }
+
+    /** 工单坐标 hash */
+    public static String workorderCoordinate(String controllerId) {
+        return PREFIX + ":workorder:coordinate:" + controllerId;
+    }
+
+    /** 控制器状态 hash */
+    public static String controllerStatus(String controllerId) {
+        return PREFIX + ":controller:" + controllerId + ":status";
+    }
+
+    /** 控制器 topic hash */
+    public static String controllerTopic(String controllerId, String topicName) {
+        return PREFIX + ":controller:" + controllerId + ":" + topicName;
+    }
+
+    /** 控制器 cmd topic hash */
+    public static String controllerCmdTopic(String controllerId, String topicName) {
+        return PREFIX + ":controller:" + controllerId + "_cmd:" + topicName;
+    }
+
+    /** 控制器 fault topic hash */
+    public static String controllerFaultTopic(String controllerId, String topicName) {
+        return PREFIX + ":controller:" + controllerId + "_fault:" + topicName;
+    }
+
+    private RedisKeys() {
+        // 工具类，禁止实例化
+    }
+}

iot-platform/src/main/java/com/iot/platform/mapper/SysrealtimeMapper.java

@@ -12,4 +12,5 @@ public interface SysrealtimeMapper {
     void inserttables(@Param("tableName")String tableName,@Param("createTime")String createTime,@Param("deviceId")String deviceId,@Param("timestamp")String timestamp,@Param("k")String k,@Param("v")Object v);
     void updatetables(@Param("tableName")String tableName,@Param("createTime")String createTime,@Param("v")Object v,@Param("timestamp")String timestamp,@Param("k")String k,@Param("deviceId")String deviceId);
     Integer selectkey(@Param("tableName")String tableName,@Param("k")String k);
+    List<String> selectAllKeys(@Param("tableName")String tableName);
 }

iot-platform/src/main/java/com/iot/platform/mqtt/AbstractMqttConsumer.java

@@ -0,0 +1,177 @@
+package com.iot.platform.mqtt;
+
+import com.iot.platform.config.IotProperties;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public abstract class AbstractMqttConsumer {
+
+    protected final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    protected IotProperties iotProperties;
+
+    private String brokerUrl;
+    private String brokerHost;
+    private int brokerPort;
+    private String mqttUsername;
+    private String mqttPassword;
+
+    private static final int QOS = 1;
+    private static final int CONNECT_TIMEOUT = 3000;
+    private static final int RECONNECT_INTERVAL = 5000;
+    private static final int MAX_AVAILABILITY_CHECKS = 60;
+
+    protected MqttClient mqttClient;
+    protected MqttConnectOptions connOpts;
+    protected volatile boolean isMqttConnected = false;
+    protected final ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+    protected abstract String getSubscribeTopic();
+
+    protected abstract String generateClientId();
+
+    protected abstract void handleMessage(String topic, String messageContent) throws Exception;
+
+    protected void onDestroy() {
+        // 子类可覆盖，用于关闭自定义资源
+    }
+
+    @PostConstruct
+    public void connectAndSubscribe() {
+        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
+        String brokerAddr = this.brokerUrl.replace("tcp://", "");
+        int colonIdx = brokerAddr.lastIndexOf(':');
+        this.brokerHost = brokerAddr.substring(0, colonIdx);
+        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
+        this.mqttUsername = iotProperties.getMqtt().getUsername();
+        this.mqttPassword = iotProperties.getMqtt().getPassword();
+
+        try {
+            checkServerAvailability();
+            String clientId = generateClientId();
+            mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
+            initMqttConnectOptions();
+            setMqttCallback();
+            connectAndSubscribeTopic();
+        } catch (MqttException | InterruptedException e) {
+            log.error("MQTT客户端初始化失败：", e);
+        }
+    }
+
+    private void checkServerAvailability() throws InterruptedException {
+        boolean serverAvailable = false;
+        int attempts = 0;
+        while (!serverAvailable && attempts < MAX_AVAILABILITY_CHECKS) {
+            try (Socket socket = new Socket()) {
+                socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
+                serverAvailable = true;
+                log.info("MQTT服务器连通性检测通过");
+            } catch (Exception e) {
+                attempts++;
+                log.warn("MQTT服务器不可达，5秒后重试... (attempt {}/{}})", attempts, MAX_AVAILABILITY_CHECKS);
+                Thread.sleep(RECONNECT_INTERVAL);
+            }
+        }
+        if (!serverAvailable) {
+            throw new IllegalStateException(
+                "MQTT服务器在" + MAX_AVAILABILITY_CHECKS + "次尝试后仍不可达: " + brokerHost + ":" + brokerPort);
+        }
+    }
+
+    private void initMqttConnectOptions() {
+        connOpts = new MqttConnectOptions();
+        connOpts.setCleanSession(true);
+        connOpts.setAutomaticReconnect(true);
+        connOpts.setConnectionTimeout(10);
+        connOpts.setUserName(mqttUsername);
+        connOpts.setPassword(mqttPassword.toCharArray());
+    }
+
+    private void setMqttCallback() {
+        mqttClient.setCallback(new MqttCallback() {
+            @Override
+            public void connectionLost(Throwable cause) {
+                log.error("MQTT连接断开，开始重连：" + cause.getMessage());
+                isMqttConnected = false;
+                reconnect();
+            }
+
+            @Override
+            public void messageArrived(String topic, MqttMessage message) throws Exception {
+                if (isMqttConnected) {
+                    executorService.submit(() -> {
+                        try {
+                            String messageContent = new String(message.getPayload(), "UTF-8");
+                            handleMessage(topic, messageContent);
+                        } catch (Exception e) {
+                            log.error("消息处理失败：", e);
+                        }
+                    });
+                }
+            }
+
+            @Override
+            public void deliveryComplete(IMqttDeliveryToken token) {
+                // 无需处理
+            }
+        });
+    }
+
+    private void connectAndSubscribeTopic() throws MqttException {
+        if (!mqttClient.isConnected()) {
+            IMqttToken connectToken = mqttClient.connectWithResult(connOpts);
+            if (connectToken.isComplete()) {
+                mqttClient.subscribe(getSubscribeTopic(), QOS);
+                isMqttConnected = true;
+                log.info("MQTT连接成功，已订阅主题：" + getSubscribeTopic());
+            }
+        }
+    }
+
+    public void reconnect() {
+        int maxReconnectAttempts = 3;
+        for (int attempt = 1; attempt <= maxReconnectAttempts; attempt++) {
+            try {
+                Thread.sleep(RECONNECT_INTERVAL);
+                if (mqttClient != null && !mqttClient.isConnected()) {
+                    mqttClient.connect(connOpts);
+                    mqttClient.subscribe(getSubscribeTopic(), QOS);
+                    isMqttConnected = true;
+                    log.info("MQTT重连成功（第" + attempt + "次尝试）");
+                    break;
+                }
+            } catch (MqttException | InterruptedException e) {
+                log.error("MQTT重连失败（第" + attempt + "次尝试）：" + e.getMessage());
+                if (attempt == maxReconnectAttempts) {
+                    log.error("已达最大重连次数，停止重连");
+                }
+            }
+        }
+    }
+
+    @PreDestroy
+    public void disconnect() {
+        try {
+            if (mqttClient != null && mqttClient.isConnected()) {
+                mqttClient.disconnect();
+                mqttClient.close();
+                log.info("MQTT连接已断开");
+            }
+            executorService.shutdown();
+            onDestroy();
+        } catch (MqttException e) {
+            log.error("MQTT断开连接失败：", e);
+        }
+    }
+}

iot-platform/src/main/java/com/iot/platform/mqtt/MqttFaultConsumer.java

@@ -1,30 +1,14 @@
 package com.iot.platform.mqtt;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.iot.platform.common.AjaxResult;
-import com.iot.platform.domain.SysController;
 import com.iot.platform.domain.SysDevice;
 import com.iot.platform.domain.SysFault;
 import com.iot.platform.service.*;
 import com.iot.platform.common.utils.NumericIdGenerator;
-import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import com.iot.platform.config.IotProperties;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.core.io.ClassPathResource;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
-import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.client.RestTemplate;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import javax.net.ssl.*;
-import java.io.InputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.security.KeyStore;
-import java.security.cert.CertificateFactory;
-import java.security.cert.X509Certificate;
 import java.sql.SQLException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -32,293 +16,101 @@ import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * 添加告警信息
  */
 @Component
-public class MqttFaultConsumer {
+public class MqttFaultConsumer extends AbstractMqttConsumer {
 
-    private static final Logger log = LoggerFactory.getLogger(MqttFaultConsumer.class);
-    private static ExecutorService threadPool= Executors.newCachedThreadPool();
     @Autowired
     public SysControllerService sysControllerService;
     @Autowired
-    public MqttDynamicConsumer messageListenerService2;
-    @Autowired
     public SysFaultService sysFaultService;
     @Autowired
     public SysrealtimeService sysrealtimeService;
-
     @Autowired
-    private IotProperties iotProperties;
-
-    // ========== 移除SSL，改用普通TCP连接 ==========
-    private String brokerUrl;
-    private String brokerHost;
-    private int brokerPort;
-
-    private static final int QOS = 1;
-    // 主题（保留原有主题）
-    private static final String SUBSCRIBE_TOPIC = "+/fault_prot";
-    private static final int CONNECT_TIMEOUT = 3000;
-    // 重试时间
-    private static final int RECONNECT_INTERVAL = 5000;
-
-    private String mqttUsername;
-    private String mqttPassword;
-    private static final ExecutorService mysqlWritePool = Executors.newFixedThreadPool(5); // 根据实际情况调整核心线程数
-
-    // 移除：SSL相关的CA_CERT_PATH常量（无需证书配置）
-    // 你的其他成员变量保持不变
+    public SysWorkorderService sysWorkorderService;
+    @Autowired
+    public SysAlarmService sysAlarmService;
     @Autowired
-    public TDengineService tDengineMapceshi2;
+    public NumericIdGenerator numericIdGenerator;
+    @Autowired
+    public TDegnineAlarm tDegnineAlarm;
     @Autowired
     private RestTemplate restTemplate;
 
+    private final ObjectMapper objectMapper = new ObjectMapper();
+    private static final ExecutorService mysqlWritePool = Executors.newFixedThreadPool(5);
 
-    private MqttClient mqttClient;
-    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
-    private boolean isMqttConnected = false;
-    private MqttConnectOptions connOpts;
-    // 注意：移除未定义的threadPool引用（避免编译错误）
-
-    @PostConstruct
-    public void connectAndSubscribe() {
-        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
-        String brokerAddr = this.brokerUrl.replace("tcp://", "");
-        int colonIdx = brokerAddr.lastIndexOf(':');
-        this.brokerHost = brokerAddr.substring(0, colonIdx);
-        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
-        this.mqttUsername = iotProperties.getMqtt().getUsername();
-        this.mqttPassword = iotProperties.getMqtt().getPassword();
-
-        try {
-            // 1. 检测普通MQTT服务器连通性（移除SSL相关，保留基础TCP检测）
-            checkServerAvailability();
-
-            // 2. 创建MQTT客户端（启用普通TCP的brokerUrl，修复原代码注释掉的创建逻辑）
-            String clientId = generateClientIdByOs();
-            mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence()); // 取消注释并修正
-
-            // 3. 初始化MQTT连接选项（移除SSL配置，新增账号密码）
-            initMqttConnectOptions();
-
-            // 4. 设置MQTT回调（保持核心逻辑，修复消息解析bug）
-            setMqttCallback();
-
-            // 5. 建立连接并订阅主题（保持不变）
-            connectAndSubscribeTopic();
-
-        } catch (MqttException | InterruptedException e) {
-            log.error("MQTT客户端初始化失败：", e);
-        }
-    }
-
-    /**
-     * 检测普通MQTT服务器连通性（移除SSL，仅检测TCP端口）
-     */
-    private void checkServerAvailability() throws InterruptedException {
-        boolean serverAvailable = false;
-        while (!serverAvailable) {
-            try (Socket socket = new Socket()) {
-                socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
-                serverAvailable = true;
-                log.info("普通MQTT服务器连通性检测通过");
-            } catch (Exception e) {
-                log.error("普通MQTT服务器不可达，5秒后重试...");
-                Thread.sleep(RECONNECT_INTERVAL);
-            }
-        }
+    private static final Map<String, String> KEY_MAPPING = new HashMap<>();
+    static {
+        KEY_MAPPING.put("timestamp", "devicetimestamp");
+        KEY_MAPPING.put("type", "devicetype");
+        KEY_MAPPING.put("desc", "devicedesc");
     }
 
-    /**
-     * 初始化MQTT连接选项（移除SSL配置，新增账号密码认证）
-     */
-    private void initMqttConnectOptions() {
-        connOpts = new MqttConnectOptions();
-        connOpts.setCleanSession(true);
-        connOpts.setAutomaticReconnect(true);
-        connOpts.setConnectionTimeout(10);
-
-        // ========== 新增：配置MQTT账号密码 ==========
-        connOpts.setUserName(mqttUsername);
-        connOpts.setPassword(mqttPassword.toCharArray()); // 密码要求传入char数组
-
-        // 移除：原有的SSL配置方法调用（configureSslAndStrictHostnameVerify()）
+    @Override
+    protected String getSubscribeTopic() {
+        return "+/fault_prot";
     }
 
-    /**
-     * 按操作系统生成唯一ClientId（保持不变）
-     */
-    private String generateClientIdByOs() {
+    @Override
+    protected String generateClientId() {
         String osName = System.getProperty("os.name").toLowerCase();
         return osName.contains("windows") ? "mqttx_e216fbf1620" : "mqttx_e216fbf1621";
     }
 
-    /**
-     * 设置MQTT回调函数（保留核心逻辑，修复2个关键bug）
-     */
-    private void setMqttCallback() {
-        mqttClient.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.error("MQTT连接断开，开始重连：" + cause.getMessage());
-                isMqttConnected = false;
-                reconnect();
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) {
-                if (isMqttConnected) {
-                    executorService.submit(() -> {
-                        try {
-                            ObjectMapper objectMapper = new ObjectMapper();
-                            // 修复bug1：原代码先读取messageContent，却误用mqtt（message.toString()）解析，改为使用messageContent
-                            // 正确读取消息负载（UTF-8编码，避免乱码）
-                            String messageContent = new String(message.getPayload(), "UTF-8");
-                            Map<String, Object> messageMap = objectMapper.readValue(messageContent, Map.class);
-                            insertTDegine(messageMap, topic);
-                            // 修复bug2：用正确的messageContent解析为SysFault对象
-                            SysFault sysFault = objectMapper.readValue(messageContent, SysFault.class);
-                            // 业务处理
-                            mysqlWritePool.submit(() -> {triggermethod(topic,sysFault);});
-                        } catch (Exception e) {
-                            log.error("消息处理失败：", e);
-                        }
-                    });
-                }
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                // 消息投递完成回调（无需处理）
-            }
-        });
+    @Override
+    protected void handleMessage(String topic, String messageContent) throws Exception {
+        Map<String, Object> messageMap = objectMapper.readValue(messageContent, Map.class);
+        insertTDegine(messageMap, topic);
+        SysFault sysFault = objectMapper.readValue(messageContent, SysFault.class);
+        mysqlWritePool.submit(() -> triggermethod(topic, sysFault));
     }
 
-    /**
-     * 建立MQTT连接并订阅主题（保持不变）
-     */
-    private void connectAndSubscribeTopic() throws MqttException {
-        if (!mqttClient.isConnected()) {
-            IMqttToken connectToken = mqttClient.connectWithResult(connOpts);
-            if (connectToken.isComplete()) {
-                mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                isMqttConnected = true;
-                log.info("MQTT连接成功，已订阅主题：" + SUBSCRIBE_TOPIC);
-            }
-        }
-    }
-
-    /**
-     * MQTT重连逻辑（移除SSL日志标识，保持功能不变）
-     */
-    public void reconnect() {
-        int maxReconnectAttempts = 3;
-        for (int attempt = 1; attempt <= maxReconnectAttempts; attempt++) {
-            try {
-                Thread.sleep(RECONNECT_INTERVAL);
-                if (mqttClient != null && !mqttClient.isConnected()) {
-                    mqttClient.connect(connOpts);
-                    mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                    isMqttConnected = true;
-                    log.info("MQTT重连成功（第" + attempt + "次尝试）");
-                    break; // 重连成功后退出循环
-                }
-            } catch (MqttException | InterruptedException e) {
-                log.error("MQTT重连失败（第" + attempt + "次尝试）：" + e.getMessage());
-                if (attempt == maxReconnectAttempts) {
-                    log.error("已达最大重连次数，停止重连");
-                }
-            }
-        }
-    }
-
-    /**
-     * 销毁时断开MQTT连接，关闭线程池（移除未定义的threadPool.shutdown()）
-     */
-    @PreDestroy
-    public void disconnect() {
-        try {
-            if (mqttClient != null && mqttClient.isConnected()) {
-                mqttClient.disconnect();
-                mqttClient.close();
-                log.info("MQTT连接已断开");
-            }
-            executorService.shutdown();
-            // 移除：原代码中未定义的threadPool.shutdown();（避免编译错误）
-        } catch (MqttException e) {
-            log.error("MQTT断开连接失败：", e);
-        }
-    }
-
-    @Autowired
-    public SysWorkorderService sysWorkorderService;
-    @Autowired
-    public SysAlarmService sysAlarmService;
-    @Autowired
-    public NumericIdGenerator numericIdGenerator;
-    @Autowired
-    public TDegnineAlarm tDegnineAlarm;
-    private static final Map<String, String> KEY_MAPPING = new HashMap<>();
-    static {
-        KEY_MAPPING.put("timestamp", "devicetimestamp");
-        KEY_MAPPING.put("type", "devicetype");
-        KEY_MAPPING.put("desc", "devicedesc");
+    @Override
+    protected void onDestroy() {
+        mysqlWritePool.shutdown();
     }
 
-
     public void insertTDegine(Map<String, Object> weather, String topic) throws SQLException {
         LocalDate localDate = LocalDate.now();
         int year = localDate.getYear();
-        int month = localDate.getMonthValue(); // 直接获取1 - 12的月份，无需加1
-        // 超级表名称
+        int month = localDate.getMonthValue();
         String supertablename = topic.split("/")[0];
-        // 普通表名称
-        String table =topic.split("/")[0] + "_" + year + month;
-        // 创建新Map存储结果，避免修改原Map和并发修改异常
-        Map<String, Object> newMap = new HashMap<>();
+        String table = topic.split("/")[0] + "_" + year + month;
 
-        // 遍历原始Map，替换指定键名
+        Map<String, Object> newMap = new HashMap<>();
         for (Map.Entry<String, Object> entry : weather.entrySet()) {
             String originalKey = entry.getKey();
             Object value = entry.getValue();
-            // 判断是否需要替换键名
             if (KEY_MAPPING.containsKey(originalKey)) {
-                // 使用映射后的新键名
                 newMap.put(KEY_MAPPING.get(originalKey), value);
             } else {
-                // 不需要替换的键名，直接保留
                 newMap.put(originalKey, value);
             }
         }
         tDegnineAlarm.shibaihou(newMap, supertablename, table, topic.split("/")[1]);
     }
 
-
-    public void triggermethod(String topic,SysFault weather){
-        String deviceId=weather.getDevice_id();
-        String timestamp=weather.getTimestamp();
-        String type=weather.getType();
-        Integer code=weather.getCode();
-        String desc=weather.getDesc();
-
+    public void triggermethod(String topic, SysFault weather) {
+        String deviceId = weather.getDevice_id();
+        String timestamp = weather.getTimestamp();
+        String type = weather.getType();
+        String desc = weather.getDesc();
 
         LocalDate localDate = LocalDate.now();
         int year = localDate.getYear();
-        int month = localDate.getMonthValue(); // 直接获取1 - 12的月份，无需加1
+        int month = localDate.getMonthValue();
         String formattedMonth = String.format("%02d", month);
 
-        String controllerId=weather.getController_id();
-        String[] topics=topic.split("/");
+        String controllerId = weather.getController_id();
+        String[] topics = topic.split("/");
 
-        //判读是否有表的存在，如果有就不用管，如果没有就创建这个表（采用按照月去进行分表的操作）
         List<String> tablename = sysrealtimeService.selecttables();
         List<Boolean> a = new ArrayList<>();
-        String controllername=controllerId+year+formattedMonth+"_fault";
+        String controllername = controllerId + year + formattedMonth + "_fault";
 
         for (int i = 0; i < tablename.size(); i++) {
             a.add(tablename.get(i).equals(controllername));
@@ -327,41 +119,27 @@ public class MqttFaultConsumer {
             sysFaultService.createmessage(controllername);
         }
 
-        /**
-         * 修改消息发送修改为只要绑定这个设备或者说是控制器权限的人都可以收到这个故障的信息
-         * （或者说是将已读的状态修改为当前人的信息）
-         */
-        //记录当前经纬度（查询数据库信息）
-        if(type.equals("触发")){
-            //记录触发经纬度
-            SysDevice jingdu=sysControllerService.selectjingweidu(topics[0],"经度");
-            SysDevice weidu=sysControllerService.selectjingweidu(topics[0],"纬度");
-            //随机生成消息id
-            String companyid="GJ"+numericIdGenerator.nextId();
-            //当前时间
+        if ("触发".equals(type)) {
+            SysDevice jingdu = sysControllerService.selectjingweidu(topics[0], "经度");
+            SysDevice weidu = sysControllerService.selectjingweidu(topics[0], "纬度");
+            String companyid = "GJ" + numericIdGenerator.nextId();
             LocalDateTime currentTime = LocalDateTime.now();
             DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
             String currentTimeStr = currentTime.format(formatter);
-            sysAlarmService.insertalarm(controllername,companyid,desc,"0",currentTimeStr,"0",controllerId,deviceId,jingdu.getV(),weidu.getV());
-            //将信息添加到主库中
-            sysFaultService.insertfault(companyid,desc,"0",currentTimeStr,"0",controllerId,deviceId,jingdu.getV(),weidu.getV(),"");
-            //获取当前驾驶人
-            //sysWorkorderService.selectdeviceId(topics[0]);
-        }else if (type.equals("恢复")){
-            //记录触发经纬度
-            SysDevice jingdu=sysControllerService.selectjingweidu(topics[0],"经度");
-            SysDevice weidu=sysControllerService.selectjingweidu(topics[0],"纬度");
-            //随机生成消息id
-            String companyid="GJ"+numericIdGenerator.nextId();
-            //当前时间
+            sysAlarmService.insertalarm(controllername, companyid, desc, "0", currentTimeStr, "0", controllerId, deviceId, jingdu.getV(), weidu.getV());
+            sysFaultService.insertfault(companyid, desc, "0", currentTimeStr, "0", controllerId, deviceId, jingdu.getV(), weidu.getV(), "");
+        } else if ("恢复".equals(type)) {
+            SysDevice jingdu = sysControllerService.selectjingweidu(topics[0], "经度");
+            SysDevice weidu = sysControllerService.selectjingweidu(topics[0], "纬度");
+            String companyid = "GJ" + numericIdGenerator.nextId();
             LocalDateTime currentTime = LocalDateTime.now();
             DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
             String currentTimeStr = currentTime.format(formatter);
-            sysAlarmService.insertalarm(controllername,companyid,desc,"1",currentTimeStr,"0",controllerId,deviceId,jingdu.getV(),weidu.getV());
-            sysFaultService.updatefault("1","0",jingdu.getV(),weidu.getV(),desc,controllerId,deviceId,currentTimeStr);
-
+            sysAlarmService.insertalarm(controllername, companyid, desc, "1", currentTimeStr, "0", controllerId, deviceId, jingdu.getV(), weidu.getV());
+            sysFaultService.updatefault("1", "0", jingdu.getV(), weidu.getV(), desc, controllerId, deviceId, currentTimeStr);
         }
+
         String url = "https://esos-iot.com:9443/syscar/gaojing?controllerId=" + topics[0];
         restTemplate.postForObject(url, null, String.class);
     }
-}
+}

iot-platform/src/main/java/com/iot/platform/mqtt/MqttGenericConsumer.java

@@ -1,45 +1,21 @@
 package com.iot.platform.mqtt;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.iot.platform.domain.ControllerData;
 import com.iot.platform.domain.topics;
 import com.iot.platform.service.SysControllerService;
-import com.iot.platform.service.TDengineService;
-import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import com.iot.platform.config.IotProperties;
 import org.springframework.beans.factory.annotation.Autowired;
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import javax.net.ssl.*;
-import java.io.InputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.security.KeyStore;
-import java.security.cert.CertificateFactory;
-import java.security.cert.X509Certificate;
-import java.util.*;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import org.springframework.core.io.ClassPathResource;
 import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+import java.util.List;
 
 /**
  * 存储控制器数据
  */
 @Component
-public class MqttGenericConsumer {
+public class MqttGenericConsumer extends AbstractMqttConsumer {
 
-    private static final Logger log = LoggerFactory.getLogger(MqttGenericConsumer.class);
-    private static ExecutorService threadPool= Executors.newCachedThreadPool();
-    /**
-     *
-     * 添加mysql数据
-     *
-     */
     @Autowired
     private StringRedisTemplate stringRedisTemplate;
     @Autowired
@@ -47,197 +23,23 @@ public class MqttGenericConsumer {
     @Autowired
     public MqttDynamicConsumer messageListenerService2;
 
-    @Autowired
-    private IotProperties iotProperties;
-
-    // ========== 移除SSL，改用普通TCP连接 ==========
-    private String brokerUrl;
-    private String brokerHost;
-    private int brokerPort;
-
-    private static final int QOS = 1;
-    // 保留原有订阅主题
-    private static final String SUBSCRIBE_TOPIC = "+/generics";
-    private static final int CONNECT_TIMEOUT = 3000;
-    private static final int RECONNECT_INTERVAL = 5000;
-
-    private String mqttUsername;
-    private String mqttPassword;
+    private final ObjectMapper objectMapper = new ObjectMapper();
 
-    // 移除：SSL相关的CA_CERT_PATH常量（无需证书配置）
-    // 你的其他成员变量保持不变
-    @Autowired
-    public TDengineService tDengineMapceshi2;
-    private MqttClient mqttClient;
-    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
-    private boolean isMqttConnected = false;
-    private MqttConnectOptions connOpts;
-    private ObjectMapper objectMapper = new ObjectMapper();
-
-
-    // 移除：未定义的threadPool引用（避免编译错误）
-    @PostConstruct
-    public void connectAndSubscribe() {
-        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
-        String brokerAddr = this.brokerUrl.replace("tcp://", "");
-        int colonIdx = brokerAddr.lastIndexOf(':');
-        this.brokerHost = brokerAddr.substring(0, colonIdx);
-        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
-        this.mqttUsername = iotProperties.getMqtt().getUsername();
-        this.mqttPassword = iotProperties.getMqtt().getPassword();
-
-        try {
-            // 1. 检测普通MQTT服务器连通性（移除SSL，仅检测TCP端口）
-            checkServerAvailability();
-            // 2. 创建MQTT客户端（使用普通TCP的brokerUrl，保持逻辑不变）
-            String clientId = generateClientIdByOs();
-            mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
-            // 3. 初始化MQTT连接选项（移除SSL配置，新增账号密码）
-            initMqttConnectOptions();
-            // 4. 设置MQTT回调（保持核心逻辑，修复消息解析bug）
-            setMqttCallback();
-            // 5. 建立连接并订阅主题（保持逻辑不变，移除SSL日志标识）
-            connectAndSubscribeTopic();
-        } catch (MqttException | InterruptedException e) {
-            log.error("MQTT客户端初始化失败：", e);
-        }
+    @Override
+    protected String getSubscribeTopic() {
+        return "+/generics";
     }
 
-    /**
-     * 检测普通MQTT服务器连通性（移除SSL，仅验证TCP端口可达性）
-     */
-    private void checkServerAvailability() throws InterruptedException {
-        boolean serverAvailable = false;
-        while (!serverAvailable) {
-            try (Socket socket = new Socket()) {
-                // 改用普通brokerHost和brokerPort
-                socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
-                serverAvailable = true;
-                log.info("普通MQTT服务器连通性检测通过");
-            } catch (Exception e) {
-                log.error("普通MQTT服务器不可达，5秒后重试...");
-                Thread.sleep(RECONNECT_INTERVAL);
-            }
-        }
-    }
-
-    /**
-     * 初始化MQTT连接选项（移除SSL配置，新增账号密码认证）
-     */
-    private void initMqttConnectOptions() {
-        connOpts = new MqttConnectOptions();
-        connOpts.setCleanSession(true);
-        connOpts.setAutomaticReconnect(true);
-        connOpts.setConnectionTimeout(10);
-
-        // ========== 新增：配置MQTT账号密码 ==========
-        connOpts.setUserName(mqttUsername);
-        connOpts.setPassword(mqttPassword.toCharArray()); // 密码要求传入char数组
-
-        // 移除：原有的SSL配置方法调用（configureSslAndStrictHostnameVerify()）
-    }
-
-    /**
-     * 按操作系统生成唯一ClientId（保持原有逻辑不变）
-     */
-    private String generateClientIdByOs() {
+    @Override
+    protected String generateClientId() {
         String osName = System.getProperty("os.name").toLowerCase();
         return osName.contains("windows") ? "mqttx_e216fbf1615" : "mqttx_e216fbf1616";
     }
 
-    /**
-     * 设置MQTT回调函数（保留核心逻辑，修复2个关键bug）
-     */
-    private void setMqttCallback() {
-        mqttClient.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.error("MQTT连接断开，开始重连：" + cause.getMessage());
-                isMqttConnected = false;
-                reconnect();
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) throws Exception {
-                if (isMqttConnected) {
-                    executorService.submit(() -> {
-                        try {
-                            ObjectMapper objectMapper = new ObjectMapper();
-                            // 修复bug1：原代码先读取正确的messageContent，却误用mqtt（message.toString()）解析
-                            // 正确读取消息负载（UTF-8编码，避免乱码）
-                            String messageContent = new String(message.getPayload(), "UTF-8");
-                            // 修复bug2：用正确的messageContent解析为ControllerData对象
-                            ControllerData controllerData = objectMapper.readValue(messageContent, ControllerData.class);
-                            // 业务处理
-                            triggermethod(controllerData);
-                        } catch (Exception e) {
-                            log.error("消息处理失败：", e);
-                        }
-                    });
-                }
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                // 消息投递完成回调（无需处理）
-            }
-        });
-    }
-
-    /**
-     * 建立MQTT连接并订阅主题（保持逻辑不变，移除SSL日志标识）
-     */
-    private void connectAndSubscribeTopic() throws MqttException {
-        if (!mqttClient.isConnected()) {
-            IMqttToken connectToken = mqttClient.connectWithResult(connOpts);
-            if (connectToken.isComplete()) {
-                mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                isMqttConnected = true;
-                log.info("MQTT连接成功，已订阅主题：" + SUBSCRIBE_TOPIC);
-            }
-        }
-    }
-
-    /**
-     * MQTT重连逻辑（移除SSL日志标识，保持功能不变）
-     */
-    public void reconnect() {
-        int maxReconnectAttempts = 3;
-        for (int attempt = 1; attempt <= maxReconnectAttempts; attempt++) {
-            try {
-                Thread.sleep(RECONNECT_INTERVAL);
-                if (mqttClient != null && !mqttClient.isConnected()) {
-                    mqttClient.connect(connOpts);
-                    mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                    isMqttConnected = true;
-                    log.info("MQTT重连成功（第" + attempt + "次尝试）");
-                    break; // 重连成功后退出循环
-                }
-            } catch (MqttException | InterruptedException e) {
-                log.error("MQTT重连失败（第" + attempt + "次尝试）：" + e.getMessage());
-                if (attempt == maxReconnectAttempts) {
-                    log.error("已达最大重连次数，停止重连");
-                }
-            }
-        }
-    }
-
-    /**
-     * 销毁时断开MQTT连接，关闭线程池（移除未定义的threadPool.shutdown()）
-     */
-    @PreDestroy
-    public void disconnect() {
-        try {
-            if (mqttClient != null && mqttClient.isConnected()) {
-                mqttClient.disconnect();
-                mqttClient.close();
-                log.info("MQTT连接已断开");
-            }
-            executorService.shutdown();
-            // 移除：原代码中未定义的threadPool.shutdown();（避免编译错误）
-        } catch (MqttException e) {
-            log.error("MQTT断开连接失败：", e);
-        }
+    @Override
+    protected void handleMessage(String topic, String messageContent) throws Exception {
+        ControllerData controllerData = objectMapper.readValue(messageContent, ControllerData.class);
+        triggermethod(controllerData);
     }
 
     public void triggermethod(ControllerData weather) throws Exception {
@@ -247,40 +49,38 @@ public class MqttGenericConsumer {
         List<topics> topics = weather.getTopics();
         List<topics> cmdtopics = weather.getCmd_topics();
         topics faultprot = weather.getFault_prot();
-        //需要检索全部的数据是否存在，如果存储就进行修改，如果不存在就进行添加
-        Integer controllercountcount=0;
+
+        Integer controllercountcount = 0;
         for (topics topicsMap : topics) {
             Integer count = sysControllerService.selectsyscontrollercount(topicsMap.getPath());
             if (count <= 0) {
                 stringRedisTemplate.persist(controllerId);
-                stringRedisTemplate.opsForHash().put(controllerId+":"+topicsMap.getName(), "path", topicsMap.getPath());
-                //将数据存储到redis
+                stringRedisTemplate.opsForHash().put(controllerId + ":" + topicsMap.getName(), "path", topicsMap.getPath());
                 sysControllerService.insertsyscontroller(controllerId, timestamp, fleetId, topicsMap.getName(), topicsMap.getPath());
                 controllercountcount++;
             }
         }
-        Integer controllercountcmdcount=0;
+
+        Integer controllercountcmdcount = 0;
         for (topics cmdtopicsMap : cmdtopics) {
-            //将数据存储到redis
             Integer count = sysControllerService.selectsyscontrollercountcmd(cmdtopicsMap.getPath());
-            if (count<=0) {
-                stringRedisTemplate.opsForHash().put(controllerId+"_cmd:"+cmdtopicsMap.getName(), "path", cmdtopicsMap.getPath());
+            if (count <= 0) {
+                stringRedisTemplate.opsForHash().put(controllerId + "_cmd:" + cmdtopicsMap.getName(), "path", cmdtopicsMap.getPath());
                 stringRedisTemplate.persist(controllerId);
                 sysControllerService.insertsyscontrollercmd(controllerId, timestamp, fleetId, cmdtopicsMap.getName(), cmdtopicsMap.getPath());
                 controllercountcmdcount++;
             }
         }
-        Integer count =sysControllerService.selectsyscontrollercountfault(faultprot.getPath());
-        if (count<=0){
-            stringRedisTemplate.opsForHash().put(controllerId+"_fault:"+faultprot.getName(), "path", faultprot.getPath());
+
+        Integer count = sysControllerService.selectsyscontrollercountfault(faultprot.getPath());
+        if (count <= 0) {
+            stringRedisTemplate.opsForHash().put(controllerId + "_fault:" + faultprot.getName(), "path", faultprot.getPath());
             sysControllerService.insertsyscontrollerfault(controllerId, timestamp, fleetId, faultprot.getName(), faultprot.getPath());
             stringRedisTemplate.persist(controllerId);
         }
-        //如果控制器中的数据出现更新情况，就触发该方法进行重连
-        if (controllercountcount>0) {
-            //关闭连接线程
+
+        if (controllercountcount > 0) {
             messageListenerService2.destroy();
-            //重新建立连接
             messageListenerService2.initMqttConnection();
         }
     }

iot-platform/src/main/java/com/iot/platform/mqtt/MqttStatusConsumer.java

@@ -1,269 +1,65 @@
 package com.iot.platform.mqtt;
+
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.iot.platform.service.SysControllerService;
 import com.iot.platform.service.SysStatusService;
-import com.iot.platform.service.TDengineService;
-import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import com.iot.platform.config.IotProperties;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.core.io.ClassPathResource;
 import org.springframework.data.redis.core.StringRedisTemplate;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import javax.net.ssl.*;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.security.KeyStore;
-import java.security.cert.CertificateFactory;
-import java.security.cert.X509Certificate;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
-import java.util.Collection;
-import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * 存储控制器状态数据
  */
 @Component
-public class MqttStatusConsumer {
-
-    private static final Logger log = LoggerFactory.getLogger(MqttStatusConsumer.class);
+public class MqttStatusConsumer extends AbstractMqttConsumer {
 
-    private static ExecutorService threadPool= Executors.newCachedThreadPool();
     @Autowired
     private StringRedisTemplate stringRedisTemplate;
     @Autowired
     public SysControllerService sysControllerService;
     @Autowired
-    public MqttDynamicConsumer messageListenerService2;
-    @Autowired
     public SysStatusService sysStatusService;
 
-    @Autowired
-    private IotProperties iotProperties;
-
-    // 移除SSL，改用普通TCP连接（对应mqtt://，端口默认1883，可根据你的实际服务端修改）
-    private String brokerUrl;
-    private String brokerHost;
-    private int brokerPort;
-    private static final int QOS = 1;
-    // 主题
-    private static final String SUBSCRIBE_TOPIC = "+/status";
-    private static final int CONNECT_TIMEOUT = 3000;
-    // 重试时间
-    private static final int RECONNECT_INTERVAL = 5000;
-
-    private String mqttUsername;
-    private String mqttPassword;
-
-    // 你的其他成员变量保持不变
-    @Autowired
-    public TDengineService tDengineMapceshi2;
-    private MqttClient mqttClient;
-    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
-    private boolean isMqttConnected = false;
-    private MqttConnectOptions connOpts;
-    // 注意：移除了threadPool的定义（原代码@PreDestroy中存在，若未定义需补充或删除，此处暂删除多余引用）
-
-    @PostConstruct
-    public void connectAndSubscribe() {
-        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
-        String brokerAddr = this.brokerUrl.replace("tcp://", "");
-        int colonIdx = brokerAddr.lastIndexOf(':');
-        this.brokerHost = brokerAddr.substring(0, colonIdx);
-        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
-        this.mqttUsername = iotProperties.getMqtt().getUsername();
-        this.mqttPassword = iotProperties.getMqtt().getPassword();
+    private final ObjectMapper objectMapper = new ObjectMapper();
 
-        try {
-            // 1. 检测普通MQTT服务器连通性（移除SSL相关，保留基础连通性检测）
-            checkServerAvailability();
-
-            // 2. 创建MQTT客户端（直接使用普通TCP的brokerUrl）
-            String clientId = generateClientIdByOs();
-            mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
-
-            // 3. 初始化MQTT连接选项（已移除SSL配置，新增账号密码）
-            initMqttConnectOptions();
-
-            // 4. 设置MQTT回调（保持不变，仅修复消息处理中的小bug）
-            setMqttCallback();
-
-            // 5. 建立连接并订阅主题（保持不变）
-            connectAndSubscribeTopic();
-
-        } catch (MqttException | InterruptedException e) {
-            log.error("MQTT客户端初始化失败：", e);
-        }
-    }
-
-    /**
-     * 检测普通MQTT服务器连通性（移除SSL相关，仅检测TCP端口）
-     */
-    private void checkServerAvailability() throws InterruptedException {
-        boolean serverAvailable = false;
-        while (!serverAvailable) {
-            try (Socket socket = new Socket()) {
-                socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
-                serverAvailable = true;
-                log.info("普通MQTT服务器连通性检测通过");
-            } catch (Exception e) {
-                log.error("普通MQTT服务器不可达，5秒后重试...");
-                Thread.sleep(RECONNECT_INTERVAL);
-            }
-        }
+    @Override
+    protected String getSubscribeTopic() {
+        return "+/status";
     }
 
-    /**
-     * 初始化MQTT连接选项（移除SSL配置，新增账号密码认证）
-     */
-    private void initMqttConnectOptions() {
-        connOpts = new MqttConnectOptions();
-        connOpts.setCleanSession(true);
-        connOpts.setAutomaticReconnect(true);
-        connOpts.setConnectionTimeout(10);
-
-        // ========== 新增：配置MQTT账号密码 ==========
-        connOpts.setUserName(mqttUsername);
-        connOpts.setPassword(mqttPassword.toCharArray()); // 密码要求传入char数组
-
-        // 移除：原有的SSL配置方法调用（configureSslAndStrictHostnameVerify()）
-    }
-
-    /**
-     * 按操作系统生成唯一ClientId（保持不变）
-     */
-    private String generateClientIdByOs() {
+    @Override
+    protected String generateClientId() {
         String osName = System.getProperty("os.name").toLowerCase();
         return osName.contains("windows") ? "mqttx_e216fbf1613" : "mqttx_e216fbf1614";
     }
 
-    /**
-     * 设置MQTT回调函数（保持不变，修复消息处理中的bug：误用mqtt变量，应使用messageContent）
-     */
-    private void setMqttCallback() {
-        mqttClient.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.error("MQTT连接断开，开始重连：" + cause.getMessage());
-                isMqttConnected = false;
-                reconnect();
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) throws Exception {
-                if (isMqttConnected) {
-                    executorService.submit(() -> {
-                        try {
-                            // 修复bug：原代码先定义messageContent，后误用mqtt变量（未赋值，实际是message.toString()）
-                            ObjectMapper objectMapper = new ObjectMapper();
-                            // 正确读取消息负载（UTF-8编码）
-                            String messageContent = new String(message.getPayload(), "UTF-8");
-                            // 改用messageContent解析，避免空指针或解析错误
-                            Map<String, Object> listOfMaps = objectMapper.readValue(messageContent, Map.class);
-                            // 业务处理
-                            triggermethod(listOfMaps);
-                        } catch (Exception e) {
-                            log.error("消息处理失败：", e);
-                        }
-                    });
-                }
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                // 消息投递完成回调（无需处理）
-            }
-        });
+    @Override
+    protected void handleMessage(String topic, String messageContent) throws Exception {
+        Map<String, Object> weather = objectMapper.readValue(messageContent, Map.class);
+        triggermethod(weather);
     }
 
-    /**
-     * 建立MQTT连接并订阅主题（保持不变）
-     */
-    private void connectAndSubscribeTopic() throws MqttException {
-        if (!mqttClient.isConnected()) {
-            IMqttToken connectToken = mqttClient.connectWithResult(connOpts);
-            if (connectToken.isComplete()) {
-                mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                isMqttConnected = true;
-                log.info("MQTT连接成功，已订阅主题：" + SUBSCRIBE_TOPIC);
-            }
-        }
-    }
+    public void triggermethod(Map<String, Object> weather) throws Exception {
+        String controllerId = weather.get("controller_id").toString();
+        String fleetId = weather.get("fleet_id").toString();
+        String status = weather.get("status").toString();
 
-    /**
-     * MQTT重连逻辑（保持不变）
-     */
-    public void reconnect() {
-        int maxReconnectAttempts = 3;
-        for (int attempt = 1; attempt <= maxReconnectAttempts; attempt++) {
-            try {
-                Thread.sleep(RECONNECT_INTERVAL);
-                if (mqttClient != null && !mqttClient.isConnected()) {
-                    mqttClient.connect(connOpts);
-                    mqttClient.subscribe(SUBSCRIBE_TOPIC, QOS);
-                    isMqttConnected = true;
-                    log.info("MQTT重连成功（第" + attempt + "次尝试）");
-                    break; // 重连成功后退出循环
-                }
-            } catch (MqttException | InterruptedException e) {
-                log.error("MQTT重连失败（第" + attempt + "次尝试）：" + e.getMessage());
-                if (attempt == maxReconnectAttempts) {
-                    log.error("已达最大重连次数，停止重连");
-                }
-            }
-        }
-    }
-
-    /**
-     * 销毁时断开MQTT连接，关闭线程池（移除threadPool多余引用）
-     */
-    @PreDestroy
-    public void disconnect() {
-        try {
-            if (mqttClient != null && mqttClient.isConnected()) {
-                mqttClient.disconnect();
-                mqttClient.close();
-                log.info("MQTT连接已断开");
-            }
-            executorService.shutdown();
-            // 移除：原代码中的threadPool.shutdown();（若未定义该变量，直接删除，避免编译错误）
-        } catch (MqttException e) {
-            log.error("MQTT断开连接失败：", e);
-        }
-    }
-
-//    @Async
-    public void triggermethod(Map<String,Object> weather) throws Exception {
-        String controllerId=weather.get("controller_id").toString();
-        String fleetId=weather.get("fleet_id").toString();
-        String status=weather.get("status").toString();
-        //当前时间
         LocalDateTime currentTime = LocalDateTime.now();
         DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
         String currentTimeStr = currentTime.format(formatter);
-        //添加数据到超级表中
-        stringRedisTemplate.opsForHash().put(controllerId + "status:", "fleet_id",weather.get("fleet_id").toString());
-        stringRedisTemplate.opsForHash().put(controllerId + "status:", "status",weather.get("status").toString());
-        //查询数据库中是否存在该条数据
-        Integer count=sysStatusService.selectstatuscount(controllerId);
-        if (count<=0){
-            //添加数据到mysql数据库中
-            sysStatusService.insertsysstatus(controllerId,fleetId,status,currentTimeStr);
-        }else{
-            sysStatusService.updatestatus(controllerId,fleetId,status,currentTimeStr);
+
+        stringRedisTemplate.opsForHash().put(controllerId + "status:", "fleet_id", fleetId);
+        stringRedisTemplate.opsForHash().put(controllerId + "status:", "status", status);
+
+        Integer count = sysStatusService.selectstatuscount(controllerId);
+        if (count <= 0) {
+            sysStatusService.insertsysstatus(controllerId, fleetId, status, currentTimeStr);
+        } else {
+            sysStatusService.updatestatus(controllerId, fleetId, status, currentTimeStr);
         }
     }
-
 }

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

@@ -1,71 +1,64 @@
 package com.iot.platform.service;
 
 import com.iot.platform.mapper.SysrealtimeMapper;
-import org.apache.ibatis.annotations.Param;
+import org.apache.commons.lang3.StringUtils;
 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;
+import java.util.regex.Pattern;
 
 @Service
 public class SysrealtimeService {
     private static final Logger log = LoggerFactory.getLogger(SysrealtimeService.class);
+    private static final Pattern TABLE_NAME_PATTERN = Pattern.compile("^[a-zA-Z_][a-zA-Z0-9_]*$");
+    private static final int MAX_TABLE_NAME_LENGTH = 64;
 
     @Resource
     public SysrealtimeMapper sysrealtimeMapper;
 
     public void createrealtime(String tableName) {
+        validateTableName(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);
+        sysrealtimeMapper.createrealtime(tableName);
+        log.info("表[{}]创建成功", tableName);
     }
 
-    public List<String> selecttables(){
+    public List<String> selecttables() {
         return sysrealtimeMapper.selecttables();
     }
-    public void inserttables(String tableName,String createTime,String deviceId,String timestamp,String k,Object v){
+
+    public void inserttables(String tableName, String createTime, String deviceId, String timestamp, String k, Object v) {
+        validateTableName(tableName);
         sysrealtimeMapper.inserttables(tableName, createTime, deviceId, timestamp, k, v);
     }
-    public void updatetables(String tableName,String createTime,Object v,String timestamp,String k,String deviceId){
+
+    public void updatetables(String tableName, String createTime, Object v, String timestamp, String k, String deviceId) {
+        validateTableName(tableName);
         sysrealtimeMapper.updatetables(tableName, createTime, v, timestamp, k, deviceId);
     }
 
-    public Integer selectkey(String tableName,String k){
+    public Integer selectkey(String tableName, String k) {
+        validateTableName(tableName);
         return sysrealtimeMapper.selectkey(tableName, k);
     }
 
+    public List<String> selectAllKeys(String tableName) {
+        validateTableName(tableName);
+        return sysrealtimeMapper.selectAllKeys(tableName);
+    }
+
+    private void validateTableName(String tableName) {
+        if (StringUtils.isBlank(tableName)) {
+            throw new IllegalArgumentException("表名不能为空");
+        }
+        if (tableName.length() > MAX_TABLE_NAME_LENGTH) {
+            throw new IllegalArgumentException("表名过长，最大" + MAX_TABLE_NAME_LENGTH + "字符");
+        }
+        if (!TABLE_NAME_PATTERN.matcher(tableName).matches()) {
+            throw new IllegalArgumentException("非法表名: " + tableName);
+        }
+    }
 }

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

@@ -8,7 +8,10 @@ import com.iot.platform.service.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
@@ -29,8 +32,6 @@ public class VehicleSyncTask {
     @Autowired
     private StringRedisTemplate stringRedisTemplate;
     @Autowired
-    private RestTemplate restTemplate;
-    @Autowired
     private SysrealtimeService sysrealtimeService;
     @Autowired
     public SysDeviceVoService sysDeviceVoService;
@@ -43,12 +44,21 @@ public class VehicleSyncTask {
     @Autowired
     public SysCompanyService sysCompanyService;
 
+    private final RestTemplate restTemplate;
+
+    public VehicleSyncTask() {
+        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
+        factory.setConnectTimeout(5000);
+        factory.setReadTimeout(10000);
+        this.restTemplate = new RestTemplate(factory);
+    }
+
     /**
      * 更新车辆的控制器信息
      * 30秒更新一次
      */
     @Scheduled(fixedDelay = 30000)
-    public void updatesyscar() {
+    public void updateSysCar() {
         List<SysCar> sysCarList = sysCarService.selectcontrollerId();
         for (SysCar sysCar : sysCarList) {
             if (sysCar.getControllerId() == null || sysCar.getControllerId().isEmpty()) {
@@ -58,39 +68,58 @@ public class VehicleSyncTask {
             SysDevice longitude = sysDeviceService.selectsysdevice(sysCar.getControllerId(), "经度");
 
             String redisKeyPattern = "workorder:coordinate:" + sysCar.getControllerId() + ":*";
-            Set<String> keys = stringRedisTemplate.keys(redisKeyPattern);
+            Set<String> keys = scanKeys(redisKeyPattern);
 
             if (keys == null || keys.isEmpty()) {
-                String redisKey = "workorder:coordinate:" + sysCar.getControllerId();
-                stringRedisTemplate.opsForHash().put(redisKey, "latitude", latitude.getV());
-                stringRedisTemplate.opsForHash().put(redisKey, "longitude", longitude.getV());
-                stringRedisTemplate.persist(redisKey);
-
-                String position = latitude.getV() + "," + longitude.getV();
-                sysCarService.updatecarposition(position, sysCar.getCarId());
-                String url = "https://esos-iot.com:9443/syscar/trigger?carId=" + sysCar.getCarId();
-                restTemplate.postForObject(url, null, String.class);
+                updateCarPosition(sysCar, latitude, longitude);
             } else {
                 for (String key : keys) {
                     Map<Object, Object> coordinateMap = stringRedisTemplate.opsForHash().entries(key);
                     if (coordinateMap.get("latitude").equals(latitude.getV()) && coordinateMap.get("longitude").equals(longitude.getV())) {
                         continue;
                     }
-                    String redisKey = "workorder:coordinate:" + sysCar.getControllerId();
-                    stringRedisTemplate.opsForHash().put(redisKey, "latitude", latitude.getV());
-                    stringRedisTemplate.opsForHash().put(redisKey, "longitude", longitude.getV());
-                    stringRedisTemplate.persist(redisKey);
-                    String position = latitude.getV() + "," + longitude.getV();
-                    sysCarService.updatecarposition(position, sysCar.getCarId());
-                    String url = "https://esos-iot.com:9443/syscar/trigger?carId=" + sysCar.getCarId();
-                    restTemplate.postForObject(url, null, String.class);
+                    updateCarPosition(sysCar, latitude, longitude);
                 }
             }
         }
     }
 
+    private void updateCarPosition(SysCar sysCar, SysDevice latitude, SysDevice longitude) {
+        String redisKey = "workorder:coordinate:" + sysCar.getControllerId();
+        stringRedisTemplate.opsForHash().put(redisKey, "latitude", latitude.getV());
+        stringRedisTemplate.opsForHash().put(redisKey, "longitude", longitude.getV());
+        stringRedisTemplate.persist(redisKey);
+
+        String position = latitude.getV() + "," + longitude.getV();
+        sysCarService.updatecarposition(position, sysCar.getCarId());
+        String url = "https://esos-iot.com:9443/syscar/trigger?carId=" + sysCar.getCarId();
+        try {
+            restTemplate.postForObject(url, null, String.class);
+        } catch (Exception e) {
+            log.warn("触发webhook失败 carId={}: {}", sysCar.getCarId(), e.getMessage());
+        }
+    }
+
+    private Set<String> scanKeys(String pattern) {
+        Set<String> keys = new HashSet<>();
+        ScanOptions options = ScanOptions.scanOptions().match(pattern).count(100).build();
+        try {
+            stringRedisTemplate.execute((RedisCallback<Void>) connection -> {
+                org.springframework.data.redis.core.Cursor<byte[]> cursor = connection.scan(options);
+                while (cursor.hasNext()) {
+                    keys.add(new String(cursor.next()));
+                }
+                cursor.close();
+                return null;
+            });
+        } catch (Exception e) {
+            log.error("Redis SCAN失败 pattern={}: {}", pattern, e.getMessage());
+        }
+        return keys;
+    }
+
     @Scheduled(fixedDelay = 30000)
-    public void insertdevice() {
+    public void insertDevice() {
         Set<String> activeKeys;
         try {
             activeKeys = stringRedisTemplate.opsForSet().members("DSB:active:devices");
@@ -103,6 +132,7 @@ public class VehicleSyncTask {
                 log.info("redis中无数据");
                 return;
             }
+            List<SysDeviceControl> sysDeviceControlList = sysDeviceControlService.selectdevice("device");
             for (String redisKey : activeKeys) {
                 Map<Object, Object> dataMap = stringRedisTemplate.opsForHash().entries(redisKey);
                 if (dataMap == null || dataMap.isEmpty()) {
@@ -115,7 +145,6 @@ public class VehicleSyncTask {
                     continue;
                 }
                 String controllerId = parts[1];
-                List<SysDeviceControl> sysDeviceControlList = sysDeviceControlService.selectdevice("device");
                 Integer count = sysDeviceVoService.selectcount(controllerId);
                 if (count != null && count > 0) {
                     StringBuilder keyvalue = new StringBuilder();
@@ -164,6 +193,7 @@ public class VehicleSyncTask {
     public void syncRedisToMySQL() {
         Set<String> activeKeys = stringRedisTemplate.opsForSet().members("DSB:active:devices");
         if (activeKeys == null || activeKeys.isEmpty()) return;
+
         for (String redisKey : activeKeys) {
             try {
                 Map<Object, Object> dataMap = stringRedisTemplate.opsForHash().entries(redisKey);
@@ -178,15 +208,14 @@ 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();
+                    log.error("创建表失败: {} | {}", controllerId, e.getMessage(), e);
                     continue;
                 }
+
                 String createTime = getStringValue(dataMap, "createTime");
                 String timestamp = getStringValue(dataMap, "timestamp");
                 String deviceId = getStringValue(dataMap, "device_id");
@@ -194,6 +223,9 @@ public class VehicleSyncTask {
                     continue;
                 }
 
+                List<String> existingKeys = sysrealtimeService.selectAllKeys(controllerId);
+                Set<String> existingKeySet = existingKeys != null ? new HashSet<>(existingKeys) : Collections.emptySet();
+
                 for (Map.Entry<Object, Object> entry : dataMap.entrySet()) {
                     String fieldKey = entry.getKey().toString();
                     if ("createTime".equals(fieldKey) || "timestamp".equals(fieldKey) || "device_id".equals(fieldKey)) {
@@ -201,15 +233,15 @@ public class VehicleSyncTask {
                     }
                     String fieldValue = getStringValue(dataMap, fieldKey);
                     if (fieldValue == null) continue;
-                    Integer count = sysrealtimeService.selectkey(controllerId, fieldKey);
-                    if (count != null && count > 0) {
+
+                    if (existingKeySet.contains(fieldKey)) {
                         sysrealtimeService.updatetables(controllerId, createTime, fieldValue, timestamp, fieldKey, deviceId);
                     } else {
                         sysrealtimeService.inserttables(controllerId, createTime, deviceId, timestamp, fieldKey, fieldValue);
                     }
                 }
             } catch (Exception e) {
-                log.error("同步设备失败: {} | {}", redisKey, e.getMessage());
+                log.error("同步设备失败: {} | {}", redisKey, e.getMessage(), e);
             }
         }
     }
@@ -224,7 +256,7 @@ public class VehicleSyncTask {
      * 根据公司去查询
      */
     @Scheduled(fixedDelay = 30000)
-    public void insertindicators() {
+    public void insertIndicators() {
         try {
             LocalDate today = LocalDate.now();
             DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

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

@@ -7,15 +7,15 @@ spring:
             # 主库数据源
             master:
                 url: jdbc:mysql://47.104.204.180:3306/data?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
-                username: root
-                password: Zhu059300()__
+                username: ${MYSQL_USERNAME:root}
+                password: ${MYSQL_PASSWORD:Zhu059300()__}
             # 从库数据源
             slave:
                 # 从数据源开关/默认关闭
                 enabled: true
                 url: jdbc:mysql://47.104.204.180:3306/cnc?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
-                username: root
-                password: Zhu059300()__
+                username: ${MYSQL_USERNAME:root}
+                password: ${MYSQL_PASSWORD:Zhu059300()__}
             # 初始连接数
             initialSize: 5
             # 最小连接池数量
@@ -42,13 +42,13 @@ spring:
             webStatFilter:
                 enabled: true
             statViewServlet:
-                enabled: true
+                enabled: ${DRUID_STAT_ENABLED:false}
                 # 设置白名单，不填则允许所有访问
                 allow:
                 url-pattern: /druid/*
                 # 控制台管理用户名和密码
-                login-username: ruoyi
-                login-password: 123456
+                login-username: ${DRUID_USERNAME:ruoyi}
+                login-password: ${DRUID_PASSWORD:123456}
             filter:
                 stat:
                     enabled: true

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

@@ -32,7 +32,7 @@ spring:
     host: localhost
     port: 6379
     database: 0
-    password:
+    password: ${REDIS_PASSWORD:}
     timeout: 10s
     lettuce:
       pool:
@@ -52,13 +52,23 @@ pagehelper:
   supportMethodsArguments: true
   params: count=countSql
 
+# Actuator 配置
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,info
+  endpoint:
+    health:
+      show-details: always
+
 # IoT平台配置
 iot:
   mqtt:
     broker-url: tcp://47.104.204.180:1883
-    username: NjniyrEO
-    password: 2b577892f4824d466dbc323a1ee4dfe1902c55bb
+    username: ${MQTT_USERNAME:NjniyrEO}
+    password: ${MQTT_PASSWORD:2b577892f4824d466dbc323a1ee4dfe1902c55bb}
   tdengine:
     url: jdbc:TAOS://localhost:6030/
-    username: root
-    password: taosdata
+    username: ${TDENGINE_USERNAME:root}
+    password: ${TDENGINE_PASSWORD:taosdata}

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

@@ -37,4 +37,8 @@
     <select id="selectkey" resultType="Integer">
         select COUNT(*) from `${tableName}` where k=#{k}
     </select>
+
+    <select id="selectAllKeys" resultType="String">
+        SELECT k FROM `${tableName}`
+    </select>
 </mapper>

iot-platform/src/test/java/com/iot/platform/common/RedisKeysTest.java

@@ -0,0 +1,57 @@
+package com.iot.platform.common;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RedisKeysTest {
+
+    @Test
+    @DisplayName("ACTIVE_DEVICES 应包含统一前缀")
+    void activeDevices_containsPrefix() {
+        assertThat(RedisKeys.ACTIVE_DEVICES).isEqualTo("iot:dsb:active:devices");
+    }
+
+    @Test
+    @DisplayName("deviceTelemetry 应生成正确的 key 格式")
+    void deviceTelemetry_formatsCorrectly() {
+        String key = RedisKeys.deviceTelemetry("ctrl001", "temperature");
+        assertThat(key).isEqualTo("iot:dsb:ctrl001:temperature");
+    }
+
+    @Test
+    @DisplayName("workorderCoordinate 应生成正确的 key 格式")
+    void workorderCoordinate_formatsCorrectly() {
+        String key = RedisKeys.workorderCoordinate("ctrl001");
+        assertThat(key).isEqualTo("iot:workorder:coordinate:ctrl001");
+    }
+
+    @Test
+    @DisplayName("controllerStatus 应生成正确的 key 格式")
+    void controllerStatus_formatsCorrectly() {
+        String key = RedisKeys.controllerStatus("ctrl001");
+        assertThat(key).isEqualTo("iot:controller:ctrl001:status");
+    }
+
+    @Test
+    @DisplayName("controllerTopic 应生成正确的 key 格式")
+    void controllerTopic_formatsCorrectly() {
+        String key = RedisKeys.controllerTopic("ctrl001", "metrics");
+        assertThat(key).isEqualTo("iot:controller:ctrl001:metrics");
+    }
+
+    @Test
+    @DisplayName("controllerCmdTopic 应生成正确的 key 格式")
+    void controllerCmdTopic_formatsCorrectly() {
+        String key = RedisKeys.controllerCmdTopic("ctrl001", "cmd1");
+        assertThat(key).isEqualTo("iot:controller:ctrl001_cmd:cmd1");
+    }
+
+    @Test
+    @DisplayName("controllerFaultTopic 应生成正确的 key 格式")
+    void controllerFaultTopic_formatsCorrectly() {
+        String key = RedisKeys.controllerFaultTopic("ctrl001", "fault1");
+        assertThat(key).isEqualTo("iot:controller:ctrl001_fault:fault1");
+    }
+}

iot-platform/src/test/java/com/iot/platform/mqtt/MqttGenericConsumerTest.java

@@ -0,0 +1,24 @@
+package com.iot.platform.mqtt;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MqttGenericConsumerTest {
+
+    @Test
+    @DisplayName("getSubscribeTopic 应返回 +/generics")
+    void getSubscribeTopic_returnsCorrectValue() {
+        MqttGenericConsumer consumer = new MqttGenericConsumer();
+        assertThat(consumer.getSubscribeTopic()).isEqualTo("+/generics");
+    }
+
+    @Test
+    @DisplayName("generateClientId 应包含 mqttx 前缀")
+    void generateClientId_containsPrefix() {
+        MqttGenericConsumer consumer = new MqttGenericConsumer();
+        String clientId = consumer.generateClientId();
+        assertThat(clientId).startsWith("mqttx_e216fbf16");
+    }
+}

iot-platform/src/test/java/com/iot/platform/service/SysrealtimeServiceTest.java

@@ -0,0 +1,83 @@
+package com.iot.platform.service;
+
+import com.iot.platform.mapper.SysrealtimeMapper;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+class SysrealtimeServiceTest {
+
+    @Mock
+    private SysrealtimeMapper sysrealtimeMapper;
+
+    @InjectMocks
+    private SysrealtimeService service;
+
+    @Test
+    @DisplayName("createrealtime: 空表名应抛出 IllegalArgumentException")
+    void createrealtime_nullTableName_throws() {
+        assertThatThrownBy(() -> service.createrealtime(null))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("表名不能为空");
+    }
+
+    @Test
+    @DisplayName("createrealtime: SQL注入表名应抛出 IllegalArgumentException")
+    void createrealtime_sqlInjectionTableName_throws() {
+        assertThatThrownBy(() -> service.createrealtime("users; DROP TABLE users;"))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("非法表名");
+    }
+
+    @Test
+    @DisplayName("createrealtime: 超长表名应抛出 IllegalArgumentException")
+    void createrealtime_tooLongTableName_throws() {
+        assertThatThrownBy(() -> service.createrealtime("a".repeat(65)))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("表名过长");
+    }
+
+    @Test
+    @DisplayName("createrealtime: 合法表名应调用 mapper")
+    void createrealtime_validTableName_callsMapper() {
+        service.createrealtime("device_data_202401");
+        verify(sysrealtimeMapper).createrealtime("device_data_202401");
+    }
+
+    @Test
+    @DisplayName("inserttables: 非法表名应抛出异常且不调用 mapper")
+    void inserttables_invalidTableName_throwsWithoutCallingMapper() {
+        assertThatThrownBy(() -> service.inserttables("bad;name", "2024-01-01", "D1", "ts", "k", "v"))
+            .isInstanceOf(IllegalArgumentException.class);
+        verifyNoInteractions(sysrealtimeMapper);
+    }
+
+    @Test
+    @DisplayName("updatetables: 合法表名应调用 mapper")
+    void updatetables_validTableName_callsMapper() {
+        service.updatetables("dev_001", "2024-01-01", "v1", "ts", "key1", "D1");
+        verify(sysrealtimeMapper).updatetables("dev_001", "2024-01-01", "v1", "ts", "key1", "D1");
+    }
+
+    @Test
+    @DisplayName("selectkey: 非法表名应抛出异常")
+    void selectkey_invalidTableName_throws() {
+        assertThatThrownBy(() -> service.selectkey("--comment", "k"))
+            .isInstanceOf(IllegalArgumentException.class)
+            .hasMessageContaining("非法表名");
+    }
+
+    @Test
+    @DisplayName("selectAllKeys: 合法表名应调用 mapper")
+    void selectAllKeys_validTableName_callsMapper() {
+        service.selectAllKeys("realtime_001");
+        verify(sysrealtimeMapper).selectAllKeys("realtime_001");
+    }
+}

start.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+# IoT 平台启动脚本 — 自动加载 .env 环境变量后启动
+
+APP_DIR="$(cd "$(dirname "$0")" && pwd)"
+cd "$APP_DIR"
+
+if [ -f ".env" ]; then
+    set -a
+    source .env
+    set +a
+    echo "[start.sh] 已加载 .env 环境变量"
+else
+    echo "[start.sh] 警告: 未找到 .env 文件"
+fi
+
+./ry.sh "$@"