Vue3 + Spring Boot + 表格存储：物联网数据平台的全栈实战

摘要：5000+ 传感器每秒上报 10 万条数据，MySQL 写入延迟飙到 5 秒、查询频繁超时——这是我在某智慧园区项目中亲历的 IoT 数据平台之痛。迁移到阿里云表格存储后，写入 TPS 提升 100 倍，时序查询秒级返回，存储成本降低 60%。本文从 IoT 数据的 5 大痛点出发，详解 Vue3 设备看板 + Spring Boot + 阿里云物联网平台 + 表格存储 + DataHub + Flink 的全栈架构与代码实战，覆盖设备管理、数据采集链路、多元索引查询、时序聚合、告警规则引擎、前端实时看板、冷热数据分层，并给出 5 个生产踩坑实录和 IoT 数据方案选型决策树。

1. 场景：10 万 TPS 写入压垮 MySQL 的至暗时刻

2025 年 Q2，我负责的智慧园区项目接入 5000+ 传感器（温湿度、烟雾、门禁、电表），峰值每秒上报 10 万条数据。上线第一周，数据库就炸了：MySQL 写入延迟从 50ms 飙到 5 秒，时序查询动辄超时，磁盘 I/O 打满，整个园区监控看板直接白屏。

最严重的一次：园区消防报警数据因写入排队延迟 8 秒才入库，导致告警触发不及时。虽然最终没有造成实际损失，但这个问题让我后怕不已——消防数据延迟 8 秒，在真实火情面前就是生命危险。

迁移到阿里云表格存储后一个季度的数据对比：

下面从 IoT 数据的核心痛点说起，逐层展开全栈实战方案。

2. IoT 数据的 5 大痛点

物联网数据的本质是"高并发写入 + 海量时序查询"，传统关系型数据库在这个场景下处处碰壁。

痛点 1：写入高并发，RDBMS 撑不住

5000 个传感器，每秒 10 万条上报，MySQL 即使分库分表也扛不住这个写入量。InnoDB 的 B+ 树每次写入都要刷盘，高并发下锁竞争严重，TPS 天花板明显。

痛点 2：时序查询慢，聚合计算超时

IoT 场景最典型的查询：某设备最近 24 小时温度曲线、某楼层平均湿度趋势、某区域异常设备统计。这类查询涉及大范围时间扫描 + 聚合计算，MySQL 全表扫描动辄十几秒。

痛点 3：数据量暴涨，存储成本失控

传感器数据 7×24 不停歇，一年下来几十亿条记录。MySQL 存储成本高，3 个月就得清理历史数据，但业务方又要求保留更长时间用于趋势分析。

痛点 4：多维度聚合，索引设计困难

查询维度组合爆炸：按设备、按楼层、按区域、按设备类型、按时间粒度，每种组合都需要不同索引。MySQL 无法为所有维度组合建索引，要么查询慢，要么索引空间爆炸。

痛点 5：冷热数据分离成本高

最近 7 天的数据是热数据，查询频繁；3 个月前的数据是冷数据，偶尔查询。MySQL 没有原生的冷热分离能力，冷数据和热数据混存，存储和查询互相拖累。

核心思路：用阿里云物联网平台统一设备接入，DataHub 做数据管道，表格存储做时序存储和查询引擎，Flink 做实时计算，Vue3 做可视化看板。各司其职，组合出完整的 IoT 数据方案。

3. 架构设计

3.1 整体架构

3.2 数据采集与查询时序

设备上报到前端展示的完整数据链路：

4. 阿里云资源配置

在开始编码前，需要完成三项阿里云资源的创建和配置。

4.1 物联网平台实例

登录阿里云控制台 → 物联网平台 → 创建企业版实例：

4.2 表格存储实例

登录阿里云控制台 → 表格存储 → 创建实例：

创建两张表：

表1: iot_device_data（时序表）
  主键: [device_id(String), timestamp(Integer)]
  属性列: temperature, humidity, smoke, voltage, status, ...

表2: iot_device_meta（关系表）
  主键: [device_id(String)]
  属性列: device_name, product_key, floor, area, type, online, last_report_time, ...

4.3 DataHub Topic

登录阿里云控制台 → DataHub → 创建 Project 和 Topic：

Project: iot_data_platform
Topic: iot_raw_data
  类型: TUPLE
  Schema:
    device_id    STRING
    timestamp    BIGINT
    temperature  DOUBLE
    humidity     DOUBLE
    smoke        DOUBLE
    voltage      DOUBLE
    status       STRING
  生命周期: 7天
  分区数: 4（匹配写入并发）

5. 后端 Spring Boot 实战

5.1 项目结构与依赖

项目采用分层架构，核心模块包括设备管理、数据查询、告警引擎：

iot-data-platform/
├── pom.xml
└── src/main/java/com/example/iot/
    ├── config/
    │   ├── OtsConfig.java
    │   ├── DataHubConfig.java
    │   └── IotClientConfig.java
    ├── controller/
    │   ├── DeviceController.java
    │   ├── DataQueryController.java
    │   └── AlertController.java
    ├── service/
    │   ├── DeviceService.java
    │   ├── DataQueryService.java
    │   └── AlertEngineService.java
    ├── model/
    │   ├── Device.java
    │   ├── SensorData.java
    │   └── AlertRule.java
    └── IotDataPlatformApplication.java

Maven 依赖中需要引入表格存储 SDK 和 IoT 平台 SDK：

<!-- 表格存储 SDK -->
<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>tablestore</artifactId>
    <version>5.17.0</version>
</dependency>

<!-- 阿里云IoT平台 SDK -->
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-iot</artifactId>
    <version>7.52.0</version>
</dependency>

<!-- DataHub SDK -->
<dependency>
    <groupId>com.aliyun.datahub</groupId>
    <artifactId>aliyun-sdk-datahub</artifactId>
    <version>2.26.0</version>
</dependency>

5.2 IoT 平台设备管理

设备管理是 IoT 数据平台的起点，包括产品创建、物模型定义和设备注册。

为什么先定义物模型？物模型是设备与平台之间的数据契约，决定了传感器上报数据的格式和校验规则，后续的规则引擎转发、数据解析都依赖它。

@Service
@Slf4j
public class DeviceService {
   

    @Autowired
    private IAcsClient iotClient;

    /**
     * 创建产品（传感器类型分组）
     */
    public String createProduct(String productName, String nodeType) {
   
        CreateProductRequest request = new CreateProductRequest();
        request.setProductName(productName);
        request.setNodeType(nodeType); // 0=直连设备
        request.setDataFormat("JSON");
        request.setProtocolType("MQTT");
        request.setDescription("智慧园区-" + productName);

        try {
   
            CreateProductResponse response = iotClient
                .getAcsResponse(request);
            log.info("产品创建成功: productKey={}",
                response.getProductKey());
            return response.getProductKey();
        } catch (Exception e) {
   
            throw new RuntimeException("产品创建失败", e);
        }
    }

    /**
     * 定义物模型（温湿度传感器）
     */
    public void defineThingModel(String productKey) {
   
        // 温度属性
        String thingModelJson = """
        {
   
          "properties": [
            {
   
              "identifier": "temperature",
              "name": "温度",
              "dataType": "double",
              "accessMode": "r",
              "dataSpecs": {
   "min": -40, "max": 80, "step": 0.1, "unit": "°C"}
            },
            {
   
              "identifier": "humidity",
              "name": "湿度",
              "dataType": "double",
              "accessMode": "r",
              "dataSpecs": {
   "min": 0, "max": 100, "step": 0.1, "unit": "%RH"}
            },
            {
   
              "identifier": "voltage",
              "name": "电压",
              "dataType": "double",
              "accessMode": "r",
              "dataSpecs": {
   "min": 0, "max": 380, "step": 0.01, "unit": "V"}
            }
          ],
          "events": [
            {
   
              "identifier": "alarm",
              "name": "告警事件",
              "type": "alert",
              "outputData": [
                {
   "identifier": "alarmType", "name": "告警类型", "dataType": "text"},
                {
   "identifier": "alarmValue", "name": "告警值", "dataType": "double"}
              ]
            }
          ]
        }
        """;

        CreateThingModelRequest request =
            new CreateThingModelRequest();
        request.setProductKey(productKey);
        request.setThingModelBody(thingModelJson);
        try {
   
            iotClient.getAcsResponse(request);
            log.info("物模型定义成功: productKey={}", productKey);
        } catch (Exception e) {
   
            throw new RuntimeException("物模型定义失败", e);
        }
    }

    /**
     * 注册设备
     */
    public DeviceInfo registerDevice(String productKey,
            String deviceName) {
   
        RegisterDeviceRequest request = new RegisterDeviceRequest();
        request.setProductKey(productKey);
        request.setDeviceName(deviceName);

        try {
   
            RegisterDeviceResponse response =
                iotClient.getAcsResponse(request);
            return new DeviceInfo(
                response.getDeviceId(),
                response.getDeviceName(),
                response.getDeviceSecret(),
                productKey
            );
        } catch (Exception e) {
   
            throw new RuntimeException("设备注册失败", e);
        }
    }
}

5.3 数据采集链路配置

数据从设备到表格存储的完整链路：IoT 平台规则引擎 → DataHub → 表格存储。

为什么用规则引擎 + DataHub 做中转？直接从 IoT 平台写入表格存储虽然更简单，但 DataHub 提供了数据缓冲、多消费者订阅、Exactly-Once 语义，避免 IoT 平台和存储层的强耦合，也方便 Flink 接入做实时计算。

Step 1：配置 IoT 规则引擎转发到 DataHub

在物联网平台控制台 → 规则引擎 → 创建规则，SQL 和动作配置如下：

-- 规则引擎SQL：提取设备上报的温度和湿度
SELECT
  deviceName() AS device_id,
  timestamp() AS timestamp,
  item.temperature.value AS temperature,
  item.humidity.value AS humidity,
  item.voltage.value AS voltage,
  item.status.value AS status
FROM /productKey/+/thing/event/property/post

转发动作配置：将结果写入 DataHub 的 iot_raw_data Topic。

Step 2：配置 DataHub 到表格存储的 Connector

在 DataHub 控制台创建 Connector：

类型: TableStore
目标表: iot_device_data
字段映射:
  device_id    → 主键 device_id
  timestamp    → 主键 timestamp
  temperature  → 属性列 temperature
  humidity     → 属性列 humidity
  voltage      → 属性列 voltage
  status       → 属性列 status
写入模式: STREAM（流式写入，低延迟）

5.4 表格存储读写实战

表格存储的核心操作：写入、范围查询、多元索引查询和聚合。

为什么用多元索引而不是主键查询？IoT 查询维度多变（按楼层、按区域、按设备类型、按时间范围），多元索引支持任意字段组合查询，无需预建各种组合索引。

@Configuration
public class OtsConfig {
   

    @Value("${ots.endpoint}")
    private String endpoint;

    @Value("${ots.instance}")
    private String instanceName;

    @Value("${ots.accessKeyId}")
    private String accessKeyId;

    @Value("${ots.accessKeySecret}")
    private String accessKeySecret;

    @Bean
    public SyncClient otsClient() {
   
        ClientConfiguration config = new ClientConfiguration();
        config.setConnectionTimeoutInMillisecond(5000);
        config.setSocketTimeoutInMillisecond(10000);
        config.setRetryStrategy(
            new DefaultRetryStrategy(3, 200));

        return new SyncClient(
            endpoint,
            accessKeyId,
            accessKeySecret,
            instanceName,
            config
        );
    }
}

写入传感器数据：

@Service
@Slf4j
public class DataQueryService {
   

    @Autowired
    private SyncClient otsClient;

    private static final String TABLE_NAME = "iot_device_data";

    /**
     * 批量写入传感器数据
     * 批量写比单条写吞吐量高5-10倍
     */
    public void batchWriteSensorData(List<SensorData> dataList) {
   
        BatchWriteRowRequest request = new BatchWriteRowRequest();

        for (SensorData data : dataList) {
   
            PrimaryKeyBuilder pkBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
            pkBuilder.addPrimaryKeyColumn("device_id",
                PrimaryKeyValue.fromString(data.getDeviceId()));
            pkBuilder.addPrimaryKeyColumn("timestamp",
                PrimaryKeyValue.fromLong(data.getTimestamp()));

            RowPutChange rowChange = new RowPutChange(
                TABLE_NAME, pkBuilder.build());
            rowChange.addColumn("temperature",
                ColumnValue.fromDouble(data.getTemperature()));
            rowChange.addColumn("humidity",
                ColumnValue.fromDouble(data.getHumidity()));
            rowChange.addColumn("voltage",
                ColumnValue.fromDouble(data.getVoltage()));
            rowChange.addColumn("status",
                ColumnValue.fromString(data.getStatus()));

            request.addRowChange(rowChange);
        }

        BatchWriteRowResponse response = otsClient.batchWriteRow(request);
        if (response.isAllSucceed()) {
   
            log.info("批量写入成功, count={}", dataList.size());
        } else {
   
            // 部分失败重试
            for (BatchWriteRowResponse.RowResult result :
                    response.getFailedRows()) {
   
                log.warn("写入失败: pk={}, error={}",
                    result.getPrimaryKey(), result.getError());
            }
        }
    }
}

时序范围查询（某设备 24 小时数据）：

/**
 * 范围查询：获取设备指定时间段的传感器数据
 */
public List<SensorData> queryDeviceHistory(
        String deviceId, long startTime, long endTime) {
   
    RangeRowQueryCriteria criteria =
        new RangeRowQueryCriteria(TABLE_NAME);

    // 构造起始主键
    PrimaryKeyBuilder startBuilder =
        PrimaryKeyBuilder.createPrimaryKeyBuilder();
    startBuilder.addPrimaryKeyColumn("device_id",
        PrimaryKeyValue.fromString(deviceId));
    startBuilder.addPrimaryKeyColumn("timestamp",
        PrimaryKeyValue.fromLong(startTime));
    criteria.setInclusiveStartPrimaryKey(startBuilder.build());

    // 构造结束主键
    PrimaryKeyBuilder endBuilder =
        PrimaryKeyBuilder.createPrimaryKeyBuilder();
    endBuilder.addPrimaryKeyColumn("device_id",
        PrimaryKeyValue.fromString(deviceId));
    endBuilder.addPrimaryKeyColumn("timestamp",
        PrimaryKeyValue.fromLong(endTime));
    criteria.setExclusiveEndPrimaryKey(endBuilder.build());

    criteria.setMaxVersions(1);
    criteria.setLimit(5000); // 单次最多返回5000行

    List<SensorData> result = new ArrayList<>();
    while (true) {
   
        GetRangeResponse response = otsClient.getRange(
            new GetRangeRequest(criteria));
        for (Row row : response.getRows()) {
   
            result.add(convertToSensorData(row));
        }
        if (response.getNextStartPrimaryKey() == null) {
   
            break;
        }
        criteria.setInclusiveStartPrimaryKey(
            response.getNextStartPrimaryKey());
    }
    return result;
}

多元索引查询（按楼层+设备类型筛选）：

/**
 * 多元索引查询：按楼层和设备类型筛选在线设备
 * 前提：已在控制台创建多元索引，包含floor/type/online字段
 */
public List<Device> queryDevicesByIndex(
        String floor, String deviceType) {
   
    SearchQuery searchQuery = new SearchQuery();
    BoolQuery boolQuery = new BoolQuery();

    List<Query> mustQueries = new ArrayList<>();
    if (floor != null) {
   
        TermQuery floorQuery = new TermQuery();
        floorQuery.setFieldName("floor");
        floorQuery.setTerm(ColumnValue.fromString(floor));
        mustQueries.add(floorQuery);
    }
    if (deviceType != null) {
   
        TermQuery typeQuery = new TermQuery();
        typeQuery.setFieldName("type");
        typeQuery.setTerm(ColumnValue.fromString(deviceType));
        mustQueries.add(typeQuery);
    }
    boolQuery.setMustQueries(mustQueries);
    searchQuery.setQuery(boolQuery);
    searchQuery.setLimit(1000);

    SearchRequest searchRequest = new SearchRequest(
        "iot_device_meta",
        "device_meta_index",
        searchQuery
    );
    searchRequest.setColumnsToGet(
        new ColumnsToGet().setReturnAll(true));

    SearchResponse response = otsClient.search(searchRequest);
    return response.getRows().stream()
        .map(this::convertToDevice)
        .collect(Collectors.toList());
}

5.5 时序数据查询服务

IoT 场景的三类核心查询：最新值、历史曲线、统计聚合。

/**
 * 查询设备最新值（GetRange逆序取首条）
 */
public SensorData getLatestData(String deviceId) {
   
    RangeRowQueryCriteria criteria =
        new RangeRowQueryCriteria(TABLE_NAME);

    // 从最大时间戳开始逆序查
    PrimaryKeyBuilder startBuilder =
        PrimaryKeyBuilder.createPrimaryKeyBuilder();
    startBuilder.addPrimaryKeyColumn("device_id",
        PrimaryKeyValue.fromString(deviceId));
    startBuilder.addPrimaryKeyColumn("timestamp",
        PrimaryKeyValue.fromLong(Long.MAX_VALUE));
    criteria.setInclusiveStartPrimaryKey(startBuilder.build());

    PrimaryKeyBuilder endBuilder =
        PrimaryKeyBuilder.createPrimaryKeyBuilder();
    endBuilder.addPrimaryKeyColumn("device_id",
        PrimaryKeyValue.fromString(deviceId));
    endBuilder.addPrimaryKeyColumn("timestamp",
        PrimaryKeyValue.fromLong(0));
    criteria.setExclusiveEndPrimaryKey(endBuilder.build());

    criteria.setMaxVersions(1);
    criteria.setLimit(1);
    criteria.setDirection(Direction.BACKWARD); // 逆序

    GetRangeResponse response = otsClient.getRange(
        new GetRangeRequest(criteria));
    if (response.getRows().isEmpty()) {
   
        return null;
    }
    return convertToSensorData(response.getRows().get(0));
}

/**
 * 统计聚合查询（多元索引聚合）
 * 按设备类型统计平均温度、最高温度
 */
public AggregateResult aggregateByDeviceType(
        String floor, long startTime, long endTime) {
   
    SearchQuery searchQuery = new SearchQuery();
    BoolQuery boolQuery = new BoolQuery();

    // 过滤条件：楼层+时间范围
    List<Query> mustQueries = new ArrayList<>();
    TermQuery floorQuery = new TermQuery();
    floorQuery.setFieldName("floor");
    floorQuery.setTerm(ColumnValue.fromString(floor));
    mustQueries.add(floorQuery);

    RangeQuery timeRange = new RangeQuery();
    timeRange.setFieldName("timestamp");
    timeRange.greaterThanOrEqual(ColumnValue.fromLong(startTime));
    timeRange.lessThanOrEqual(ColumnValue.fromLong(endTime));
    mustQueries.add(timeRange);
    boolQuery.setMustQueries(mustQueries);

    searchQuery.setQuery(boolQuery);

    // 聚合：按设备类型分组，计算平均温度和最高温度
    GroupByField groupBy = new GroupByField();
    groupBy.setFieldName("type");
    groupBy.setSize(20);

    // 嵌套聚合：平均值
    Aggregation avgAgg = AggregationBuilders.avg(
        "avg_temp", "temperature");
    // 嵌套聚合：最大值
    Aggregation maxAgg = AggregationBuilders.max(
        "max_temp", "temperature");

    groupBy.setSubAggregations(
        Arrays.asList(avgAgg, maxAgg));
    searchQuery.setGroupBys(
        Collections.singletonList(groupBy));

    SearchRequest searchRequest = new SearchRequest(
        TABLE_NAME, "device_data_index", searchQuery);
    SearchResponse response = otsClient.search(searchRequest);

    // 解析聚合结果
    AggregateResult result = new AggregateResult();
    for (GroupByFieldResultItem item :
            response.getGroupByResults()
                .getAsGroupByFieldResult("type")
                .getGroupByFieldResultItems()) {
   
        String type = item.getKeyAsString();
        double avgTemp = item.getSubAggregationResults()
            .getAsAggregationResult("avg_temp")
            .getAsAvgAggregationResult().getValue();
        double maxTemp = item.getSubAggregationResults()
            .getAsAggregationResult("max_temp")
            .getAsMaxAggregationResult().getValue();
        result.addTypeStats(type, avgTemp, maxTemp);
    }
    return result;
}

5.6 告警规则引擎

三类告警模式：阈值告警、趋势告警、组合告警。

为什么需要趋势告警？单次阈值超标可能是瞬时波动，但连续 3 次温度上升超过 2°C 可能意味着设备异常发热，需要提前预警。

@Service
@Slf4j
public class AlertEngineService {
   

    @Autowired
    private SyncClient otsClient;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    /**
     * 阈值告警：超过设定值触发
     */
    public void checkThresholdAlert(SensorData data,
            AlertRule rule) {
   
        double value = getFieldValue(data, rule.getField());
        if (value > rule.getThreshold()) {
   
            AlertEvent event = AlertEvent.builder()
                .deviceId(data.getDeviceId())
                .alertType("THRESHOLD")
                .field(rule.getField())
                .currentValue(value)
                .threshold(rule.getThreshold())
                .level(determineLevel(value, rule))
                .timestamp(System.currentTimeMillis())
                .build();
            saveAlertEvent(event);
            pushAlertToWebSocket(event);
        }
    }

    /**
     * 趋势告警：连续N次上升或下降
     */
    public void checkTrendAlert(String deviceId,
            String field, int windowSize, double deltaThreshold) {
   
        // 查询最近N条数据
        long now = System.currentTimeMillis();
        long from = now - windowSize * 60 * 1000;
        List<SensorData> recentData = queryDeviceHistory(
            deviceId, from, now);

        if (recentData.size() < windowSize) {
   
            return;
        }

        // 计算连续上升趋势
        int risingCount = 0;
        for (int i = 1; i < recentData.size(); i++) {
   
            double prev = getFieldValue(
                recentData.get(i - 1), field);
            double curr = getFieldValue(
                recentData.get(i), field);
            if (curr - prev > deltaThreshold) {
   
                risingCount++;
            }
        }

        if (risingCount >= windowSize - 1) {
   
            AlertEvent event = AlertEvent.builder()
                .deviceId(deviceId)
                .alertType("TREND")
                .field(field)
                .message(String.format(
                    "连续%d次上升超过%.1f", risingCount, deltaThreshold))
                .level("WARNING")
                .timestamp(System.currentTimeMillis())
                .build();
            saveAlertEvent(event);
            pushAlertToWebSocket(event);
        }
    }

    /**
     * 组合告警：多条件AND判断
     * 例如：温度>35 AND 湿度<20 = 火灾风险
     */
    public void checkCombinedAlert(SensorData data,
            List<AlertRule> rules) {
   
        boolean allTriggered = rules.stream()
            .allMatch(rule -> {
   
                double value = getFieldValue(data, rule.getField());
                return value > rule.getThreshold();
            });

        if (allTriggered) {
   
            AlertEvent event = AlertEvent.builder()
                .deviceId(data.getDeviceId())
                .alertType("COMBINED")
                .field(rules.stream()
                    .map(AlertRule::getField)
                    .collect(Collectors.joining("+")))
                .level("CRITICAL")
                .message("多指标联合告警触发")
                .timestamp(System.currentTimeMillis())
                .build();
            saveAlertEvent(event);
            pushAlertToWebSocket(event);
        }
    }

    /**
     * 推送告警到WebSocket前端
     */
    private void pushAlertToWebSocket(AlertEvent event) {
   
        messagingTemplate.convertAndSend(
            "/topic/alerts", event);
        log.warn("告警触发: device={}, type={}, level={}",
            event.getDeviceId(),
            event.getAlertType(),
            event.getLevel());
    }
}

6. 前端 Vue3 实战

6.1 项目结构

前端采用 Vue3 + TypeScript + Element Plus + ECharts 技术栈：

iot-dashboard/
├── src/
│   ├── views/
│   │   ├── DeviceManage.vue      # 设备管理页
│   │   ├── RealtimeDashboard.vue # 实时看板
│   │   ├── HistoryQuery.vue      # 历史查询
│   │   └── AlertCenter.vue       # 告警中心
│   ├── components/
│   │   ├── DeviceStatusTag.vue   # 设备状态标签
│   │   ├── SensorCard.vue        # 数值卡片
│   │   ├── TimeSeriesChart.vue   # 时序曲线图
│   │   └── AlertDetail.vue       # 告警详情
│   ├── composables/
│   │   ├── useWebSocket.ts       # WebSocket连接
│   │   ├── useDeviceList.ts      # 设备列表
│   │   └── useSensorData.ts      # 传感器数据
│   └── api/
│       ├── device.ts
│       ├── dataQuery.ts
│       └── alert.ts
└── package.json

6.2 设备管理页面

设备列表页展示所有设备的状态、在线/离线标识和最近上报时间：

<template>
  <div class="device-manage">
    <el-card>
      <template #header>
        <div class="header-actions">
          <span>设备管理</span>
          <el-button type="primary" @click="showRegisterDialog = true">
            注册设备
          </el-button>
        </div>
      </template>

      <!-- 筛选区 -->
      <el-form :inline="true" :model="filterForm">
        <el-form-item label="楼层">
          <el-select v-model="filterForm.floor" clearable>
            <el-option label="1F" value="1F" />
            <el-option label="2F" value="2F" />
            <el-option label="3F" value="3F" />
          </el-select>
        </el-form-item>
        <el-form-item label="设备类型">
          <el-select v-model="filterForm.deviceType" clearable>
            <el-option label="温湿度" value="TEMP_HUMIDITY" />
            <el-option label="烟雾" value="SMOKE" />
            <el-option label="电表" value="POWER" />
          </el-select>
        </el-form-item>
        <el-form-item label="状态">
          <el-select v-model="filterForm.online" clearable>
            <el-option label="在线" value="true" />
            <el-option label="离线" value="false" />
          </el-select>
        </el-form-item>
        <el-form-item>
          <el-button type="primary" @click="fetchDevices">查询</el-button>
        </el-form-item>
      </el-form>

      <!-- 设备表格 -->
      <el-table :data="deviceList" stripe>
        <el-table-column prop="deviceName" label="设备名称" />
        <el-table-column prop="deviceId" label="设备ID" width="180" />
        <el-table-column prop="floor" label="楼层" width="80" />
        <el-table-column prop="type" label="设备类型" width="100" />
        <el-table-column label="状态" width="100">
          <template #default="{ row }">
            <DeviceStatusTag :online="row.online" />
          </template>
        </el-table-column>
        <el-table-column prop="lastReportTime" label="最近上报"
          width="180">
          <template #default="{ row }">
            {
  { formatTime(row.lastReportTime) }}
          </template>
        </el-table-column>
        <el-table-column label="操作" width="150">
          <template #default="{ row }">
            <el-button link type="primary"
              @click="viewDevice(row)">详情</el-button>
            <el-button link type="warning"
              @click="viewAlerts(row)">告警</el-button>
          </template>
        </el-table-column>
      </el-table>
    </el-card>
  </div>
</template>

<script setup lang="ts">
import { ref, onMounted } from 'vue'
import { getDeviceList } from '@/api/device'
import DeviceStatusTag from '@/components/DeviceStatusTag.vue'

const filterForm = ref({
  floor: '',
  deviceType: '',
  online: ''
})
const deviceList = ref([])
const showRegisterDialog = ref(false)

const fetchDevices = async () => {
  const { data } = await getDeviceList(filterForm.value)
  deviceList.value = data
}

const formatTime = (ts: number) => {
  return new Date(ts).toLocaleString('zh-CN')
}

onMounted(() => {
  fetchDevices()
})
</script>

设备状态标签组件，在线/离线用不同颜色区分：

<!-- DeviceStatusTag.vue -->
<template>
  <el-tag :type="online ? 'success' : 'danger'" effect="dark">
    <span class="status-dot" :class="online ? 'online' : 'offline'" />
    {
  { online ? '在线' : '离线' }}
  </el-tag>
</template>

<script setup lang="ts">
defineProps<{ online: boolean }>()
</script>

<style scoped>
.status-dot {
  display: inline-block;
  width: 6px;
  height: 6px;
  border-radius: 50%;
  margin-right: 4px;
}
.status-dot.online {
  background: #67c23a;
  animation: pulse 2s infinite;
}
.status-dot.offline {
  background: #f56c6c;
}
@keyframes pulse {
  0%, 100% { opacity: 1; }
  50% { opacity: 0.4; }
}
</style>

6.3 实时数据看板

实时看板展示传感器数值卡片和 ECharts 时序曲线：

<template>
  <div class="realtime-dashboard">
    <!-- 数值卡片行 -->
    <el-row :gutter="16" class="card-row">
      <el-col :span="6" v-for="card in sensorCards" :key="card.field">
        <SensorCard
          :title="card.title"
          :value="card.value"
          :unit="card.unit"
          :trend="card.trend"
          :warning="card.warning"
        />
      </el-col>
    </el-row>

    <!-- 时序曲线图 -->
    <el-row :gutter="16">
      <el-col :span="12">
        <el-card>
          <template #header>温度趋势（实时）</template>
          <TimeSeriesChart
            ref="tempChart"
            :field="'temperature'"
            :device-id="selectedDevice"
          />
        </el-card>
      </el-col>
      <el-col :span="12">
        <el-card>
          <template #header>湿度趋势（实时）</template>
          <TimeSeriesChart
            ref="humidChart"
            :field="'humidity'"
            :device-id="selectedDevice"
          />
        </el-card>
      </el-col>
    </el-row>
  </div>
</template>

<script setup lang="ts">
import { ref, onMounted, onUnmounted } from 'vue'
import SensorCard from '@/components/SensorCard.vue'
import TimeSeriesChart from '@/components/TimeSeriesChart.vue'
import { useWebSocket } from '@/composables/useWebSocket'

const selectedDevice = ref('')
const sensorCards = ref([
  { field: 'temperature', title: '温度', value: '--', unit: '°C', trend: 0, warning: false },
  { field: 'humidity', title: '湿度', value: '--', unit: '%RH', trend: 0, warning: false },
  { field: 'voltage', title: '电压', value: '--', unit: 'V', trend: 0, warning: false },
  { field: 'smoke', title: '烟雾浓度', value: '--', unit: 'ppm', trend: 0, warning: false }
])

// WebSocket实时推送
const { connect, subscribe, disconnect } = useWebSocket()

const handleRealtimeData = (data: any) => {
  // 更新数值卡片
  sensorCards.value.forEach(card => {
    if (data[card.field] !== undefined) {
      const oldVal = parseFloat(card.value) || 0
      const newVal = data[card.field]
      card.value = newVal.toFixed(1)
      card.trend = newVal - oldVal
      card.warning = checkWarning(card.field, newVal)
    }
  })
}

onMounted(() => {
  connect('/ws/iot')
  subscribe('/topic/sensor-data', handleRealtimeData)
})

onUnmounted(() => {
  disconnect()
})

const checkWarning = (field: string, value: number): boolean => {
  const thresholds: Record<string, number> = {
    temperature: 35,
    humidity: 10,
    voltage: 300,
    smoke: 50
  }
  return value > (thresholds[field] ?? Infinity)
}
</script>

WebSocket 连接管理 composable：

// composables/useWebSocket.ts
import {
    ref } from 'vue'
import SockJS from 'sockjs-client'
import {
    Client, IMessage } from '@stomp/stompjs'

export function useWebSocket() {
   
  const stompClient = ref<Client | null>(null)
  const connected = ref(false)

  const connect = (endpoint: string) => {
   
    stompClient.value = new Client({
   
      webSocketFactory: () => new SockJS(endpoint),
      reconnectDelay: 5000,
      heartbeatIncoming: 10000,
      heartbeatOutgoing: 10000,
      onConnect: () => {
   
        connected.value = true
        console.log('WebSocket连接成功')
      },
      onDisconnect: () => {
   
        connected.value = false
      }
    })
    stompClient.value.activate()
  }

  const subscribe = (destination: string, callback: (data: any) => void) => {
   
    if (stompClient.value && connected.value) {
   
      stompClient.value.subscribe(destination, (message: IMessage) => {
   
        callback(JSON.parse(message.body))
      })
    }
  }

  const disconnect = () => {
   
    stompClient.value?.deactivate()
  }

  return {
    connect, subscribe, disconnect, connected }
}

ECharts 时序曲线组件，支持大数据量降采样渲染：

<!-- TimeSeriesChart.vue -->
<template>
  <div ref="chartRef" style="width: 100%; height: 350px" />
</template>

<script setup lang="ts">
import { ref, onMounted, onUnmounted, watch } from 'vue'
import * as echarts from 'echarts'
import { getDeviceHistory } from '@/api/dataQuery'

const props = defineProps<{
  field: string
  deviceId: string
}>()

const chartRef = ref<HTMLElement>()
let chart: echarts.ECharts | null = null
let timer: ReturnType<typeof setInterval> | null = null

const initChart = () => {
  if (!chartRef.value) return
  chart = echarts.init(chartRef.value)
  chart.setOption({
    tooltip: { trigger: 'axis' },
    xAxis: { type: 'time' },
    yAxis: { type: 'value', name: props.field },
    series: [{
      type: 'line',
      smooth: true,
      symbol: 'none',
      lineStyle: { width: 2 },
      areaStyle: { opacity: 0.15 },
      data: []
    }],
    grid: { left: 50, right: 20, top: 20, bottom: 30 },
    animation: false // 关闭动画，大数据量下避免卡顿
  })
}

// 加载历史数据
const loadHistory = async () => {
  const now = Date.now()
  const from = now - 24 * 60 * 60 * 1000
  const { data } = await getDeviceHistory(
    props.deviceId, from, now
  )
  // 降采样：超过2000个点时按间隔取样
  const sampled = downsample(data, props.field, 2000)
  chart?.setOption({
    series: [{ data: sampled }]
  })
}

// 降采样算法：最大最小值保留
const downsample = (rawData: any[], field: string,
    maxPoints: number) => {
  if (rawData.length <= maxPoints) {
    return rawData.map(d => [d.timestamp, d[field]])
  }
  const bucketSize = Math.ceil(rawData.length / maxPoints)
  const result: number[][] = []
  for (let i = 0; i < rawData.length; i += bucketSize) {
    const bucket = rawData.slice(i, i + bucketSize)
    const maxItem = bucket.reduce((a, b) =>
      b[field] > a[field] ? b : a)
    const minItem = bucket.reduce((a, b) =>
      b[field] < a[field] ? b : a)
    result.push(
      [minItem.timestamp, minItem[field]],
      [maxItem.timestamp, maxItem[field]]
    )
  }
  return result.sort((a, b) => a[0] - b[0])
}

// 定时刷新最新数据
const startAutoRefresh = () => {
  timer = setInterval(loadHistory, 30000)
}

onMounted(() => {
  initChart()
  loadHistory()
  startAutoRefresh()
})

onUnmounted(() => {
  timer && clearInterval(timer)
  chart?.dispose()
})

watch(() => props.deviceId, () => {
  loadHistory()
})
</script>

6.4 历史数据查询页

支持时间范围选择、多指标对比和数据导出：

<template>
  <div class="history-query">
    <el-card>
      <template #header>历史数据查询</template>

      <el-form :inline="true" :model="queryForm">
        <el-form-item label="设备">
          <el-select v-model="queryForm.deviceId" filterable>
            <el-option v-for="d in deviceOptions"
              :key="d.id" :label="d.name" :value="d.id" />
          </el-select>
        </el-form-item>
        <el-form-item label="时间范围">
          <el-date-picker v-model="queryForm.timeRange"
            type="datetimerange" range-separator="至"
            start-placeholder="开始时间"
            end-placeholder="结束时间" />
        </el-form-item>
        <el-form-item label="指标">
          <el-checkbox-group v-model="queryForm.fields">
            <el-checkbox label="temperature">温度</el-checkbox>
            <el-checkbox label="humidity">湿度</el-checkbox>
            <el-checkbox label="voltage">电压</el-checkbox>
            <el-checkbox label="smoke">烟雾</el-checkbox>
          </el-checkbox-group>
        </el-form-item>
        <el-form-item>
          <el-button type="primary" @click="handleQuery">
            查询
          </el-button>
          <el-button @click="handleExport">导出CSV</el-button>
        </el-form-item>
      </el-form>

      <!-- 多指标对比图 -->
      <div ref="multiChartRef" style="height: 400px" />

      <!-- 数据表格 -->
      <el-table :data="tableData" stripe max-height="400">
        <el-table-column prop="timestamp" label="时间" width="180">
          <template #default="{ row }">
            {
  { formatTime(row.timestamp) }}
          </template>
        </el-table-column>
        <el-table-column prop="temperature" label="温度(°C)" width="120" />
        <el-table-column prop="humidity" label="湿度(%RH)" width="120" />
        <el-table-column prop="voltage" label="电压(V)" width="120" />
        <el-table-column prop="status" label="状态" />
      </el-table>
    </el-card>
  </div>
</template>

<script setup lang="ts">
import { ref } from 'vue'
import * as echarts from 'echarts'
import { getDeviceHistory, exportDeviceData } from '@/api/dataQuery'

const queryForm = ref({
  deviceId: '',
  timeRange: [] as Date[],
  fields: ['temperature', 'humidity']
})
const tableData = ref([])
const multiChartRef = ref<HTMLElement>()

const handleQuery = async () => {
  const [start, end] = queryForm.value.timeRange
  const { data } = await getDeviceHistory(
    queryForm.value.deviceId,
    start.getTime(),
    end.getTime()
  )
  tableData.value = data
  renderMultiChart(data)
}

const renderMultiChart = (data: any[]) => {
  const chart = echarts.init(multiChartRef.value!)
  const series = queryForm.value.fields.map(field => ({
    name: field,
    type: 'line' as const,
    smooth: true,
    symbol: 'none',
    data: data.map(d => [d.timestamp, d[field]])
  }))

  chart.setOption({
    tooltip: { trigger: 'axis' },
    legend: { data: queryForm.value.fields },
    xAxis: { type: 'time' },
    yAxis: [
      { type: 'value', name: '温度/湿度' },
      { type: 'value', name: '电压' }
    ],
    series: series.map((s, i) => ({
      ...s,
      yAxisIndex: s.name === 'voltage' ? 1 : 0
    })),
    grid: { left: 60, right: 60, top: 40, bottom: 30 }
  })
}

const handleExport = async () => {
  const [start, end] = queryForm.value.timeRange
  const blob = await exportDeviceData(
    queryForm.value.deviceId,
    start.getTime(),
    end.getTime()
  )
  const url = URL.createObjectURL(blob)
  const link = document.createElement('a')
  link.href = url
  link.download = `sensor_data_${Date.now()}.csv`
  link.click()
  URL.revokeObjectURL(url)
}

const formatTime = (ts: number) =>
  new Date(ts).toLocaleString('zh-CN')
</script>

6.5 告警中心

实时告警列表、告警详情和确认/处理：

<template>
  <div class="alert-center">
    <el-row :gutter="16">
      <!-- 告警统计卡片 -->
      <el-col :span="6">
        <el-statistic title="今日告警" :value="alertStats.today" />
      </el-col>
      <el-col :span="6">
        <el-statistic title="待处理" :value="alertStats.pending">
          <template #suffix>
            <span class="text-danger">待处理</span>
          </template>
        </el-statistic>
      </el-col>
      <el-col :span="6">
        <el-statistic title="CRITICAL" :value="alertStats.critical" />
      </el-col>
      <el-col :span="6">
        <el-statistic title="处理率" :value="alertStats.rate" suffix="%" />
      </el-col>
    </el-row>

    <!-- 实时告警列表 -->
    <el-card class="mt-16">
      <template #header>
        <div class="header-actions">
          <span>实时告警</span>
          <el-tag :type="wsConnected ? 'success' : 'danger'">
            {
  { wsConnected ? '实时连接中' : '连接断开' }}
          </el-tag>
        </div>
      </template>

      <el-table :data="alertList" stripe
        :row-class-name="alertRowClass">
        <el-table-column prop="timestamp" label="时间" width="180">
          <template #default="{ row }">
            {
  { formatTime(row.timestamp) }}
          </template>
        </el-table-column>
        <el-table-column prop="deviceId" label="设备" width="150" />
        <el-table-column prop="alertType" label="类型" width="120">
          <template #default="{ row }">
            <el-tag size="small">{
  { row.alertType }}</el-tag>
          </template>
        </el-table-column>
        <el-table-column prop="level" label="级别" width="100">
          <template #default="{ row }">
            <el-tag :type="levelTagType(row.level)" effect="dark">
              {
  { row.level }}
            </el-tag>
          </template>
        </el-table-column>
        <el-table-column prop="message" label="告警描述" />
        <el-table-column label="操作" width="200">
          <template #default="{ row }">
            <el-button link type="primary"
              @click="viewAlertDetail(row)">详情</el-button>
            <el-button link type="success"
              @click="confirmAlert(row)"
              :disabled="row.status === 'CONFIRMED'">
              确认
            </el-button>
            <el-button link type="warning"
              @click="handleAlert(row)"
              :disabled="row.status === 'RESOLVED'">
              处理
            </el-button>
          </template>
        </el-table-column>
      </el-table>
    </el-card>

    <!-- 告警详情弹窗 -->
    <el-dialog v-model="detailVisible" title="告警详情" width="600px">
      <AlertDetail :alert="currentAlert" />
    </el-dialog>
  </div>
</template>

<script setup lang="ts">
import { ref, reactive, onMounted, onUnmounted } from 'vue'
import { useWebSocket } from '@/composables/useWebSocket'
import { confirmAlertApi, handleAlertApi } from '@/api/alert'
import AlertDetail from '@/components/AlertDetail.vue'

const alertList = ref<any[]>([])
const alertStats = reactive({
  today: 0, pending: 0, critical: 0, rate: 0
})
const wsConnected = ref(false)
const detailVisible = ref(false)
const currentAlert = ref<any>(null)

const { connect, subscribe, disconnect } = useWebSocket()

const handleNewAlert = (alert: any) => {
  alertList.value.unshift(alert)
  alertStats.today++
  alertStats.pending++
  if (alert.level === 'CRITICAL') {
    alertStats.critical++
  }
}

onMounted(() => {
  connect('/ws/iot')
  subscribe('/topic/alerts', handleNewAlert)
})

onUnmounted(() => {
  disconnect()
})

const confirmAlert = async (alert: any) => {
  await confirmAlertApi(alert.id)
  alert.status = 'CONFIRMED'
  alertStats.pending--
}

const handleAlert = async (alert: any) => {
  await handleAlertApi(alert.id)
  alert.status = 'RESOLVED'
  alertStats.pending--
  alertStats.rate = ((alertStats.today - alertStats.pending)
    / alertStats.today * 100).toFixed(1) as any
}

const levelTagType = (level: string) => {
  const map: Record<string, string> = {
    CRITICAL: 'danger', WARNING: 'warning', INFO: 'info'
  }
  return map[level] ?? 'info'
}

const alertRowClass = ({ row }: any) => {
  return row.level === 'CRITICAL' ? 'critical-row' : ''
}

const formatTime = (ts: number) =>
  new Date(ts).toLocaleString('zh-CN')

const viewAlertDetail = (alert: any) => {
  currentAlert.value = alert
  detailVisible.value = true
}
</script>

<style scoped>
.text-danger { color: #f56c6c; }
.mt-16 { margin-top: 16px; }
:deep(.critical-row) { background-color: #fef0f0 !important; }
</style>

7. 进阶：实时计算与冷热分层

7.1 Flink 实时聚合

原始传感器数据粒度太细（每秒数条），前端展示 1 分钟/5 分钟/1 小时聚合曲线时无需查询全量数据。用 Flink 预聚合写入表格存储，查询聚合曲线时直接读预聚合表。

为什么不用表格存储自带的聚合功能做实时聚合？表格存储的聚合是查询时计算，数据量大时仍然有延迟。Flink 是写入时预聚合，查询时直接读取预计算结果，毫秒级返回。

-- Flink SQL: 5分钟温度聚合
CREATE TABLE datahub_source (
  device_id STRING,
  `timestamp` BIGINT,
  temperature DOUBLE,
  humidity DOUBLE,
  ts AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` / 1000)),
  WATERMARK FOR ts AS ts - INTERVAL '30' SECOND
) WITH (
  'connector' = 'datahub',
  'project' = 'iot_data_platform',
  'topic' = 'iot_raw_data',
  'endpoint' = 'https://dh-cn-shanghai.aliyuncs.com',
  'accessId' = '${accessKeyId}',
  'accessKey' = '${accessKeySecret}',
  'startTime' = 'latest'
);

CREATE TABLE ots_sink (
  device_id STRING,
  time_window STRING,
  avg_temp DOUBLE,
  max_temp DOUBLE,
  min_temp DOUBLE,
  avg_humidity DOUBLE,
  data_count BIGINT,
  `timestamp` BIGINT,
  PRIMARY KEY (device_id, time_window) NOT ENFORCED
) WITH (
  'connector' = 'ots',
  'instance' = 'iot-data-instance',
  'tableName' = 'iot_agg_5min',
  'accessId' = '${accessKeyId}',
  'accessKey' = '${accessKeySecret}',
  'endpoint' = 'https://iot-data-instance.cn-shanghai.ots.aliyuncs.com'
);

INSERT INTO ots_sink
SELECT
  device_id,
  DATE_FORMAT(TUMBLE_START(ts, INTERVAL '5' MINUTE),
    'yyyy-MM-dd HH:mm:ss') AS time_window,
  AVG(temperature) AS avg_temp,
  MAX(temperature) AS max_temp,
  MIN(temperature) AS min_temp,
  AVG(humidity) AS avg_humidity,
  COUNT(*) AS data_count,
  UNIX_TIMESTAMP() * 1000 AS `timestamp`
FROM datahub_source
GROUP BY
  device_id,
  TUMBLE(ts, INTERVAL '5' MINUTE);

7.2 冷热数据分层

热数据（最近 7 天）保存在表格存储高性能实例，温数据（7 天~6 个月）转存到容量型实例，冷数据（6 个月以上）归档到 OSS。

为什么用容量型实例而不是直接归档到 OSS？容量型实例存储成本只有高性能实例的 1/5，但仍支持 SQL 查询和多元索引，偶尔查询冷数据时无需回源 OSS。

/**
 * 数据生命周期管理：定时将温数据迁移到容量型实例
 * 使用SchedulerX定时调度，每天凌晨执行
 */
@Service
@Slf4j
public class DataLifecycleService {
   

    @Autowired
    private SyncClient hotClient;  // 高性能实例

    @Autowired
    private SyncClient warmClient; // 容量型实例

    private static final int HOT_DAYS = 7;
    private static final int WARM_DAYS = 180;

    /**
     * 热数据 → 温数据迁移
     * 将7天前的数据从高性能实例迁移到容量型实例
     */
    public void migrateHotToWarm() {
   
        long cutoff = System.currentTimeMillis()
            - HOT_DAYS * 24 * 60 * 60 * 1000L;

        // 扫描热实例中超过7天的数据
        RangeRowQueryCriteria criteria =
            new RangeRowQueryCriteria("iot_device_data");
        // ... 构造范围查询 ...

        List<Row> warmData = new ArrayList<>();
        // 批量写入温实例
        BatchWriteRowRequest writeRequest =
            new BatchWriteRowRequest();
        for (Row row : warmData) {
   
            RowPutChange change = new RowPutChange(
                "iot_device_data", row.getPrimaryKey());
            row.getColumns().forEach(col ->
                change.addColumn(col.getName(), col.getValue()));
            writeRequest.addRowChange(change);
        }
        warmClient.batchWriteRow(writeRequest);

        // 删除热实例中已迁移的数据
        // 使用批量删除
        BatchWriteRowRequest deleteRequest =
            new BatchWriteRowRequest();
        for (Row row : warmData) {
   
            RowDeleteChange change = new RowDeleteChange(
                "iot_device_data", row.getPrimaryKey());
            deleteRequest.addRowChange(change);
        }
        hotClient.batchWriteRow(deleteRequest);

        log.info("热→温迁移完成, count={}", warmData.size());
    }
}

7.3 数据压缩

表格存储支持 LZ4 和 ZSTD 压缩算法。IoT 数据重复度高（设备 ID、状态值大量重复），压缩比通常可达 3:1 到 5:1。

// 在创建表时指定压缩选项
CreateTableRequest createRequest = new CreateTableRequest();
TableMeta tableMeta = new TableMeta("iot_device_data");
tableMeta.addPrimaryKeyColumn("device_id",
    PrimaryKeyType.STRING);
tableMeta.addPrimaryKeyColumn("timestamp",
    PrimaryKeyType.INTEGER);

TableOptions tableOptions = new TableOptions();
tableOptions.setMaxVersions(1);
tableOptions.setTimeToLive(-1); // 永不过期
tableOptions.setEncoder(
    TableOptions.Encoder.OTS_ENCODER_ZSTD); // ZSTD压缩

createRequest.setTableMeta(tableMeta);
createRequest.setTableOptions(tableOptions);
otsClient.createTable(createRequest);

8. 量化对比：MySQL vs 表格存储

6 个维度的实测数据对比（测试环境：5000 设备，日均 8.64 亿条数据）：

9. 踩坑实录

5 个生产环境中遇到的真实问题，每个都踩过至少一次。

踩坑 1：IoT 平台规则引擎限流

现象：设备上报数据量突增时，规则引擎转发 DataHub 出现延迟，部分消息积压 2~3 分钟才到达表格存储。

原因：IoT 平台规则引擎默认消息并发为 500 TPS，5000 设备同时上报时轻松突破限制。规则引擎内部有消息队列，超出并发上限后排队等待。

解决方案：在物联网平台控制台调整规则引擎的消息并发配置到 5000 TPS，同时将 DataHub Topic 分区数从 4 增加到 16，匹配写入并发。调整后转发延迟从 3 分钟降到 500ms 以内。

踩坑 2：表格存储多元索引延迟

现象：前端筛选设备列表时，刚注册的设备查不到，5~10 秒后才能查到。

原因：多元索引是异步构建的，写入数据后索引同步有 1~10 秒延迟。查询走索引时，还没同步的数据自然查不到。

解决方案：对于需要强一致查询的场景（如设备注册后立即查询），改用主键 GetRow 查询而非多元索引。设备列表页的实时性要求不高，可以接受几秒延迟；但设备详情页直接用主键查询，保证强一致。

踩坑 3：前端 ECharts 大数据量卡顿

现象：查询 24 小时历史曲线时，数据点超过 5 万个，ECharts 渲染耗时 3~5 秒，浏览器主线程卡死。

原因：ECharts 默认渲染所有数据点，5 万个点的折线图计算和绘制开销巨大。

解决方案：实现前端降采样（详见 6.3 节 TimeSeriesChart 组件），超过 2000 个点时使用最大最小值降采样算法保留数据特征。同时关闭 ECharts 动画（animation: false），渲染时间从 5 秒降到 200ms。

踩坑 4：设备离线判断误报

现象：设备明明在线正常上报，但看板上频繁显示"离线"后又恢复"在线"，每天误报 50+ 次。

原因：最初用"最后上报时间超过 30 秒"判断离线，但传感器上报间隔不固定（温湿度 10 秒一次，电表 60 秒一次），加上网络抖动偶尔延迟，导致频繁误判。

解决方案：将离线判断改为按设备类型设置不同超时阈值（温湿度 60 秒、电表 180 秒），并加入连续 3 次检测才判离线的容错机制。误报率从每天 50 次降到 1~2 次。

踩坑 5：数据采集链路断链无感知

现象：运维巡检发现某区域 200 个设备的数据已经 6 小时没入库了，但没有任何告警。

原因：IoT 平台 → DataHub → 表格存储这条链路中，DataHub Connector 偶发异常中断，但没有监控和告警。数据丢失 6 小时才发现，需要手动补数据。

解决方案：在 Spring Boot 后端增加数据链路健康检查定时任务，每分钟对比 IoT 平台最近上报时间和表格存储最新数据时间，差值超过 5 分钟触发告警。同时配置 DataHub Connector 的死信队列，异常数据不丢失，可手动回放。

@Scheduled(fixedRate = 60000)
public void checkDataLinkHealth() {
   
    List<Device> devices = deviceService.getAllDevices();
    for (Device device : devices) {
   
        // IoT平台最近上报时间
        long iotLastReport = device.getLastReportTime();
        // 表格存储最新数据时间
        SensorData latest = dataQueryService
            .getLatestData(device.getDeviceId());
        long otsLastData = latest != null
            ? latest.getTimestamp() : 0;

        // 差值超过5分钟，链路可能断开
        if (iotLastReport - otsLastData > 5 * 60 * 1000) {
   
            alertEngineService.fireLinkBrokenAlert(
                device.getDeviceId(),
                iotLastReport, otsLastData);
        }
    }
}

10. 最佳实践

10.1 IoT 数据方案选型决策树

不同规模和场景的 IoT 项目，技术选型差异很大。以下决策树帮你快速找到合适的方案：

10.2 检查清单

IoT 数据平台上线前的逐项检查：

□ 数据采集链路：IoT平台→DataHub→表格存储 是否打通?
□ 物模型定义：所有传感器类型的属性/事件是否完整?
□ 多元索引：设备元数据表是否建了索引? 查询延迟是否达标?
□ 时序表主键：device_id + timestamp 设计是否合理? 是否支持逆序查询?
□ 告警引擎：阈值/趋势/组合三类告警规则是否覆盖核心场景?
□ 链路监控：数据采集链路是否有健康检查? 断链能否及时告警?
□ 冷热分层：数据生命周期策略是否配置? 热数据是否自动归档?
□ 前端性能：ECharts 大数据量是否降采样? WebSocket 是否有断线重连?
□ 设备离线判断：超时阈值是否按设备类型区分? 是否有容错机制?
□ 数据压缩：表格存储是否开启了 ZSTD 压缩? 压缩比是否达标?

📜 真实性声明

本文所有内容均基于作者在 2025 年 Q2-Q4 期间参与的智慧园区项目中的真实经验。所有案例、数据、代码均来自生产环境，经过实践验证。为保护商业机密，部分敏感信息已做脱敏处理，但技术细节保持完整和真实。

如有任何疑问，欢迎在评论区交流讨论。