物联网设备接入与通信

设备接入认证

物联网平台接入认证的核心目标是确保只有合法设备能够接入平台，防止仿冒和伪造设备。常见的认证方式包括一机一密、一型一密和 X.509 证书认证。

一机一密

一机一密指每台设备拥有唯一的设备密钥（DeviceSecret），设备在连接平台时使用该密钥对通信参数进行签名，平台端校验签名合法性后放行连接。

认证流程

设备                    IoT 平台
 |                         |
 |-- 1. 携带 ProductKey + DeviceName + ClientId + Timestamp -->|
 |                                                              |
 |                       2. 平台根据 ProductKey + DeviceName    |
 |                          查询对应的 DeviceSecret             |
 |                                                              |
 |<-- 3. 返回 Challenge / 非对称应答 (或不返回,由设备主动签名)-|
 |                                                              |
 |-- 4. 使用 DeviceSecret 对参数签名:                           |
 |    sign = HMACSHA1(DeviceSecret, params)                    |
 |                                                              |
 |                        5. 平台用相同 DeviceSecret 计算签名   |
 |                          并与设备上传 signature 比对          |
 |                                                              |
 |<-- 6. 认证通过,建立 MQTT/TCP 连接 ---------------------------|

HMAC 签名示例

java

// 设备端签名算法
public class DeviceAuthUtil {

    public static String sign(String deviceSecret, String params) throws Exception {
        Mac mac = Mac.getInstance("HmacSHA1");
        SecretKeySpec key = new SecretKeySpec(
            deviceSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA1");
        mac.init(key);
        byte[] digest = mac.doFinal(params.getBytes(StandardCharsets.UTF_8));
        return Hex.encodeHexString(digest).toUpperCase();
    }

    // 待签名字符串: "clientId123deviceNameMyDeviceproductKeya1b2c3timestamp1710000000"
    public static String buildSignContent(String clientId, String deviceName,
                                           String productKey, long timestamp) {
        return "clientId" + clientId
             + "deviceName" + deviceName
             + "productKey" + productKey
             + "timestamp" + timestamp;
    }
}

建表模型

sql

-- 设备认证信息表
CREATE TABLE device_auth (
    id            BIGINT AUTO_INCREMENT PRIMARY KEY,
    product_key   VARCHAR(32)  NOT NULL COMMENT '产品 key',
    device_name   VARCHAR(64)  NOT NULL COMMENT '设备名称',
    device_secret VARCHAR(64)  NOT NULL COMMENT '设备密钥',
    status        TINYINT      NOT NULL DEFAULT 0 COMMENT '0-未激活 1-已激活 2-已禁用',
    active_time   DATETIME     COMMENT '首次激活时间',
    gmt_create    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_modified  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_product_device (product_key, device_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='设备认证信息';

一型一密

一型一密指同一型号（ProductKey）下的设备共享一个 ProductSecret，设备首次上线时通过 ProductSecret 完成自动注册并获得唯一的 DeviceSecret，后续使用 DeviceSecret 做一机一密认证。

自动注册流程

设备                          IoT 平台
 |                               |
 |-- 1. 携带 ProductKey + DeviceName + ProductSecret 签名 -->|
 |                               |
 |                   2. 校验 ProductSecret 合法性            |
 |                   3. 自动生成 DeviceSecret                |
 |                   4. 写入 device_auth 表                  |
 |                                                          |
 |<-- 5. 返回 DeviceSecret (仅首次注册时返回) -------------|
 |                                                          |
 |-- 6. 后续连接使用一机一密流程 (使用已分配的 DeviceSecret) |

一型一密配置

yaml

# 产品密钥配置
iot:
  product:
    enabled: true
    auto-register: true            # 开启自动注册
    max-devices-per-product: 10000 # 单产品最大设备数
    secret-ttl-seconds: 86400      # ProductSecret 有效性缓存时间

java

@Service
public class DeviceAutoRegisterService {

    @Autowired
    private DeviceAuthMapper deviceAuthMapper;

    public DeviceRegisterResult autoRegister(String productKey, String deviceName,
                                              String clientId, long timestamp, String signature) {
        // 1. 从缓存/数据库获取 ProductSecret
        String productSecret = getProductSecret(productKey);
        // 2. 校验签名
        String expectedSign = sign(productSecret,
            "clientId" + clientId + "deviceName" + deviceName +
            "productKey" + productKey + "timestamp" + timestamp);
        if (!expectedSign.equals(signature)) {
            throw new AuthException("product secret sign mismatch");
        }
        // 3. 检查设备是否已注册
        DeviceAuth device = deviceAuthMapper.selectByProductKeyAndDeviceName(productKey, deviceName);
        if (device != null) {
            // 已注册,返回已有 DeviceSecret
            return DeviceRegisterResult.existed(device.getDeviceSecret());
        }
        // 4. 自动注册,生成新的 DeviceSecret
        String newDeviceSecret = generateDeviceSecret();
        deviceAuthMapper.insert(new DeviceAuth(productKey, deviceName, newDeviceSecret));
        return DeviceRegisterResult.newly(newDeviceSecret);
    }
}

X.509 证书认证

X.509 证书认证基于 PKI（公钥基础设施）体系，由 CA 为每个设备签发数字证书，通过 TLS 双向认证（mTLS）完成设备身份核验。适用于安全等级要求较高的场景，如工业物联网、车联网。

认证架构

          CA 根证书 (自签名)
                |
      +---------+---------+
      |                   |
  中间 CA 证书         中间 CA 证书
      |                   |
  +---+---+         +----+----+
  |       |         |        |
设备A  设备B      设备C    设备D

TLS 双向认证握手

证书管理

sql

-- 设备证书表
CREATE TABLE device_certificate (
    id              BIGINT AUTO_INCREMENT PRIMARY KEY,
    product_key     VARCHAR(32)  NOT NULL,
    device_name     VARCHAR(64)  NOT NULL,
    cert_serial     VARCHAR(128) NOT NULL COMMENT '证书序列号',
    cert_cn         VARCHAR(128) NOT NULL COMMENT '证书通用名称',
    issuer_dn       VARCHAR(256) NOT NULL COMMENT '颁发者 DN',
    cert_pem        TEXT         NOT NULL COMMENT '证书 PEM 内容',
    private_key_pem TEXT         COMMENT '私钥 PEM (仅首次返回)',
    cert_status     TINYINT      NOT NULL DEFAULT 0 COMMENT '0-有效 1-吊销 2-过期',
    not_before      DATETIME     NOT NULL COMMENT '证书生效时间',
    not_after       DATETIME     NOT NULL COMMENT '证书过期时间',
    gmt_create      DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_cert_serial (cert_serial),
    UNIQUE KEY uk_product_device (product_key, device_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='设备证书表';

-- 证书吊销列表
CREATE TABLE cert_revocation_list (
    id              BIGINT AUTO_INCREMENT PRIMARY KEY,
    cert_serial     VARCHAR(128) NOT NULL,
    revoked_at      DATETIME     NOT NULL COMMENT '吊销时间',
    reason          VARCHAR(64)  COMMENT '吊销原因',
    crl_publish_at  DATETIME     NOT NULL COMMENT 'CRL 发布时间'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='证书吊销列表';

Spring Boot mTLS 配置

yaml

server:
  ssl:
    enabled: true
    client-auth: need               # 双向认证
    key-store: classpath:iot-platform.p12
    key-store-password: ${KEYSTORE_PASS}
    key-store-type: PKCS12
    trust-store: classpath:trust-store.p12
    trust-store-password: ${TRUSTSTORE_PASS}
    trust-store-type: PKCS12
  port: 8443

证书校验示例

java

@Component
public class X509AuthFilter implements Filter {

    @Override
    public void doFilter(ServletRequest req, ServletResponse res, FilterChain chain)
            throws IOException, ServletException {
        HttpServletRequest request = (HttpServletRequest) req;

        X509Certificate[] certs = (X509Certificate[]) request.getAttribute(
            "jakarta.servlet.request.X509Certificate");
        if (certs == null || certs.length == 0) {
            throw new AuthenticationException("missing client certificate");
        }

        X509Certificate clientCert = certs[0];
        try {
            // 验证证书未过期
            clientCert.checkValidity();
            // 验证证书是否已吊销 (CRL/OCSP)
            checkRevocation(clientCert);
            // 提取设备身份信息
            String deviceName = extractCN(clientCert.getSubjectX500Principal());
            String certSerial = clientCert.getSerialNumber().toString();
            // 校验设备名与证书的绑定关系
            validateDeviceCertBinding(deviceName, certSerial);
        } catch (CertPathValidatorException | CRLException e) {
            throw new AuthenticationException("certificate validation failed", e);
        }
        chain.doFilter(req, res);
    }
}

设备注册流程

设备注册支持三种模式：预注册、自动注册、批量导入。

预注册

管理员在平台侧预先录入设备信息（ProductKey、DeviceName、DeviceSecret），设备出厂时写入凭证，首次上线直接认证。

java

@PostMapping("/api/v1/devices/pre-register")
public Result<Void> preRegister(@Valid @RequestBody PreRegisterReq req) {
    // 校验产品是否存在
    Product product = productService.getByProductKey(req.getProductKey());
    Assert.notNull(product, "product not found");

    // 检查设备名是否已存在
    DeviceAuth exist = deviceAuthMapper.selectByProductKeyAndDeviceName(
        req.getProductKey(), req.getDeviceName());
    Assert.isNull(exist, "device already registered");

    // 写入设备认证信息
    DeviceAuth device = new DeviceAuth();
    device.setProductKey(req.getProductKey());
    device.setDeviceName(req.getDeviceName());
    device.setDeviceSecret(generateDeviceSecret());
    device.setStatus(DeviceStatus.INACTIVE);
    deviceAuthMapper.insert(device);

    return Result.success(device);
}

自动注册

设备使用一型一密方式首次上线时由平台自动完成注册。需要控制自动注册的速率和总设备上限，避免被恶意刷量。

yaml

iot:
  auto-register:
    enabled: true
    rate-limit: 100       # 每分钟每产品最大自动注册数
    max-devices: 100000   # 单产品最大设备数
    ip-whitelist:         # 可选注册 IP 白名单
      - 10.0.0.0/8
      - 172.16.0.0/12

批量导入

通过 CSV/Excel 文件批量导入设备，适用于量产场景。

java

public class DeviceBatchImportService {

    public BatchImportResult importFromCsv(MultipartFile file, String productKey) {
        List<DeviceAuth> devices = new ArrayList<>();
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(file.getInputStream(), StandardCharsets.UTF_8))) {
            String line;
            while ((line = reader.readLine()) != null) {
                String[] fields = line.split(",");
                if (fields.length < 1) continue;
                DeviceAuth device = new DeviceAuth();
                device.setProductKey(productKey);
                device.setDeviceName(fields[0].trim());
                device.setDeviceSecret(generateDeviceSecret());
                device.setStatus(DeviceStatus.INACTIVE);
                devices.add(device);
            }
        }
        // 批量写入,使用 ignore 避免重复
        deviceAuthMapper.batchInsertIgnore(devices);
        return BatchImportResult.of(devices.size());
    }
}

sql

-- 批量写入 (MySQL)
INSERT IGNORE INTO device_auth (product_key, device_name, device_secret, status)
VALUES
('pk1', 'dev001', 'sec001', 0),
('pk1', 'dev002', 'sec002', 0),
('pk1', 'dev003', 'sec003', 0);

三种注册方式对比

模式	适用场景	安全性	运维复杂度	出厂流程
预注册	高安全要求、固定设备	最高	高	需预烧录 DeviceSecret
自动注册	消费级设备、快速量产	中等	低	只需烧录 ProductSecret
批量导入	企业级设备批量接入	高	中	烧录批量导出的密钥

设备影子

设备影子（Device Shadow）是一个 JSON 文档，用于缓存设备的最新状态。设备离线时平台存储其最新状态，设备上线后自动同步。应用层可以读取影子获取设备状态，也可以更新影子中的 desired 状态指挥设备。

影子文档结构

json

{
  "state": {
    "reported": {
      "temperature": 25.6,
      "humidity": 68.2,
      "switch": "on",
      "firmware_ver": "2.1.0"
    },
    "desired": {
      "temperature": 26.0,
      "switch": "off"
    }
  },
  "metadata": {
    "reported": {
      "temperature": { "timestamp": 1710000001 },
      "humidity":    { "timestamp": 1710000001 },
      "switch":      { "timestamp": 1710000002 },
      "firmware_ver":{ "timestamp": 1710000000 }
    },
    "desired": {
      "temperature": { "timestamp": 1710000100 },
      "switch":      { "timestamp": 1710000100 }
    }
  },
  "version": 42,
  "timestamp": 1710000100
}

字段	说明
`state.reported`	设备主动上报的状态（设备 -> 平台）
`state.desired`	应用层期望设备达到的状态（平台 -> 设备）
`metadata`	每个属性的更新时间戳
`version`	影子文档版本号，递增且用于乐观锁冲突检测
`timestamp`	影子文档最后更新时间

影子文档存储

sql

CREATE TABLE device_shadow (
    id            BIGINT AUTO_INCREMENT PRIMARY KEY,
    product_key   VARCHAR(32)   NOT NULL,
    device_name   VARCHAR(64)   NOT NULL,
    shadow_doc    JSON          NOT NULL COMMENT '影子文档 JSON',
    version       BIGINT        NOT NULL DEFAULT 0 COMMENT '影子版本号',
    gmt_create    DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_modified  DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_product_device (product_key, device_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='设备影子';

影子状态同步机制

状态更新流程

设备                          IoT 平台                      应用
 |                             |                            |
 |-- MQTT: report state ------>|                            |
 |   topic: /shadow/update     |                            |
 |   payload: {"temperature":  |                            |
 |              25.6}          |                            |
 |                             |                            |
 |                  1. 校验版本号 (乐观锁)                   |
 |                  2. 合并 reported 字段                    |
 |                  3. version++                            |
 |                  4. 比对 desired,若有差异则下发 desired   |
 |                                                          |
 |<-- MQTT: desired delta ----|                            |
 |   topic: /shadow/update    |                            |
 |   payload: {"switch":"off"}|                            |
 |                                                          |
 |-- MQTT: update delta ack ->|                            |
 |                             |                            |
 |                             |-- HTTP GET /shadow ------->|
 |                             |<-- shadow JSON ------------|

设备上报状态

java

// 设备端上报 MQTT 消息
public class ShadowReporter {

    private static final String SHADOW_UPDATE_TOPIC = "/shadow/update";

    public void reportTemperature(double temperature) {
        JsonObject reported = new JsonObject();
        reported.addProperty("temperature", temperature);
        reported.addProperty("timestamp", System.currentTimeMillis() / 1000);

        JsonObject state = new JsonObject();
        state.add("reported", reported);

        JsonObject payload = new JsonObject();
        payload.add("state", state);
        payload.addProperty("version", shadowVersion); // 当前影子版本号

        mqttClient.publish(SHADOW_UPDATE_TOPIC, payload.toString().getBytes(), 1);
    }
}

平台端影子更新

java

@Service
public class ShadowUpdateService {

    @Autowired
    private DeviceShadowMapper shadowMapper;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private static final String SHADOW_LOCK_PREFIX = "shadow:lock:";

    @Transactional
    public boolean updateShadow(String productKey, String deviceName,
                                 JsonObject reported, Long expectedVersion) {
        String lockKey = SHADOW_LOCK_PREFIX + productKey + ":" + deviceName;
        RLock lock = redissonClient.getLock(lockKey);

        try {
            // 分布式锁防止并发更新
            if (!lock.tryLock(3, 10, TimeUnit.SECONDS)) {
                throw new ShadowUpdateException("acquire shadow lock timeout");
            }

            DeviceShadow shadow = shadowMapper.selectByProductKeyAndDeviceName(
                productKey, deviceName);

            // 乐观锁版本校验
            if (expectedVersion != null && !expectedVersion.equals(shadow.getVersion())) {
                throw new ShadowVersionConflictException(
                    "version conflict, expected: " + expectedVersion
                    + ", actual: " + shadow.getVersion());
            }

            // 合并 reported 字段
            JsonObject currentDoc = JsonParser.parseString(shadow.getShadowDoc()).getAsJsonObject();
            JsonObject currentReported = currentDoc.getAsJsonObject("state")
                .getAsJsonObject("reported");
            mergeJson(currentReported, reported);
            updateMetadata(currentDoc, "reported", reported);

            // 版本递增
            long newVersion = shadow.getVersion() + 1;
            currentDoc.addProperty("version", newVersion);
            currentDoc.addProperty("timestamp", System.currentTimeMillis() / 1000);

            // 更新数据库
            shadowMapper.updateShadowDoc(productKey, deviceName,
                currentDoc.toString(), newVersion, shadow.getVersion());

            // 检查 desired 是否需要下发
            checkAndDeliverDesired(productKey, deviceName, currentDoc);

            return true;
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

desired 下发

java

private void checkAndDeliverDesired(String productKey, String deviceName, JsonObject shadowDoc) {
    JsonObject state = shadowDoc.getAsJsonObject("state");
    JsonObject desired = state.getAsJsonObject("desired");
    JsonObject reported = state.getAsJsonObject("reported");

    if (desired == null || desired.size() == 0) {
        return;
    }

    // 计算 delta: desired - reported
    JsonObject delta = new JsonObject();
    for (Map.Entry<String, JsonElement> entry : desired.entrySet()) {
        String key = entry.getKey();
        if (reported == null || !reported.has(key)
            || !reported.get(key).equals(entry.getValue())) {
            delta.add(key, entry.getValue());
        }
    }

    if (delta.size() > 0) {
        // 通过 MQTT 下发 delta 到设备
        String topic = "/shadow/delta/" + productKey + "/" + deviceName;
        JsonObject deltaMsg = new JsonObject();
        deltaMsg.add("delta", delta);
        deltaMsg.addProperty("version", shadowDoc.get("version").getAsLong());
        mqttGateway.sendToTopic(topic, deltaMsg.toString());
    }
}

影子缓存优化

影子文档频繁读写，直接操作数据库会有性能瓶颈。通过引入 Redis 缓存减少数据库压力，同时可以缓存设备最近的影子状态减少设备端请求。

缓存架构

   设备/应用                  Redis (缓存层)                     MySQL (持久层)
      |                          |                                  |
      |-- 读影子 --------------->|                                  |
      |<-- 返回 (LRU 缓存) -----|                                  |
      |                          |                                  |
      |-- 写影子 --------------->|                                  |
      |                          |-- 异步写 behind (延迟双写) ---->|
      |                          |   (Redis 作为主存储)             |
      |                          |                                  |
      |                          |  定期全量持久化 / binlog 同步    |
      |                          |<---------------------------------|

缓存实现

java

@Component
public class ShadowCacheManager {

    private static final String SHADOW_CACHE_KEY = "shadow:%s:%s";
    private static final Duration CACHE_TTL = Duration.ofHours(2);

    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private DeviceShadowMapper shadowMapper;

    public JsonObject getShadow(String productKey, String deviceName) {
        String cacheKey = String.format(SHADOW_CACHE_KEY, productKey, deviceName);

        // 1. 查缓存
        String cached = redisTemplate.opsForValue().get(cacheKey);
        if (cached != null) {
            return JsonParser.parseString(cached).getAsJsonObject();
        }

        // 2. 缓存 miss,查数据库
        DeviceShadow shadow = shadowMapper.selectByProductKeyAndDeviceName(
            productKey, deviceName);
        if (shadow == null) {
            return createEmptyShadow(productKey, deviceName);
        }

        // 3. 回填缓存
        JsonObject doc = JsonParser.parseString(shadow.getShadowDoc()).getAsJsonObject();
        redisTemplate.opsForValue().set(cacheKey, doc.toString(), CACHE_TTL);
        return doc;
    }

    @Transactional
    public void updateShadowAndCache(String productKey, String deviceName,
                                      JsonObject reported, Long version) {
        String cacheKey = String.format(SHADOW_CACHE_KEY, productKey, deviceName);

        // 1. 更新数据库 (事务内)
        shadowMapper.updateReported(productKey, deviceName, reported.toString(), version);

        // 2. 更新缓存 (缓存为主,DB 异步持久化的场景可使用 Redis 作为主写入)
        JsonObject doc = getShadow(productKey, deviceName);
        mergeIntoReported(doc, reported);
        doc.addProperty("version", doc.get("version").getAsLong() + 1);
        redisTemplate.opsForValue().set(cacheKey, doc.toString(), CACHE_TTL);
    }
}

断连恢复优化

设备离线期间影子持续缓存最新状态，设备上线后一次性推送离线期间的 desired delta，而非逐条推送。

java

// 设备上线时批量同步离线期间的 desired 变更
public void onDeviceConnected(String productKey, String deviceName) {
    JsonObject shadow = shadowCacheManager.getShadow(productKey, deviceName);
    JsonObject state = shadow.getAsJsonObject("state");
    if (state == null) return;

    JsonObject desired = state.getAsJsonObject("desired");
    JsonObject reported = state.getAsJsonObject("reported");
    if (desired == null || desired.size() == 0) return;

    // 计算完整 delta 一次性下发
    JsonObject delta = new JsonObject();
    for (Map.Entry<String, JsonElement> entry : desired.entrySet()) {
        String key = entry.getKey();
        if (reported == null || !reported.has(key)
            || !reported.get(key).equals(entry.getValue())) {
            delta.add(key, entry.getValue());
        }
    }

    if (delta.size() > 0) {
        JsonObject deltaMsg = new JsonObject();
        deltaMsg.add("delta", delta);
        deltaMsg.addProperty("version", shadow.get("version").getAsLong());
        mqttGateway.sendToTopic(
            "/shadow/delta/" + productKey + "/" + deviceName,
            deltaMsg.toString());
    }
}

数据上报

设备端采集数据后上报到平台，平台完成解析、校验、存储和转发。数据上报涉及消息格式、解析脚本、频率控制和时序数据存储等环节。

消息格式规范

平台支持多种消息格式，设备可根据自身资源情况选择。

JSON 格式

json

{
  "id": "msg_1710000001",
  "method": "thing.event.property.post",
  "params": {
    "temperature": 25.6,
    "humidity": 68.2,
    "pressure": 1013.2
  },
  "timestamp": 1710000001000
}

二进制格式

text

字段定义:
  Bytes 0-1:   消息类型 (0x01 = 属性上报)
  Bytes 2-3:   消息体长度 (N)
  Bytes 4-N+3: 消息体 (TLV 编码)
  Bytes N+4:   校验和 (CRC8)

TLV 编码示例:
  Tag=0x01 (temperature)  Length=4  Value=0x41CC0000 (25.6 float)
  Tag=0x02 (humidity)     Length=4  Value=0x42886666 (68.2 float)

Protobuf 格式

protobuf

syntax = "proto3";

package iot.device;

message PropertyReport {
    string   msg_id      = 1;
    int64    timestamp   = 2;
    float    temperature = 3;
    float    humidity    = 4;
    float    pressure    = 5;
    repeated TagValue   tags = 10;
}

message TagValue {
    string key   = 1;
    string value = 2;
}

TLV 编解码示例

java

public class TlvCodec {

    public static byte[] encode(List<TlvEntry> entries) {
        ByteBuffer buf = ByteBuffer.allocate(1024);
        buf.order(ByteOrder.BIG_ENDIAN);
        for (TlvEntry entry : entries) {
            buf.putShort(entry.getTag());
            buf.putShort((short) entry.getValue().length);
            buf.put(entry.getValue());
        }
        buf.flip();
        byte[] result = new byte[buf.remaining()];
        buf.get(result);
        return result;
    }

    public static List<TlvEntry> decode(byte[] data) {
        List<TlvEntry> entries = new ArrayList<>();
        ByteBuffer buf = ByteBuffer.wrap(data);
        buf.order(ByteOrder.BIG_ENDIAN);
        while (buf.remaining() >= 4) {
            short tag  = buf.getShort();
            short len  = buf.getShort();
            byte[] val = new byte[len];
            buf.get(val);
            entries.add(new TlvEntry(tag, val));
        }
        return entries;
    }
}

数据解析脚本

对于二进制、TLV 等非标准格式，平台提供脚本引擎进行解析，将原始字节转换为平台统一的 JSON 格式。

脚本引擎架构

 设备上报二进制数据
        |
        v
 协议适配层 (识别 Protocol Type)
        |
        v
 脚本引擎 (Groovy/JS/Nashorn)
   - 从脚本仓库加载对应脚本
   - 沙箱执行 (禁止网络/文件操作)
   - 输入: byte[] / Map<String,Object>
   - 输出: JSON (统一格式)
        |
        v
 平台统一处理 (校验/存储/转发)

脚本管理

sql

CREATE TABLE parse_script (
    id              BIGINT AUTO_INCREMENT PRIMARY KEY,
    product_key     VARCHAR(32)   NOT NULL,
    script_name     VARCHAR(128)  NOT NULL COMMENT '脚本名称',
    script_lang     VARCHAR(16)   NOT NULL DEFAULT 'groovy' COMMENT 'groovy/js',
    script_content  MEDIUMTEXT    NOT NULL COMMENT '脚本内容',
    script_version  INT           NOT NULL DEFAULT 1,
    status          TINYINT       NOT NULL DEFAULT 0 COMMENT '0-草稿 1-已发布 2-已下线',
    gmt_create      DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_modified    DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_product_script (product_key, script_name, script_version)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='解析脚本';

Groovy 脚本示例

groovy

// 二进制温湿度传感器解析脚本
import java.nio.ByteBuffer
import java.nio.ByteOrder

def parse(byte[] rawData) {
    ByteBuffer buf = ByteBuffer.wrap(rawData)
    buf.order(ByteOrder.LITTLE_ENDIAN)

    def result = [:]
    result.msgId       = "raw_" + System.currentTimeMillis()
    result.method      = "thing.event.property.post"
    result.timestamp   = System.currentTimeMillis()

    def params = [:]
    // 读取温度 (2 bytes, 0.1 精度)
    short tempRaw = buf.getShort()
    params.temperature = tempRaw / 10.0

    // 读取湿度 (2 bytes, 0.1 精度)
    short humRaw = buf.getShort()
    params.humidity = humRaw / 10.0

    // 读取开关状态 (1 byte)
    byte switchRaw = buf.get()
    params.switch = switchRaw == 1 ? "on" : "off"

    result.params = params
    return result
}

脚本引擎执行

java

@Component
public class ScriptEngineExecutor {

    private final Map<String, CompiledScript> scriptCache = new ConcurrentHashMap<>();

    public JsonObject execute(String productKey, byte[] rawData) {
        String scriptContent = getActiveScript(productKey);

        // 使用缓存编译脚本
        CompiledScript script = scriptCache.computeIfAbsent(productKey, k -> {
            ScriptEngineManager manager = new ScriptEngineManager();
            ScriptEngine engine = manager.getEngineByName("groovy");
            return ((Compilable) engine).compile(scriptContent);
        });

        // 沙箱绑定
        Bindings bindings = new SimpleBindings();
        bindings.put("rawData", rawData);

        // 在 Sandbox 中执行
        Object result = AccessController.doPrivileged(
            (PrivilegedAction<Object>) () -> {
                try {
                    return script.eval(bindings);
                } catch (ScriptException e) {
                    throw new ParseException("script execute failed", e);
                }
            },
            // 沙箱权限: 只允许基本运行时权限
            new java.security.PermissionCollection() {
                {
                    add(new RuntimePermission("accessDeclaredMembers"));
                }
            }
        );

        return JsonParser.parseString(result.toString()).getAsJsonObject();
    }
}

上报频率控制

防止设备大量高频上报导致平台过载，需要从设备端和平台端两个维度进行控制。

设备端上报间隔

java

// 设备端上报频率控制
public class UploadThrottler {

    private final long minIntervalMs;       // 最小上报间隔, 默认 5000ms
    private long lastUploadTime = 0;

    public UploadThrottler(long minIntervalMs) {
        this.minIntervalMs = minIntervalMs;
    }

    public boolean tryUpload() {
        long now = System.currentTimeMillis();
        if (now - lastUploadTime >= minIntervalMs) {
            lastUploadTime = now;
            return true;
        }
        // 丢弃本次上报 (或做采样降级)
        return false;
    }

    // 弹性上报: 根据数据重要性支持优先级上报
    public boolean tryUploadWithPriority(int priority) {
        long now = System.currentTimeMillis();
        long threshold = minIntervalMs / (priority + 1); // priority 0-4
        if (now - lastUploadTime >= threshold) {
            lastUploadTime = now;
            return true;
        }
        return false;
    }
}

平台端限流

java

@Component
public class UploadRateLimiter {

    // 每产品每秒最大消息数
    private final Map<String, RateLimiter> productLimiters = new ConcurrentHashMap<>();

    public boolean allowUpload(String productKey) {
        RateLimiter limiter = productLimiters.computeIfAbsent(productKey,
            k -> RateLimiter.create(1000)); // 默认 1000 TPS
        return limiter.tryAcquire();
    }

    // 动态调整限流阈值 (根据平台整体负载)
    @EventListener
    public void onLoadChange(LoadAlertEvent event) {
        double factor = event.getFactor(); // 0.1 ~ 1.0
        productLimiters.forEach((k, limiter) -> {
            double newRate = limiter.getRate() * factor;
            limiter.setRate(Math.max(newRate, 10)); // 最低 10 TPS
        });
    }
}

降级策略

yaml

iot:
  rate-limit:
    default-per-product: 1000       # 每产品默认 TPS
    default-per-device: 10          # 每设备默认 TPS
    strategy:
      overflow:                     # 超限处理策略
        - type: drop                # 丢弃 (静默丢弃)
        - type: delay               # 延迟 (放入延迟队列)
          queue-capacity: 10000
          delay-ms: 1000
        - type: sample              # 采样 (只保留关键数据点)
          sample-rate: 0.1          # 采样率 10%
      global-breach:                # 全局过载
        - type: reject              # 拒绝 (返回错误码)
        - type: degrade             # 降级 (关闭非核心功能)

时序数据存储

使用 TDengine 作为时序数据库存储设备上报数据，基于超级表（STable）模型管理同类设备数据。

超级表设计

sql

-- 创建设备数据超级表
CREATE STABLE IF NOT EXISTS iot.device_data (
    ts          TIMESTAMP           NOT NULL,   -- 采集时间
    temperature FLOAT               COMMENT '温度',
    humidity    FLOAT               COMMENT '湿度',
    pressure    FLOAT               COMMENT '压力',
    switch      TINYINT             COMMENT '开关 0/1',
    raw_msg_id  VARCHAR(64)         COMMENT '原始消息 ID'
) TAGS (
    product_key VARCHAR(32),        -- 产品标识
    device_name VARCHAR(64),        -- 设备标识
    location    VARCHAR(128),       -- 安装位置
    group_id    INT                 -- 设备分组
);

-- 为每个设备创建子表 (自动继承 STable 结构)
CREATE TABLE IF NOT EXISTS iot.device_pk1_dev001
    USING iot.device_data TAGS ('pk1', 'dev001', 'factory-A', 1);

CREATE TABLE IF NOT EXISTS iot.device_pk1_dev002
    USING iot.device_data TAGS ('pk1', 'dev002', 'warehouse-B', 2);

数据写入

java

@Service
public class TimeSeriesWriter {

    @Autowired
    private JdbcTemplate tdJdbcTemplate;

    public void writeDeviceData(String productKey, String deviceName,
                                 JsonObject params, long timestamp) {
        String tableName = "device_" + productKey + "_" + deviceName;

        String sql = "INSERT INTO " + tableName
            + " (ts, temperature, humidity, pressure, switch, raw_msg_id) "
            + "VALUES (?, ?, ?, ?, ?, ?)";

        tdJdbcTemplate.update(sql,
            new Timestamp(timestamp),
            getFloat(params, "temperature"),
            getFloat(params, "humidity"),
            getFloat(params, "pressure"),
            getInt(params, "switch"),
            getString(params, "raw_msg_id")
        );
    }

    // 批量写入提升吞吐
    @Transactional
    public void batchWrite(List<DeviceDataPoint> points) {
        NamedParameterJdbcTemplate namedJdbc = new NamedParameterJdbcTemplate(tdJdbcTemplate);

        String sql = "INSERT INTO :tableName USING iot.device_data TAGS "
            + "(:productKey, :deviceName, :location, :groupId) "
            + "(ts, temperature, humidity, pressure, switch) "
            + "VALUES (:ts, :temp, :hum, :press, :sw)";

        SqlParameterSource[] batch = points.stream()
            .map(p -> new MapSqlParameterSource()
                .addValue("tableName", "device_" + p.getProductKey() + "_" + p.getDeviceName())
                .addValue("productKey", p.getProductKey())
                .addValue("deviceName", p.getDeviceName())
                .addValue("location", p.getLocation())
                .addValue("groupId", p.getGroupId())
                .addValue("ts", p.getTimestamp())
                .addValue("temp", p.getTemperature())
                .addValue("hum", p.getHumidity())
                .addValue("press", p.getPressure())
                .addValue("sw", p.getSwitchStatus())
            )
            .toArray(SqlParameterSource[]::new);

        namedJdbc.batchUpdate(sql, batch);
    }
}

典型查询

sql

-- 查询某个设备的最近 1 小时数据
SELECT ts, temperature, humidity
FROM iot.device_pk1_dev001
WHERE ts >= NOW() - 1h
ORDER BY ts ASC;

-- 按产品聚合查询 (跨所有子表)
SELECT AVG(temperature), MAX(temperature), MIN(temperature)
FROM iot.device_data
WHERE product_key = 'pk1'
  AND ts >= NOW() - 1d;

-- 按标签过滤查询
SELECT device_name, AVG(temperature) as avg_temp
FROM iot.device_data
WHERE location = 'factory-A'
  AND ts >= NOW() - 7d
GROUP BY device_name;

-- 降采样查询 (按 5 分钟窗口聚合)
SELECT INTERVAL(ts, 5m) as window_start,
       AVG(temperature) as avg_temp,
       MAX(humidity) as max_hum
FROM iot.device_pk1_dev001
WHERE ts >= NOW() - 24h
  AND ts < NOW()
INTERVAL(5m);

命令下发

平台向设备下发指令，支持同步 RPC、异步下行和批量下发三种模式。

同步 RPC

设备在线时，平台通过 MQTT 发布消息并等待设备应答，适用于需要即时响应的场景。

RPC 流程

平台                           设备
 |                              |
 |-- MQTT /rpc/invoke/+dev1 -->|
 |   Request ID: R1            |
 |   Method: setSwitch         |
 |   Params: {"target":"off"}  |
 |                              |
 |<-- MQTT /rpc/response/R1 ---|
 |   Code: 200                 |
 |   Data: {"result":"ok"}     |
 |                              |
 | 同步等待,超时时间 5s       |
 | (若超时未收到应答则返回超时) |

同步 RPC 实现

java

@Service
public class SyncRpcService {

    private static final long DEFAULT_TIMEOUT_MS = 5000;
    private final Map<String, CompletableFuture<RpcResponse>> pendingRequests
        = new ConcurrentHashMap<>();

    @Autowired
    private MqttGateway mqttGateway;

    public RpcResponse invoke(String productKey, String deviceName,
                                String method, JsonObject params) {
        return invoke(productKey, deviceName, method, params, DEFAULT_TIMEOUT_MS);
    }

    public RpcResponse invoke(String productKey, String deviceName,
                                String method, JsonObject params, long timeoutMs) {
        String requestId = UUID.randomUUID().toString();
        String topic = "/rpc/invoke/" + productKey + "/" + deviceName;

        // 构建请求消息
        JsonObject request = new JsonObject();
        request.addProperty("id", requestId);
        request.addProperty("method", method);
        request.add("params", params);

        // 注册异步回调
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        pendingRequests.put(requestId, future);

        try {
            // 发送 MQTT 消息
            mqttGateway.sendToTopic(topic, request.toString(), 1);

            // 同步等待应答
            RpcResponse response = future.get(timeoutMs, TimeUnit.MILLISECONDS);
            return response;
        } catch (TimeoutException e) {
            pendingRequests.remove(requestId);
            throw new RpcTimeoutException("device no response within " + timeoutMs + "ms");
        } catch (Exception e) {
            pendingRequests.remove(requestId);
            throw new RpcException("rpc invoke failed", e);
        }
    }

    // 设备端应答回调 (MQTT listener)
    public void onRpcResponse(String requestId, String payload) {
        CompletableFuture<RpcResponse> future = pendingRequests.remove(requestId);
        if (future != null) {
            RpcResponse response = JsonParser.parseString(payload).getAsJsonObject();
            future.complete(response);
        }
    }
}

设备端处理

// 嵌入式设备端 (C 语言伪代码)
void on_rpc_message(const char* topic, const char* payload) {
    // 解析请求
    cJSON* root = cJSON_Parse(payload);
    char* request_id = cJSON_GetObjectItem(root, "id")->valuestring;
    char* method = cJSON_GetObjectItem(root, "method")->valuestring;

    cJSON* result = NULL;
    int code = 200;

    if (strcmp(method, "setSwitch") == 0) {
        cJSON* params = cJSON_GetObjectItem(root, "params");
        char* target = cJSON_GetObjectItem(params, "target")->valuestring;
        // 执行开关操作
        gpio_write(SWITCH_PIN, strcmp(target, "on") == 0 ? HIGH : LOW);
        result = cJSON_CreateObject();
        cJSON_AddStringToObject(result, "result", "ok");
    } else {
        code = 400;
        result = cJSON_CreateObject();
        cJSON_AddStringToObject(result, "error", "unknown method");
    }

    // 构建应答消息
    cJSON* response = cJSON_CreateObject();
    cJSON_AddStringToObject(response, "id", request_id);
    cJSON_AddNumberToObject(response, "code", code);
    cJSON_AddItemToObject(response, "data", result);

    char* resp_str = cJSON_Print(response);
    // 发布到应答 topic
    char resp_topic[128];
    snprintf(resp_topic, sizeof(resp_topic), "/rpc/response/%s", request_id);
    mqtt_publish(resp_topic, resp_str, 1);
    free(resp_str);
    cJSON_Delete(root);
}

异步下行

设备不在线时，将命令缓存到设备影子，设备上线后从影子获取 desired 状态并执行。

异步下发流程

应用                         平台                          设备
 |                            |                            |
 |-- HTTP PUT /shadow/desired|                            |
 |   payload: {"switch":"off"}|                            |
 |                            |                            |
 |                  1. 更新影子 desired                     |
 |                  2. 检查设备在线状态                      |
 |                            |                            |
 |                  3. 设备在线?                            |
 |                     YES -> 立即推送 desired delta        |
 |                     NO  -> 等待设备上线                  |
 |                            |                            |
 |                            |<-- MQTT connect -----------|
 |                            |                            |
 |                  4. 设备上线回调                         |
 |                  5. 推送缓存的 desired delta             |
 |                            |                            |
 |                            |-- delta: {"switch":"off"}->|
 |                            |                            |
 |                            |<-- 执行完成 ----------------|
 |                            |                            |
 |                  6. 更新 reported,清除已执行 desired      |

异步命令管理

java

@Entity
@Table(name = "async_command")
public class AsyncCommand {

    @Id
    private String commandId;
    private String productKey;
    private String deviceName;
    private String method;
    @Column(columnDefinition = "JSON")
    private String params;
    private Integer commandStatus;  // 0-待下发 1-已下发 2-已执行 3-超时 4-失败
    private Long ttlSeconds;
    private Date gmtCreate;
    private Date gmtModified;
}

sql

CREATE TABLE async_command (
    command_id     VARCHAR(64)   PRIMARY KEY,
    product_key    VARCHAR(32)   NOT NULL,
    device_name    VARCHAR(64)   NOT NULL,
    method         VARCHAR(64)   NOT NULL,
    params         JSON          NOT NULL,
    command_status TINYINT       NOT NULL DEFAULT 0 COMMENT '0-待下发 1-已下发 2-已执行 3-超时',
    ttl_seconds    INT           NOT NULL DEFAULT 86400,
    gmt_create     DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_modified   DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_device_status (product_key, device_name, command_status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='异步命令表';

物联网设备接入与通信 ​

设备接入认证 ​

一机一密 ​

认证流程 ​

HMAC 签名示例 ​

建表模型 ​

一型一密 ​

自动注册流程 ​

一型一密配置 ​

X.509 证书认证 ​

认证架构 ​

TLS 双向认证握手 ​

证书管理 ​

Spring Boot mTLS 配置 ​

证书校验示例 ​

设备注册流程 ​

预注册 ​

自动注册 ​

批量导入 ​

三种注册方式对比 ​

设备影子 ​

影子文档结构 ​

影子文档存储 ​

影子状态同步机制 ​

状态更新流程 ​

设备上报状态 ​

平台端影子更新 ​

desired 下发 ​

影子缓存优化 ​

缓存架构 ​

缓存实现 ​

断连恢复优化 ​

数据上报 ​

消息格式规范 ​

JSON 格式 ​

二进制格式 ​

Protobuf 格式 ​

TLV 编解码示例 ​

数据解析脚本 ​

脚本引擎架构 ​

脚本管理 ​

Groovy 脚本示例 ​

脚本引擎执行 ​

上报频率控制 ​

设备端上报间隔 ​

平台端限流 ​

降级策略 ​

时序数据存储 ​

超级表设计 ​

数据写入 ​

典型查询 ​

命令下发 ​

同步 RPC ​

RPC 流程 ​

同步 RPC 实现 ​

设备端处理 ​

异步下行 ​

异步下发流程 ​

异步命令管理 ​

物联网设备接入与通信

设备接入认证

一机一密

认证流程

HMAC 签名示例

建表模型

一型一密

自动注册流程

一型一密配置

X.509 证书认证

认证架构

TLS 双向认证握手

证书管理

Spring Boot mTLS 配置

证书校验示例

设备注册流程

预注册

自动注册

批量导入

三种注册方式对比

设备影子

影子文档结构

影子文档存储

影子状态同步机制

状态更新流程

设备上报状态

平台端影子更新

desired 下发

影子缓存优化

缓存架构

缓存实现

断连恢复优化

数据上报

消息格式规范

JSON 格式

二进制格式

Protobuf 格式

TLV 编解码示例

数据解析脚本

脚本引擎架构

脚本管理

Groovy 脚本示例

脚本引擎执行

上报频率控制

设备端上报间隔

平台端限流

降级策略

时序数据存储

超级表设计

数据写入

典型查询

命令下发

同步 RPC

RPC 流程

同步 RPC 实现

设备端处理

异步下行

异步下发流程

异步命令管理