技术博客：武汉企业MES系统开发中的实时数据采集方案

项目背景

客户是武汉经济技术开发区（沌口）的一家汽车零部件一级供应商，主要为东风本田供应内饰件。MES系统的核心需求之一是实时采集生产线设备的运行数据，包括：

数控机床的主轴转速、进给速率、负载电流
注塑机的料筒温度、注射压力、模具开合状态
传送带的运行速度、故障报警
机器人的关节角度、工作状态

设备品牌混杂：Fanuc、Siemens、Mitsubishi、台达等，协议各异。

技术挑战

协议多样：Modbus TCP、OPC UA、Fanuc FOCAS、西门子S7协议等
实时性要求高：数据采集频率100ms-1s，延迟<100ms
可靠性要求高：7×24小时不间断运行，数据丢失率<0.01%
网络环境复杂：车间电磁干扰严重，有线/无线混合组网
安全性要求：生产网络与办公网络隔离，数据单向传输

方案架构

┌──────────────────────────────────────────────┐
│                生产车间层                    │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐        │
│  │数控  │ │注塑  │ │传送带│ │机器人│        │
│  │机床  │ │机   │ │      │ │      │        │
│  └──┬───┘ └──┬───┘ └──┬───┘ └──┬───┘        │
│     │         │         │         │            │
│  ┌──▼─────────▼─────────▼────────▼───┐     │
│  │        工业网关 (Edge Gateway)      │     │
│  │  - 协议转换                        │     │
│  │  - 数据预处理                     │     │
│  │  - 本地缓存                       │     │
│  └──────────────────┬────────────────┘     │
├───────────────────┼────────────────────────┤
│                   │ 防火墙(仅允许MQTT出站)    │
├───────────────────┼────────────────────────┤
│                数据中心层                    │
│  ┌────────────────▼────────────────┐      │
│  │      MQTT Broker (EMQ X)        │      │
│  └────────────────┬────────────────┘      │
│         ┌─────────┼─────────┐               │
│  ┌──────▼────┐ ┌──▼──────┐ ┌▼──────────┐  │
│  │数据处理   │ │实时监控  │ │数据存储   │  │
│  │服务       │ │服务     │ │服务       │  │
│  └───────────┘ └─────────┘ └──────────┘  │
└──────────────────────────────────────────────┘

关键技术实现

1. 协议适配层

使用 Node-RED 作为可视化编排工具，配合自定义节点：

// Modbus TCP 读取示例
[{"id":"modbus-read","type":"modbus-read","dataType":"HoldingRegister","address":"400001","quantity":10,"rate":500},
{"id":"data-transform","type":"function","func":"// 数据转换和标准化\nmsg.payload = msg.payload.map(v => v * 0.1); return msg;"},
{"id":"mqtt-publish","type":"mqtt out","topic":"production/line1/cnc1/data"}]

2. OPC UA 接入

对于支持 OPC UA 的现代设备（如西门子 S7-1500）：

const opcua = require('node-opcua')

async function connectOpcUa(endpointUrl) {
  const client = opcua.OPCUAClient.create({})
  await client.connect(endpointUrl)
  const session = await client.createSession()
  return { client, session }
}

async function readTag(session, nodeId) {
  const dataValue = await session.read({
    nodeId: nodeId,
    attributeId: opcua.AttributeIds.Value
  })
  return dataValue.value.value
}

3. MQTT 消息管道

使用 EMQ X 作为 MQTT Broker：

Topic 设计：{factory}/{line}/{device}/{datatype}
例如：wuhan-dongfeng/line3/injection-mold1/temperature
QoS Level 1（至少送达一次）
消息体：JSON 格式，包含 timestamp、value、quality

4. 时序数据存储

选用 InfluxDB 存储时序数据：

// 写入 InfluxDB
const { InfluxDB, Point } = require('@influxdata/influxdb-client')

const influxDB = new InfluxDB({ url: 'http://localhost:8086', token: 'my-token' })
const writeApi = influxDB.getWriteApi('org','bucket')

function saveDataPoint(measurement, tags, fields, timestamp) {
  const point = new Point(measurement)
    .tag(tags)
    .floatField('value', fields.value)
    .intField('quality', fields.quality)
    .timestamp(new Date(timestamp))
  writeApi.writePoint(point)
}

5. 实时监控大屏

前端使用 ECharts + WebSocket 展示实时数据：

// WebSocket 推送实时数据
wss.on('connection', (ws) => {
  // 订阅感兴趣的数据点
  ws.on('message', (msg) => {
    const topics = JSON.parse(msg).topics
    topics.forEach(topic => {
      mqttClient.subscribe(topic)
    })
  })
})

// MQTT 收到数据后推送给 WebSocket 客户端
mqttClient.on('message', (topic, message) => {
  wss.clients.forEach(client => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(JSON.stringify({ topic, data: JSON.parse(message) }))
    }
  })
})

性能指标

系统上线后的实际运行数据：

指标	目标值	实际值
数据采集频率	≤1秒	200ms-500ms
数据传输延迟	<100ms	30-60ms
数据完整性	≥99.99%	99.997%
系统可用性	≥99.9%	99.95%
并发连接数	≥500	峰值180
存储容量	3年数据	已存12个月，占用2TB

安全措施

生产网络与办公网络之间部署工业防火墙
只允许 MQTT 出站连接（端口1883），禁止入站
数据传输 TLS 加密
操作审计日志（谁在什么时候查看了什么数据）
定期漏洞扫描和安全加固

经验总结

协议标准化：尽量推动设备供应商统一协议（首选 OPC UA）
边缘计算：在网关层做数据预处理，减少传输量和服务器压力
断点续传：网络中断时本地缓存，恢复后自动补传
设备命名规范：建立统一的设备编码体系，避免混乱
渐进式实施：先从一条产线做起，验证后再推广