MySQL 数据库设计：10张表的工业物联网 Schema

STM32 Mill 是一套面向 STM32 流量计设备的工业物联网控制系统。后端服务 stm32-mill-api 负责接收 MQTT 上行的遥测、告警、命令应答，再通过 HTTP/WS 暴露给前端。整套数据层落在 MySQL 上，共 10 张表，覆盖遥测、告警、配置、组态、审计五个领域。

这篇文章记录这 10 张表的设计思路，重点说清楚几个决策为什么这么做。

一、程序化建表，而不是手写迁移

数据库初始化分两步走。

第一步是 deploy/sql/stm32-mill-init.sql，只负责建库、建用户、授权：

CREATE DATABASE IF NOT EXISTS stm32_mill
  CHARACTER SET utf8mb4
  COLLATE utf8mb4_unicode_ci;

CREATE USER IF NOT EXISTS 'stm32_mill_app'@'%'
  IDENTIFIED BY '${STM32_MILL_DB_PASSWORD}';

GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, INDEX, ALTER, REFERENCES
  ON stm32_mill.*
  TO 'stm32_mill_app'@'%';

密码用 ${STM32_MILL_DB_PASSWORD} 占位符，部署时由 envsubst 注入。历史版本曾把密码硬编码进脚本，凡是 git 历史里出现过的密码都应轮换。应用用户只给 stm32_mill 库的增删改查和 DDL 权限，跨库污染不可能发生。

第二步才是关键。db.js 的 createSchema(pool) 在应用启动连接池成功后，依次执行 10 条 CREATE TABLE IF NOT EXISTS：

async function initializePool(pool, config, status) {
  for (let attempt = 1; attempt <= DB_CONNECT_ATTEMPTS; attempt += 1) {
    try {
      await pool.query("SELECT 1");
      await createSchema(pool);
      status.connected = true;
      return;
    } catch (error) {
      // 重试 30 次，每次间隔 2 秒
      await delay(DB_CONNECT_RETRY_MS);
    }
  }
}

容器编排里 MySQL 起得比 api 慢是常态。重试 30 次、每次 2 秒，总计 1 分钟的窗口足够 MySQL 完成初始化。表建好后服务才对外暴露端口，避免「能连 API 但查不到表」的中间态。

这种程序化建表的好处很直接：新机器只要起 MySQL 容器、灌 init.sql、起 api 容器，schema 自动到位，不需要单独跑 migration 工具。代价是表结构演进要靠下面的 ensureColumn 兜底。

二、10 张表总览

表名	领域	主键 / 唯一键	关键索引
measurements	遥测	id	(device_id, received_at)、(device_id, payload_ts)
users	用户	username	无
command_acks	命令应答	id	(device_id, received_at)、(cmd_seq)
alarms	告警事件	id	(device_id, received_at)、(code, received_at)
alarm_states	告警状态	(device_id, code)	(device_id, recovered_at, last_triggered_at)、(updated_at)
relay_rule_configs	继电器规则	device_id	(updated_at)
alarm_threshold_configs	告警阈值	device_id	(updated_at)
scada_layouts	组态布局	(device_id, status)	(updated_at)
scada_layout_versions	组态版本	(device_id, version_no)	(device_id, created_at)
operation_audit	操作审计	id	(created_at)、(device_id, created_at)、(username, created_at)

下面按领域展开。

三、measurements：JSON 列 + 索引列的混合策略

遥测表是写入量最大的一张。设备每秒上报一帧，字段包括流量、累计流量、称重、继电器位图、心跳计数、有效位、状态位、温度数组。完整 payload 是 JSON，但查询过滤只看 device_id 和时间。

CREATE TABLE IF NOT EXISTS measurements (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  device_id VARCHAR(64) NOT NULL,
  topic VARCHAR(255) NOT NULL,
  payload_ts BIGINT NULL,
  seq BIGINT NULL,
  flow DOUBLE NULL,
  total_value DOUBLE NULL,
  weight BIGINT NULL,
  relay_do INT NULL,
  relay_di INT NULL,
  heart_count BIGINT NULL,
  valid_mask INT NULL,
  status_bits INT NULL,
  temperature_json JSON NULL,
  payload JSON NOT NULL,
  received_at DATETIME(3) NOT NULL,
  INDEX idx_measurements_device_received (device_id, received_at),
  INDEX idx_measurements_device_payload_ts (device_id, payload_ts)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci

这里采用混合策略：完整 payload 存 JSON 列保证可追溯，关键字段单独提取为强类型列给查询和聚合用。温度数组单独存 temperature_json，因为它是变长数组，没法拍平成列。

为什么不全用 JSON 列加函数索引？两个原因。一是 MySQL 8.0 之前的版本对 JSON 函数索引支持不完整。二是强类型列在聚合查询和导出时类型转换成本更低，SELECT flow, total_value FROM measurements 比 SELECT payload->>'$.flow' 直观。

双索引设计：

• idx_measurements_device_received (device_id, received_at) 服务于历史查询，前端按设备 + 时间范围拉曲线
• idx_measurements_device_payload_ts (device_id, payload_ts) 服务于设备时间对齐场景，payload_ts 是设备端时钟产生的时间戳

received_at 用 DATETIME(3) 保留毫秒，因为工业设备一秒多帧的情况下，秒级精度会让排序乱掉。

游标分页用 received_at DESC, id DESC 双字段：

clauses.push("(received_at < ? OR (received_at = ? AND id < ?))");

不能只依赖 id，因为早期版本按设备 seq 去重更新旧行，会出现 id 旧但 received_at 新的历史点。

四、alarms + alarm_states：事件流和状态机分离

告警分两张表。alarms 是只追加的事件流，每次触发或恢复都写一行；alarm_states 是当前状态，按 device_id + code 唯一。

CREATE TABLE IF NOT EXISTS alarms (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  device_id VARCHAR(64) NOT NULL,
  topic VARCHAR(255) NOT NULL,
  payload_ts BIGINT NULL,
  seq BIGINT NULL,
  code VARCHAR(64) NOT NULL,
  severity VARCHAR(32) NULL,
  alarm_value DOUBLE NULL,
  firmware_version VARCHAR(128) NULL,
  payload JSON NOT NULL,
  received_at DATETIME(3) NOT NULL,
  INDEX idx_alarms_device_received (device_id, received_at),
  INDEX idx_alarms_code_received (code, received_at)
)

alarms 有两个索引：(device_id, received_at) 按设备查历史，(code, received_at) 按告警码跨设备查趋势。比如「所有设备的温度超限告警最近 24 小时分布」，走第二个索引。

CREATE TABLE IF NOT EXISTS alarm_states (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  device_id VARCHAR(64) NOT NULL,
  code VARCHAR(64) NOT NULL,
  severity VARCHAR(32) NULL,
  alarm_value DOUBLE NULL,
  payload JSON NOT NULL,
  first_triggered_at DATETIME(3) NOT NULL,
  last_triggered_at DATETIME(3) NOT NULL,
  recovered_at DATETIME(3) NULL,
  acknowledged_at DATETIME(3) NULL,
  acknowledged_by VARCHAR(64) NULL,
  updated_at DATETIME(3) NOT NULL,
  UNIQUE KEY uniq_alarm_states_device_code (device_id, code),
  INDEX idx_alarm_states_active_device (device_id, recovered_at, last_triggered_at),
  INDEX idx_alarm_states_updated (updated_at)
)

alarm_states 的设计要点：

• UNIQUE KEY uniq_alarm_states_device_code (device_id, code)：同一设备同一告警码只能有一条活跃状态，用 INSERT ... ON DUPLICATE KEY UPDATE 做 upsert
• idx_alarm_states_active_device (device_id, recovered_at, last_triggered_at)：活跃告警查询走这个索引，WHERE recovered_at IS NULL AND device_id = ? 能直接命中
• first_triggered_at 和 last_triggered_at 分开存：前者记录告警首次发生时间用于统计 MTTR，后者每次更新走刷新

活跃告警列表按严重度排序：

ORDER BY FIELD(severity, 'danger', 'error', 'warning', 'info') ASC,
         last_triggered_at DESC, id DESC

FIELD() 函数把枚举值映射成数字排序，比 CASE WHEN 简洁。

五、配置表：device_id 作主键

relay_rule_configs 和 alarm_threshold_configs 都是按设备存配置，device_id 直接做主键：

CREATE TABLE IF NOT EXISTS relay_rule_configs (
  device_id VARCHAR(64) NOT NULL PRIMARY KEY,
  status VARCHAR(32) NOT NULL,
  config JSON NOT NULL,
  cfg_version INT NULL,
  cfg_crc32 VARCHAR(8) NULL,
  cfg_len INT NULL,
  cmd_seq BIGINT NULL,
  topic VARCHAR(255) NULL,
  last_ack JSON NULL,
  acked_at DATETIME(3) NULL,
  updated_at DATETIME(3) NOT NULL,
  INDEX idx_relay_rule_configs_updated (updated_at)
)

relay_rule_configs 比 alarm_threshold_configs 多了几个字段。因为继电器规则要下发给设备，需要跟踪下发状态：

• status 取值 pending、acked、rejected，反映设备是否确认收到
• cfg_version、cfg_crc32、cfg_len 用于设备端校验配置完整性
• cmd_seq 关联下发的命令序号
• last_ack 存设备最后一次应答的完整 payload

下发命令收到 ack 时，updateRelayRulesAck 会更新状态：

const status = payload.result === "ok" ? "acked" : "rejected";

阈值表更简单，只有 device_id、config、updated_at 三个字段，因为告警阈值是服务端判定用的，不需要设备确认。

六、scada_layouts：draft / published 双份设计

组态画面表按设备拆成草稿和发布两份：

CREATE TABLE IF NOT EXISTS scada_layouts (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  device_id VARCHAR(64) NOT NULL,
  status VARCHAR(32) NOT NULL,
  name VARCHAR(128) NOT NULL,
  version INT UNSIGNED NOT NULL DEFAULT 1,
  canvas JSON NOT NULL,
  elements JSON NOT NULL,
  published_at DATETIME(3) NULL,
  updated_at DATETIME(3) NOT NULL,
  UNIQUE KEY uniq_scada_layouts_device_status (device_id, status),
  INDEX idx_scada_layouts_updated (updated_at)
)

UNIQUE KEY uniq_scada_layouts_device_status (device_id, status) 是核心约束。同一设备同一状态只能有一份，用 ON DUPLICATE KEY UPDATE 做 upsert，version 每次自增：

ON DUPLICATE KEY UPDATE
  version = version + 1,
  canvas = VALUES(canvas),
  elements = VALUES(elements),
  ...

为什么不用一张表加 is_published 字段？因为运行态页面（前端看板）需要高频读 published，编辑态高频写 draft。拆成两行后读写互不阻塞，也不需要 WHERE is_published = 1 这种过滤条件。

getScadaLayout 默认读 published：

async getScadaLayout(device, status = "published") { ... }

前端编辑器先调 saveScadaLayout 存 draft，预览无误后调 publishScadaLayout 复制到 published。

七、scada_layout_versions：版本快照回滚

每次发布都存一份快照：

CREATE TABLE IF NOT EXISTS scada_layout_versions (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  device_id VARCHAR(64) NOT NULL,
  version_no INT UNSIGNED NOT NULL,
  name VARCHAR(128) NOT NULL,
  canvas_json JSON NOT NULL,
  elements_json JSON NOT NULL,
  created_at DATETIME(3) NOT NULL,
  UNIQUE KEY uniq_scada_layout_versions_device_no (device_id, version_no),
  INDEX idx_scada_layout_versions_device_created (device_id, created_at)
)

发布流程：

async publishScadaLayout(layout) {
  const published = await upsertScadaLayout(pool, {
    ...layout,
    status: "published",
    publishedAt: layout.publishedAt || new Date().toISOString(),
  });
  if (published) {
    await insertScadaLayoutVersion(pool, published);
    await pruneScadaLayoutVersions(pool, published.device, 20);
  }
  return published;
}

pruneScadaLayoutVersions 保留最近 20 个版本，老的删掉。版本号自增：

SELECT COALESCE(MAX(version_no), 0) + 1 AS next_version_no
FROM scada_layout_versions
WHERE device_id = ?

回滚操作把指定版本的 canvas 和 elements 复制回 draft，让用户确认后再发布：

async function rollbackScadaLayout(pool, { device, versionId }) {
  const [rows] = await pool.query(
    `SELECT id, device_id, version_no, name, canvas_json, elements_json, created_at
     FROM scada_layout_versions
     WHERE device_id = ? AND id = ?
     LIMIT 1`,
    [dev, id],
  );
  if (!rows.length) return null;

  const version = rows[0];
  return upsertScadaLayout(pool, {
    device: version.device_id,
    status: "draft",  // 回滚到草稿，不直接覆盖 published
    name: version.name,
    canvas: parseJson(version.canvas_json) || {},
    elements: parseJson(version.elements_json) || [],
    updatedAt: new Date().toISOString(),
  });
}

回滚到 draft 而不是直接覆盖 published，是避免误操作把线上画面冲掉。用户回滚后还要再点一次发布才生效。

八、operation_audit：三个索引覆盖三种查询

操作审计表记录所有高风险操作（下发配置、修改阈值、发布组态、用户管理等）：

CREATE TABLE IF NOT EXISTS operation_audit (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  username VARCHAR(64) NOT NULL,
  role VARCHAR(32) NOT NULL,
  device_id VARCHAR(64) NULL,
  action VARCHAR(80) NOT NULL,
  request_json JSON NULL,
  result_json JSON NULL,
  result VARCHAR(32) NOT NULL,
  created_at DATETIME(3) NOT NULL,
  INDEX idx_operation_audit_created (created_at),
  INDEX idx_operation_audit_device_created (device_id, created_at),
  INDEX idx_operation_audit_username_created (username, created_at)
)

三个索引对应三种审计场景：

• idx_operation_audit_created (created_at)：按时间倒序拉全局审计日志
• idx_operation_audit_device_created (device_id, created_at)：查某台设备的操作历史
• idx_operation_audit_username_created (username, created_at)：查某个用户的操作历史

request_json 和 result_json 存完整请求和响应。为防止单条记录过大，超过 12000 字符的会被截断：

function normalizeAuditJson(value) {
  try {
    const text = JSON.stringify(value);
    if (text.length > 12000) {
      return JSON.stringify({ truncated: true, size: text.length });
    }
    return text;
  } catch {
    return JSON.stringify({ value: String(value) });
  }
}

device_id 可空，因为有些操作（如用户管理）不针对具体设备。

九、ensureColumn：补丁式列迁移

程序化建表解决了首次部署，但字段演进怎么办？ensureColumn 函数处理后续加列：

async function ensureColumn(pool, table, column, definition) {
  const [rows] = await pool.query(
    `SELECT COUNT(*) AS count
     FROM INFORMATION_SCHEMA.COLUMNS
     WHERE TABLE_SCHEMA = DATABASE()
       AND TABLE_NAME = ?
       AND COLUMN_NAME = ?`,
    [table, column],
  );

  if (Number(rows[0]?.count) > 0) {
    return;  // 列已存在，跳过
  }

  await pool.query(
    `ALTER TABLE ${quoteIdentifier(table)} ADD COLUMN ${quoteIdentifier(column)} ${definition}`,
  );
}

createSchema 末尾调了一次：

await ensureColumn(pool, "users", "role", "VARCHAR(32) NOT NULL DEFAULT 'admin'");

这是因为 users 表的 role 字段是后期加的。老库的 users 表没有这一列，ensureColumn 会补上；新库的 CREATE TABLE IF NOT EXISTS 已经包含 role，ensureColumn 查到列存在直接跳过。

同样思路的还有 dropUniqueDeviceSeq，用来清理早期版本遗留的唯一索引 uniq_measurements_device_seq：

async function dropUniqueDeviceSeq(pool, table, indexName) {
  const [rows] = await pool.query(
    `SELECT COUNT(*) AS count
     FROM INFORMATION_SCHEMA.STATISTICS
     WHERE TABLE_SCHEMA = DATABASE()
       AND TABLE_NAME = ?
       AND INDEX_NAME = ?`,
    [table, indexName],
  );
  if (Number(rows[0]?.count) <= 0) {
    return;
  }
  await pool.query(
    `ALTER TABLE ${quoteIdentifier(table)} DROP INDEX ${quoteIdentifier(indexName)}`,
  );
}

早期版本按 device_id + seq 做唯一约束去重，后来发现设备重启后 seq 会回绕，唯一约束反而挡住了合法上行，于是改成只追加不去重，索引也一起删掉。

十、配置层：enabled 开关和连接池

config.js 里 DB 配置的关键字段：

db: {
  enabled: Boolean(emptyToUndefined(env.DB_HOST)),
  host: env.DB_HOST || "stm32-mill-mysql",
  port: parseInteger(env.DB_PORT, 3306),
  database: env.DB_NAME || "stm32_mill",
  user: env.DB_USER || "stm32_mill_app",
  password: emptyToUndefined(env.DB_PASSWORD),
  connectionLimit: parseInteger(env.DB_CONNECTION_LIMIT, 5),
}

enabled 的判定逻辑是：只要 DB_HOST 非空就启用。没配 DB 时服务退化为纯内存模式，登录会返回 503，历史接口报 disabled。这样开发本地调试不需要起 MySQL 也能跑。

连接池 connectionLimit 默认 5，单实例够用。timezone: "Z" 强制 UTC，避免 MySQL 服务器时区配置不一致导致时间错乱。

十一、一些细节

JSON 序列化统一走 JSON.stringify。MySQL 的 JSON 列接收字符串会自动校验格式，非法 JSON 会直接报错。temperature_json 存数组时 JSON.stringify([1,2,3]) 得到 "[1,2,3]"，MySQL 会解析成 JSON 数组。

时间字段一律 DATETIME(3)。毫秒精度对工业数据有必要，一秒多帧的情况下秒级精度会让排序失真。

BIGINT UNSIGNED 主键。遥测表写入量大，32 位自增 id 几年就溢出，直接上 64 位。

utf8mb4 + utf8mb4_unicode_ci。utf8 在 MySQL 里是 3 字节截断版，存不了 emoji 和部分生僻字，统一用 utf8mb4。unicode_ci 排序规则比 general_ci 更符合 Unicode 标准。

mapXxxRow 函数统一处理出参。数据库下划线命名和前端驼峰命名之间有一层映射，避免 SQL 字段名直接漏到 API 响应里。

小结

这套 schema 的设计取向是「能跑、能查、能演进」。

• 程序化建表 + ensureColumn 牺牲了严格的 migration 审计，换来部署门槛的降低
• JSON 列 + 索引列的混合策略在查询性能和数据可追溯之间做了取舍
• 双索引覆盖设备维度和时间维度的查询，单表不建过多索引
• 组态布局的 draft/published 双份设计避免了读写互斥
• 版本快照保留 20 份，回滚走 draft 不直接覆盖 published
• 操作审计三索引对应三种审计视角

工业物联网场景的特点是写多读少、字段相对固定但偶尔会扩展、历史数据量大但实时性要求高。这套 schema 在 10 张表的规模上把这些诉求覆盖住了，没有引入过度的抽象。