第1章：云平台搭建与部署

大家好，我是老张。今天咱们聊聊云平台搭建这件事。说实话，我刚入行那会儿，云平台还是个新鲜词。现在呢？几乎每个自控系统都离不开它。这一章，我把自己这些年踩过的坑、总结的经验，都掏出来给你们看看。

1.1 阿里云IoT平台接入

阿里云IoT平台，说白了就是个“中间人”。它帮我们把现场设备的数据，安全地传到云端。我个人习惯把它分成三步走：设备注册、数据上云、规则引擎配置。

1.1.1 设备注册与认证

先说说设备注册。你想想看，成千上万个设备要连到云端，总得有个身份吧？阿里云用的是三元组认证：ProductKey、DeviceName、DeviceSecret。这三个东西，缺一个都连不上。

注册流程其实不复杂：

1. 在IoT平台创建产品（定义设备类型、数据格式）
2. 批量注册设备（生成三元组）
3. 下载设备证书（用于TLS加密通信）

嗯，这里要注意：产品创建时，数据格式建议选Alink JSON。为什么？因为后面做规则引擎解析时，JSON格式最方便。我刚开始用二进制格式，后来解析起来那个痛苦啊...

1.1.2 MQTT协议接入

设备端怎么连？主流方案是MQTT。这个协议轻量、省带宽，特别适合工业现场。我给你们看个典型的接入代码：

# Python示例：阿里云IoT MQTT接入
from aliyun_iot_device import MqttClient

# 三元组配置
product_key = "你的ProductKey"
device_name = "你的DeviceName"
device_secret = "你的DeviceSecret"

# 创建客户端
client = MqttClient(product_key, device_name, device_secret)

# 连接阿里云
client.connect()
print("设备已连接云端")

# 上报数据
payload = {
    "temperature": 25.6,
    "humidity": 68.3
}
client.publish("/sys/{}/{}/thing/event/property/post".format(
    product_key, device_name), payload)

这段代码看着简单吧？但实际部署时，有几个坑：

• 心跳保活：MQTT默认心跳是120秒，但工业现场网络不稳定，我建议改成60秒
• QoS等级：控制指令用QoS 1（至少一次），数据上报用QoS 0（最多一次）
• 重连机制：断线后自动重连，最多重试3次，间隔递增

1.1.3 规则引擎与数据流转

数据到了云端，怎么处理？阿里云提供了规则引擎。你可以理解成“数据管道”——把设备数据转发到其他服务，比如数据库、函数计算、消息队列。

我常用的配置是这样的：

数据源	处理逻辑	目标
设备属性上报	过滤异常值（温度>100℃丢弃）	时序数据库TSDB
设备事件上报	触发告警（温度>80℃发短信）	函数计算+短信服务
设备生命周期	记录上下线日志	日志服务SLS

为什么会这样设计？因为工业数据有个特点：90%是正常数据，只有10%需要关注。把异常数据单独拎出来处理，能省不少计算资源。

1.2 私有云部署方案：EMQX + TDengine

有些客户数据敏感，不想上公有云。这时候就得搞私有云部署。我推荐一套组合拳：EMQX做消息接入，TDengine做时序存储。这两个都是开源方案，成本低、性能好。

1.2.1 EMQX集群部署

EMQX是个高性能MQTT消息服务器。单机就能扛百万连接，集群更是能到千万级。部署时我建议用Docker Compose，省心：

# docker-compose.yml
version: '3.8'
services:
  emqx1:
    image: emqx/emqx:5.0
    environment:
      - EMQX_NODE_NAME=emqx@node1.emqx.local
      - EMQX_CLUSTER__DISCOVERY_STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS=emqx@node1.emqx.local,emqx@node2.emqx.local
    ports:
      - "1883:1883"  # MQTT端口
      - "8083:8083"  # HTTP API端口
      - "18083:18083" # Dashboard端口
    volumes:
      - emqx_data:/opt/emqx/data

  emqx2:
    image: emqx/emqx:5.0
    environment:
      - EMQX_NODE_NAME=emqx@node2.emqx.local
      - EMQX_CLUSTER__DISCOVERY_STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS=emqx@node1.emqx.local,emqx@node2.emqx.local
    ports:
      - "1884:1883"
      - "8084:8083"
      - "18084:18083"
    volumes:
      - emqx_data2:/opt/emqx/data

volumes:
  emqx_data:
  emqx_data2:

1.2.2 TDengine时序数据库

数据存哪？TDengine是个好选择。它专门为时序数据设计，写入速度是InfluxDB的5倍，压缩比还高。我一般这么建表：

-- 创建数据库
CREATE DATABASE factory_iot KEEP 365 DAYS 10 BLOCKS 6;

-- 使用数据库
USE factory_iot;

-- 创建超级表（定义设备模板）
CREATE STABLE sensor_data (
    ts TIMESTAMP,
    temperature FLOAT,
    humidity FLOAT,
    pressure FLOAT
) TAGS (device_id BINARY(32), location BINARY(64));

-- 创建子表（每个设备一张子表）
CREATE TABLE temp_sensor_001 USING sensor_data TAGS ('sensor_001', '车间A区');
CREATE TABLE temp_sensor_002 USING sensor_data TAGS ('sensor_002', '车间B区');

嗯，这里有个设计要点：用超级表+子表的结构。为什么？因为查询时可以按设备ID过滤，效率极高。我见过有人把所有数据放一张表，结果查询越来越慢，最后不得不停机重建。

1.2.3 EMQX与TDengine集成

EMQX收到数据后，怎么写到TDengine？有两种方式：

• 规则引擎直接写：EMQX 5.0内置了TDengine桥接，配置简单
• 通过数据中间件：用Kafka或Nginx做缓冲，再批量写入

我个人推荐第二种。为什么？因为工业现场数据量波动大，白天可能每秒几千条，晚上只有几十条。用Kafka缓冲一下，能平滑写入压力。我曾经在一个光伏电站项目里，没用缓冲，结果中午光照强时数据量暴增，TDengine写入队列直接爆了。

1.3 云边数据同步策略

最后说说云边同步。说白了，就是云端和边缘端的数据怎么保持一致。这里有几个关键策略：

1.3.1 增量同步 vs 全量同步

全量同步简单粗暴，但费带宽。增量同步只传变化的数据，效率高。我建议：

• 首次部署：全量同步，把边缘端历史数据一次性传到云端
• 日常运行：增量同步，只传变化的数据（比如温度变化超过0.5℃才上报）
• 定时校验：每天凌晨做一次全量校验，确保两边数据一致

1.3.2 断点续传机制

网络断了怎么办？边缘端要支持断点续传。我常用的方案：

# 边缘端数据缓存逻辑
import sqlite3
import time

# 本地缓存数据库
conn = sqlite3.connect('edge_cache.db')
cursor = conn.cursor()

# 创建缓存表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS data_cache (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts TIMESTAMP,
        payload TEXT,
        status INTEGER DEFAULT 0  -- 0:未上传, 1:已上传
    )
''')

# 数据先写缓存
def cache_data(ts, payload):
    cursor.execute(
        "INSERT INTO data_cache (ts, payload) VALUES (?, ?)",
        (ts, payload)
    )
    conn.commit()

# 后台线程：定时上传
def upload_worker():
    while True:
        # 查询未上传的数据
        cursor.execute(
            "SELECT id, ts, payload FROM data_cache WHERE status=0 LIMIT 100"
        )
        rows = cursor.fetchall()
        
        for row in rows:
            try:
                # 上传到云端
                upload_to_cloud(row[1], row[2])
                # 标记已上传
                cursor.execute(
                    "UPDATE data_cache SET status=1 WHERE id=?",
                    (row[0],)
                )
            except Exception as e:
                print(f"上传失败: {e}")
                break  # 失败就停，等下次重试
        
        conn.commit()
        time.sleep(5)  # 每5秒检查一次

1.3.3 冲突解决策略

云边两端同时修改数据怎么办？比如边缘端改了设备参数，云端也改了。这时候需要冲突解决策略：

• 时间戳优先：谁的时间戳新，听谁的
• 边缘优先：边缘端的数据覆盖云端（适合现场控制场景）
• 云端优先：云端的数据覆盖边缘端（适合远程管理场景）

我个人习惯用时间戳优先。为什么？因为公平。但要注意：所有设备的时间必须同步，建议用NTP服务器统一校时。我曾经在一个项目里，边缘端设备时间慢了5分钟，导致数据覆盖顺序全乱了，排查了整整两天。

本章小结

好了，这一章的内容就到这。云平台搭建这件事，说白了就是三个字：稳、快、省。稳是数据不丢，快是响应及时，省是成本可控。后面几章，咱们会深入每个环节的细节。记住我一句话：别追求花哨的技术，稳定压倒一切。