工业现场的数据孤岛问题

2026年的工业互联网项目中，最常见的需求不是"采集什么数据"，而是"怎么把分散在不同设备、不同协议、不同位置的数据统一采集上来"。一个典型工厂的车间里，PLC走的是Modbus TCP，仪表走的是RS485的Modbus RTU，环保监测设备走的是MQTT，老设备只有4-20mA模拟量输出。

边缘计算网关就是解决这个"协议碎片化"问题的中间层。它向下对接各种工业协议，把数据统一成标准格式，通过4G/5G网络上传到工业互联网平台。

边缘网关的架构设计

硬件架构

一个完整的工业边缘计算网关需要以下硬件组件：

组件	选型	作用
主控	STM32H7或Linux SoC	协议转换与数据处理
通信模组	4G Cat.4/Cat.1或5G	蜂窝网络上行
以太网口	2-4个千兆口	对接PLC和本地网络
串口	RS232/RS485	对接仪表和老设备
存储	eMMC + SD卡	断点续传缓存
电源	宽压输入9-36V	适应工业供电环境

软件架构分层

网关的软件架构从下到上分为五层：

层级	模块	职责
协议适配层	Modbus/OPC UA/MQTT驱动	采集不同设备数据
数据标准化层	数据模型映射	统一数据格式
边缘计算层	规则引擎+过滤	协议解析、报警触发
传输层	MQTT/HTTP	上行通信
管理层	远程配置+OTA	设备运维

Modbus协议采集实战

Modbus是工业现场最常用的通信协议。Modbus RTU基于串口，Modbus TCP基于以太网。网关需要同时支持两种形态。

Modbus TCP采集实现

用Python的pymodbus库读取PLC寄存器数据：

from pymodbus.client import ModbusTcpClient
import json
import time

class ModbusCollector:
    def __init__(self, host, port=502):
        self.client = ModbusTcpClient(host, port=port)
        self.connected = False

    def connect(self):
        self.connected = self.client.connect()
        return self.connected

    def read_holding_registers(self, unit_id, start_addr, count):
        """读取保持寄存器"""
        if not self.connected:
            self.connect()

        result = self.client.read_holding_registers(
            address=start_addr,
            count=count,
            slave=unit_id
        )

        if result.isError():
            return None
        return result.registers

    def collect_all(self, device_config):
        """按配置采集所有设备数据"""
        collected = {}
        for device in device_config:
            unit_id = device['unit_id']
            registers = device['registers']
            for reg in registers:
                data = self.read_holding_registers(
                    unit_id, reg['address'], reg['count']
                )
                if data:
                    tag_name = reg['tag_name']
                    # 数据类型转换
                    if reg['type'] == 'float':
                        import struct
                        value = struct.unpack('>f',
                            struct.pack('>HH', data[0], data[1]))[0]
                    elif reg['type'] == 'int16':
                        value = data[0]
                        if value > 32767:
                            value -= 65536
                    else:
                        value = data[0]

                    collected[tag_name] = {
                        'value': value,
                        'timestamp': int(time.time()),
                        'unit': reg.get('unit', '')
                    }
        return collected

Modbus RTU采集实现

对于串口设备，用Modbus RTU：

from pymodbus.client import ModbusSerialClient

class ModbusRtuCollector:
    def __init__(self, port='/dev/ttyUSB1', baudrate=9600):
        self.client = ModbusSerialClient(
            port=port,
            baudrate=baudrate,
            bytesize=8,
            parity='N',
            stopbits=1,
            timeout=1
        )

    def read_device(self, unit_id, start_addr, count):
        if self.client.connect():
            result = self.client.read_holding_registers(
                address=start_addr,
                count=count,
                slave=unit_id
            )
            self.client.close()
            if not result.isError():
                return result.registers
        return None

4G/5G通信模组驱动

工业网关通过4G/5G模组与云端通信。主流模组厂商有移远、芯讯通等，它们通过AT指令控制。网关主控通过UART与4G模组通信，初始化流程：

// 4G模组初始化（以移远EC20为例）
#include "uart_driver.h"

int modem_init(const char *port, int baudrate)
{
    int fd = uart_open(port, baudrate);
    if (fd < 0) return -1;

    // 1. 测试AT通信
    uart_send_at(fd, "AT", 1000);
    if (!uart_wait_response(fd, "OK", 2000)) {
        return -1;
    }

    // 2. 检查SIM卡
    uart_send_at(fd, "AT+CPIN?", 1000);
    if (!uart_wait_response(fd, "+CPIN: READY", 3000)) {
        return -2;
    }

    // 3. 查询网络注册状态
    uart_send_at(fd, "AT+CREG?", 1000);
    if (!uart_wait_response(fd, "+CREG: 0,1", 5000)) {
        return -3;
    }

    // 4. 查询信号强度
    uart_send_at(fd, "AT+CSQ", 1000);
    // 解析+CSQ: rssi,ber
    char *resp = uart_read_response(fd, 2000);
    int rssi = parse_csq(resp);

    // 5. 配置PDP上下文
    uart_send_at(fd, "AT+CGDCONT=1,\"IP\",\"CMNET\"", 1000);

    // 6. 激活PDP
    uart_send_at(fd, "AT+NETOPEN=1", 10000);

    return fd;
}

这段AT指令流程是所有4G模组的通用初始化路径。不同芯片的AT指令略有差异，但核心流程一致：测试通信→检查SIM→查询注册→配置APN→激活数据连接。

断点续传与数据缓存

工业现场网络不稳定，4G信号波动时数据上传会失败。网关必须有本地缓存和断点续传能力。

设计思路：采集到的数据先写入本地SQLite缓存，上传成功后删除，上传失败则保留并标记重试次数：

import sqlite3
import json
import requests
from datetime import datetime

class DataCache:
    def __init__(self, db_path='/data/iot_cache.db'):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS cache (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                device_id TEXT,
                tag_name TEXT,
                value TEXT,
                timestamp INTEGER,
                retry_count INTEGER DEFAULT 0,
                uploaded INTEGER DEFAULT 0
            )
        ''')
        self.conn.commit()

    def store(self, device_id, tag_name, value, timestamp):
        self.conn.execute(
            'INSERT INTO cache (device_id, tag_name, value, timestamp) VALUES (?,?,?,?)',
            (device_id, tag_name, json.dumps(value), timestamp)
        )
        self.conn.commit()

    def upload_pending(self, api_url, batch_size=50):
        rows = self.conn.execute(
            'SELECT * FROM cache WHERE uploaded=0 ORDER BY timestamp LIMIT ?',
            (batch_size,)
        ).fetchall()

        if not rows:
            return 0

        payload = []
        for row in rows:
            payload.append({
                'device_id': row[1],
                'tag_name': row[2],
                'value': json.loads(row[3]),
                'timestamp': row[4]
            })

        try:
            resp = requests.post(api_url, json=payload, timeout=10)
            if resp.status_code == 200:
                ids = [r[0] for r in rows]
                self.conn.executemany(
                    'DELETE FROM cache WHERE id=?',
                    [(i,) for i in ids]
                )
                self.conn.commit()
                return len(ids)
        except Exception as e:
            # 增加重试计数
            for row in rows:
                self.conn.execute(
                    'UPDATE cache SET retry_count=retry_count+1 WHERE id=?',
                    (row[0],)
                )
            self.conn.commit()

        return 0

边缘规则引擎

网关不仅仅是数据透传管道，还需要在边缘端做规则处理。典型的边缘计算规则包括：阈值报警、数据变化触发、公式计算。

class EdgeRuleEngine:
    def __init__(self):
        self.rules = []

    def add_rule(self, tag_name, condition, action):
        self.rules.append({
            'tag': tag_name,
            'condition': condition,
            'action': action
        })

    def evaluate(self, data):
        alerts = []
        for rule in self.rules:
            if rule['tag'] in data:
                value = data[rule['tag']]['value']
                if rule['condition'](value):
                    alerts.append(rule['action'](value))
        return alerts

# 规则配置示例
engine = EdgeRuleEngine()

# 温度超过80度报警
engine.add_rule(
    'temperature',
    lambda v: v > 80,
    lambda v: {'type': 'over_temp', 'value': v, 'severity': 'critical'}
)

# 功率波动超过20%触发
engine.add_rule(
    'power',
    lambda v: abs(v - 50.0) > 10.0,
    lambda v: {'type': 'power_anomaly', 'value': v, 'severity': 'warning'}
)

远程运维与串口调试

工业网关部署在客户现场，远程运维是刚需。当4G通信模组出现异常时，运维人员需要通过串口对模组做AT指令诊断。

虎王科技的随身WiFi硬件调试工具（gitee.com/zesso/hardware_tool）在这种远程串口调试场景下非常实用。它支持中兴微、ASR、展锐等多种4G/5G芯片的串口通信调试，Web化界面让运维人员在浏览器里就能发送AT指令查询模组状态、信号强度和网络注册情况，不需要派人到现场接USB-TTL线。

这种"Web化串口调试 + 边缘网关"的组合，构成了工业物联网运维的完整工具链：网关负责数据采集和边缘计算，调试工具负责通信模组诊断，两者配合实现了从设备端到通信链路的远程可观测性。

行业趋势与总结

2026年工业物联网的核心趋势是从"连接设备"走向"覆盖设备、网络、数据和生产系统的整体连接架构"。eRedCap技术在3GPP Release 18标准冻结，2026年开始商用，为工业场景提供更低成本和功耗的5G连接选项。

边缘计算网关在这条链路中的位置是确定的：它是工业现场数据进入数字世界的第一个关口。协议转换、边缘计算、断点续传、远程运维，每一环都直接影响数据质量和系统可靠性。

工业物联网方向的同学，网关开发涉及的知识面很广，这篇把核心模块都覆盖到了。觉得有帮助的收藏下方便查阅。有做工业网关的同行，评论区聊聊你们用的什么主控和通信模组方案，交流下选型经验。