工业上位机把PLC数据推上MQTT_一_架构与完整链路

工业上位机怎么把 PLC 数据推上 MQTT Broker一从采集到发布的完整链路系列说明这是《工业上位机把 PLC 数据推上 MQTT》三部曲第 1 篇。第 1 篇本篇整体架构与发布链路第 2 篇死区 最小间隔解决上报太频繁、流量烧钱第 3 篇离线队列 QoS1解决断网丢数据、连上补发。三篇都基于 PLCMonitor 真实源码src/comm/mqtt/函数名和字段都能对着看。前阵子去一个水处理站调试甲方信息化的人站在柜子旁边问我“你们这监控软件能不能直接把数据推到我们 MES我们不想自己写 OPC 轮询。” 这种需求这两年越来越多——上云、接 MES、做个手机看板说白了就一句话把 PLC 里的实时数据稳稳地送出去。这篇先把整体链路讲清楚后面两篇再钻两个最容易翻车的地方流量和断网。一、为什么是 MQTT而不是 HTTP 轮询现场很多人第一反应是起个 HTTP 接口让对方来拉。但站在上位机这边轮询有几个难受的点上位机是被动方不知道对方什么时候要只能固定频率吐闲时白忙几千个 tag 全量轮询带宽和 CPU 都浪费断网恢复后对方得自己处理哪段时间没采到对账麻烦。MQTT 是发布/订阅模型上位机只管往 Broker 发谁关心谁订阅。轻量协议头最小 2 字节、支持遗嘱和持久会话、QoS 能保至少一次/恰好一次。工业现场用 Mosquitto、EMQX 或者直接上云阿里云 IoT、AWS IoT Core都行。我们 PLCMonitor 选的是Eclipse Paho 的MQTTAsync异步 C 客户端。为什么不用同步版因为上位机主线程就是 Qt 的 UI 线程同步发送一卡就是界面卡死。异步客户端把网络 IO 扔到库自己的线程回调里再marshal回主线程UI 不堵——这点后面第 3 篇讲线程坑会再提。二、三通道变化、快照、报警为什么要分开一开始我也想过一个 topic 全怼进去。后来发现三类数据的消费方根本不是一拨人change变化通道某个 tag 值变了就发给关心实时跳变的系统组态、实时看板snapshot快照通道按周期把一批 tag 的当前值打包发给需要某个时刻全貌的系统按时入库、翻历史alarm报警通道报警事件单独发给告警平台优先级高、格式独立。分开之后订阅方想看啥订啥Broker 侧也好做权限和留存策略。代码里三个入口清晰// mqttpublisher.cppvoidMqttPublisher::onTagValueChanged(...);// 变化通道voidMqttPublisher::onBatchUpdated(...);// 快照缓存周期打包voidMqttPublisher::publishAlarm(...);// 报警通道MqttConfig里三个开关分开控制common_types.hboolpublishOnChangetrue;// 变化通道boolsnapshotEnabledfalse;// 快照通道默认关按需开boolpublishAlarmstrue;// 报警通道intsnapshotIntervalMs5000;// 快照周期三、主题怎么拼载荷长什么样主题用模板现场可以自己改层级。默认是%{station}/%{device}/%{tag}QString baseTopicplcmonitor;QString topicTemplate%{station}/%{device}/%{tag};最终主题 baseTopic / channel / 解析后的模板。channel是三通道名change/snapshot/alarm模板里的%{station}用m_stationName配置里的站名替换、%{device}取 tagKey 第一段、%{tag}取 tagKey 剩下部分。%{station}仅在站名为空时才回退成字面量station。完整映射示例假设站名配置为一号水处理站tagKey 为DefaultPLC/Main_Plc/回水温度channel change stationName 一号水处理站 ← m_stationName配置项非字面量 device DefaultPLC ← tagKey 第一段 tag Main_Plc/回水温度 ← tagKey 余下部分 topicTemplate %{station}/%{device}/%{tag} baseTopic plcmonitor ⇒ plcmonitor/change/一号水处理站/DefaultPLC/Main_Plc/回水温度注意device只是 tagKey 的首段PLC 名tag保留了设备下的组/点层级所以主题里 PLC 和点位是分开的两级订阅方可以只订阅某个 PLC 整棵子树/DefaultPLC/#。// jsonpayloadformatter.cpp::topicForQStringJsonPayloadFormatter::topicFor(constQStringbaseTopic,constQStringchannel,constQStringstation,constQStringdevice,constQStringtag){constQString bodyIPayloadFormatter::resolveTemplate(m_topicTemplate,station,device,tag);returnQString(%1/%2/%3).arg(baseTopic,channel,body);}载荷统一走 JSONformatter 设计成可替换以后接阿里云物模型、AWS 只加一个子类主干不动// change单点变化{tag:DefaultPLC/Main_Plc/回水温度,value:52.3,quality:GOOD,ts:2026-09-28T09:12:00.123Z}// snapshot周期全量{snapshot:[{tag:...,value:52.3,quality:GOOD,ts:...}],ts:...}// alarm报警事件{tag:...,message:温度高报,level:警告,ts:...}ts用 UTC 的 ISO8601带毫秒跨时区对账不糊涂。四、完整发布链路把上面串起来一次值变化从采集到发出去走的是这条线// mqttpublisher.cpp::onTagValueChanged节选voidMqttPublisher::onTagValueChanged(constQStringtagKey,constQVariantvalue){if(!m_running||!m_cfg||!m_cfg-publishOnChange)return;if(!shouldPublish(tagKey,value,now))return;// ← 第2篇死区节流在这里拦constQByteArray payloadm_formatter-formatChange(s);// 格式化constQString topicm_formatter-topicFor(m_cfg-baseTopic,// 拼主题change,m_stationName,device,tag);publishFormatted(change,topic,payload,m_cfg-qos);// 发布}voidMqttPublisher::publishFormatted(constQStringchannel,constQStringtopic,constQByteArraypayload,intqos){if(sendRaw(topic,payload,qos)){m_totalPublished;}else{enqueue(channel,topic,payload,qos);// ← 第3篇连不上就进离线队列}}sendRaw走 Paho 发发成功计数发失败没连上就丢给enqueue进离线队列。两个关键关卡——过滤和可靠——我留在后面两篇讲因为那是坑最集中的地方。五、配置项一览对着代码填配置默认说明brokerHost/brokerPort127.0.0.1/1883Broker 地址8883 走 TLSusername/password空空即匿名cleanSessionfalse持久会话断线重连可续订keepAliveSec/connectTimeoutSec60/30心跳 / 连接超时reconnectBackoffSec5断线重连退避qos1默认 QoS1至少一次deadbandRel0.005相对死区 0.5%第2篇minIntervalMs1000最小发布间隔第2篇baseTopic/topicTemplateplcmonitor/%{station}/%{device}/%{tag}主题maxQueueMem/maxQueueDiskBytes2000/2MB离线队列上限第3篇还有个不起眼但很重要的一点MQTT 整体过授权模块门控。没买 mqtt 模块start()直接返回授权未包含 mqtt 模块一个字节都不发——对外接口不是常开的基础设施。小结与下篇本篇把PLC 数据怎么变成 MQTT 消息的主干跑通了三通道分流 → 主题模板拼路径 → JSON 格式化 → 异步发布。但有两件事这篇只是点到变化通道每秒可能上万次全发 Broker 吃不消、4G 流量烧钱 ——第 2 篇讲死区 最小间隔怎么把流量砍下来车间网络抖一下没连上时的数据不能丢 ——第 3 篇讲离线队列 QoS1 怎么做到断网不丢、连上补发。本文及 PLCMonitor 系列文章均为免费分享。本文免费分享如需转载请联系作者获取授权。如果觉得这篇文章对你有帮助欢迎点赞收藏。源码获取地址https://github.com/freddiezhang1990/plcmonitor