工业设备数据采集平台架构设计

[酷] 包含4大领域：数据采集与治理、数据架构、数据能力、数据应用；15大数据模块：既包含数据采集、数据治理、数据仓库等传统技术模块，又包含云原生、因果推断、预训练等新锐技术模块。

[给力] 如何获得这份《数据智能知识地图》？关注公众号“大话数智”，即可免费下载

导读：本次分享题目为映云科技 EMQ 的工业设备数据采集平台，主要分享数据采集的工具，总结工业设备采集的经验，以及介绍工业应用中的解决方案。

今天的介绍主要围绕“Neuron v2.0 功能及核心模块介绍”展开。全文目录：

• 工业设备数据采集经验
• 工业设备数据采集架构设计
• 案例分享

分享嘉宾 张奉前 映云科技 工业物联网产品总监

编辑整理 陈妃君 深圳大学

出品社区 DataFun

01

工业设备数据采集经验

1. Data Infrastructure for IoT 开源产品组合

EMQ 公司的产品基于全球领先的 MQTT 消息服务器和流处理数据库，实现从边缘端（Edge）到云端（Cloud）的一站式物联网数据「连接、移动、处理、存储与分析」服务。

2. Neuron 的应用情况及客户反馈

经过一年多 Neuron 1.x 版本的应用，我们分析了其应用情况和客户反馈消息，总结得下以下几个经验：

• 海量设备连接问题：设备连接及数据采集不是纯粹的一个 IT 项目或软件开发，更多需从自动化方面考虑问题。不是所有老旧设备都可通过改造去提供所需要的设备数据，需要设备和项目本身决定。
• 海量设备连接问题：除了产品功能需要提升，在部署 Neuron 也是一项挑战非常大的工作，客户工业环境复杂，设备多来自不同产商，要连接所有设备并不容易，一般情况下需要厂家介入。除非客户有自己工程师在维护设备。
• 强大规则引擎问题：除了多连接和定制开发需求，我们还注意到 Neuron v1.x 中的 SCADA 相关功能（如工业级警报系统、业务规则）并未得到用户很充分的使用。一般客户不会轻易替换已有的 SCADA 系统，因此如果能对客户 SCADA 系统对接，客户会更加乐意使用。
• 访问权限问题：客户对同时连接多个设备在一个实例的需求较多，目前以多实例（集群）方案加上 docker 部署去实现多连接需求。
• 高吞吐能力问题：北向除了发送消息去 IIoT 平台之外，也需要支持更多不同的应用软件，功能扩展能力也是一种增值服务。可让在基本连接上附加一些额外功能，就可让整个处理在边缘完成。
• 高可用集群问题：边缘业务逻辑处理能力的需求较大，虽然 Neuron 本身自带一个类似 C 脚本的业务逻辑引擎，但功能上它是比较接近用在 SCADA 要求，多数都在做数据分析，所以流式业务引擎更合适。
• 代理订阅问题：实例的内存占用也是重要的课题，虽然目前版本运行时并不大，但也有优化空间。特别在多连接时，跑多个相同实例时会重载入共用函数库或资源。

--

02

工业设备数据采集架构设计

通过以上 Neuron 1.x 版本的简介和应用案例分析，开始思考是否需要对 Neuron 的部分功能进行精简，使其更轻巧、更专注于数据采集功能。综合以上诸多考虑，我们决定从 v2.0 开始对 Neuron 的架构进行重新设计，同时将其开源，以满足更多用户的多样化需求。

1. Neuron 2.x IoT connectivity serve

Neuron 2.0 版本基于 1.x 版本，主要在以下几个方面进行改进：

• 正式放弃一些 SCADA 特有功能，包括警报系统、业务逻辑引擎及以 Websockets 方式推送数据到页面等。专注于 IIoT 设备连接。
• 支持在同一实例上同时连接相同或不同协议设备，加上设备或应用之间的讯息路由。
• 北向需要支持更多数据处理引用，包括 Edge X Foundry 支持，工业标准 Sparkplug B 支持及统一挂应用如 eKuiper 等。
• 开源+商业授权。

2. EMQ 产品矩阵

相较于 1.x 版本，缩减 EMQ X Edge 产品模块。

3. Neuron 2.0 核心基础功能设计

对于 Neuron 2.0 设计主要从以下四个方面进行：

• 核心消息路由，数据标签
• 南向协议插件扩展
• 北向应用插件扩展
• 可插拔、动态加载插件

4. Neuron 2.0 的总线模式的拓扑结构

采用该总线模式的拓扑结构，可以满足一下功能需求：

① NNG 基础消息库

• 多线程的 IO 处理和消息处理的异步并发库；
• 消息路由基于 NNG 提供的高效率消息转发；
• 消息的发送和接收是线程间通讯；
• 使用共享 Buffer，没有任何的内存拷贝，具有非常高的效率。

② 功能插件加载机制

• 所有插件都需要 Adapter 作为载具；
• 用户可以根据应用场景的功能需求动态的加载不同功能的插件；
• 加载不同功能的插件到 Adapter，就能得到有不同功能的节点；
• 每个节点的运行是独立的，当升级一个节点的插件时，不会影响其他节点的运行。

③ 各个具体功能的节点

• 这些节点可以是内置固有的，如轻量级的 Web Server；
• 可以是动态增加的，如各种设备驱动；
• 可以是动态增加的，MQTT 客户端、eKuiper 接口等。

5. NNG 基础消息库

消息传输主要有以上六种传输模式，Neuron 选取了 Pair 传输模式建立星型消息总线，这一组织形式中的消息路由中心是基于 NNG 基础消息库提供了高效率的消息转发。NNG 基础消息库是一个轻量级消息层，可解决重复出现的消息问题，无需 Broker，可具有很好的可扩展性，因此在很多产品中应用该库。

6. Neuron 的层次结构图

基于上诉总线模式、NNG 基础消息库等结构，Neuron 的层次结构大致可分为三层：顶层、中间层和底层。

• 顶层是第三方应用。第三方软件都独立运行，提供应用服务，例如流式引擎，MQTT 连接等。
• 中间层是各项功能插件。各项插件只要加载到适配器上，就可提供不同服务，例如 Web server、驱动、MQTT 等。
• 底层是核心框架构件。最底层就是就 NNG 基础消息传输库；各个节点的消息传送都要依赖这库完成，而且提供了基本线程管理、动态库管理、Web server 和 MQTT 函数功能；Neuron base lib 是 Neuron 自身基础库，包括通用函数、日志；最后是各类型适配器 Adapter，包括流式，MQTT 连接和驱动连接等。

7. Neuron Manager 的设计

Neuron Manager 是很重要的模块，管理着总线模块和各种 Adapter。其主要有四个模块：

• Datatag manager：是一个数据标签配置列表，维护这个节点所有数据标签配置。
• Plugin manager：用户可随时动态加载或卸载动态.so文件并登记在插件表成为备选插件。
• Adapter manager：用户可根据插件表格（Plugin table）内说明的插件功能及提供服务去选取所需插件。
• Msg Router manager：根据数据标签配置，把收到数据消息路由到最终节点。

8. 数据标签管理及 Data Tag 设计

DataTag 数据标签是指数据在设备内的位置及属性，属性包括数据名字、类型、起始位置、占用大小及读写功能等。其主要工作流程如下：

• tagGroup 把相同读取频率，例 100ms 或 200ms，打包成一组，所以每一个服务节点都有很多数据标签组，每一组只有一个读取频率。
• 北向应用可以选择订阅那些用户需要的数据标签组，订阅后 pipe 通适登记在 subPipes。
• 插件会按照 tagGroup 的读取频率去读取设备的数据。读取数据后会发到 manager，manager 就按照这 adapter table，把数据消息路由最终站数据节点。

9. Neuron 的 Pipe/Adapter 和 Plugin 设计

在 Plugin 中分开两个主要功能：callbacks 和 interface。主要通过以下逻辑实现：

• 服务节点会按照 tagGroup 所设定读取频率去发出设备数据读取命令并执行插件功能(*request)，然后收到数据回覆后执行 callback 函数功能(*response)，把数据发送到有订阅这些数据节点。
• 如果要新增了一个服务节点，就须要执行插件(*open)跟着(*init)，最后是执行配置(*config)。
• 在删取一个服务节点之前，就须要执行插件(*uninit)然后(*close)。

10. Message 数据转换的设计

由于有不同的服务节点，提供应用或驱动服务，但每个节点能处理的数据类型都不同。可以看到不同的服务节点所处理数据类型差异很大，所以必须要有一个内在统一数据类型作为中转之用。在 Modbus 协议规格有定义能支持数据类型，就可以在服务节点内把 Modbus 数据类型转换成 Neuron Unified Data Type。一般北向应用例如 MQTT 或 eKuiper 都会以 JSON 格式去传送消息，而 JSON 的数据类型只有 boolean, string, number 等。Neuron Unified Data Type 最后转换成 JSON 数据类型发送消息。

11. Neuron 的 Plugin 的两种 作模式

Plugin 和 Adapter 之间有两种连接方法。一种是 Plugin 直接和 PLC 进行连线，多在南向中应用；另外一种是通过第三方软件连接在一起，多在北向中应用。

每个适配器 Adapter 都需要独立线程去执行，加载插件也需要另外线程去执行。所以每个服务节点都有两条执行线程。用户将所需要应用或功能插件，加载到适配器就可成为服务节点提供相应服务。一般插件都是以 Linux 动态库(.so 文件)型式加载到适配器。而且这些插件都支持热插方式，即是在运行时可随时加载或卸载。方便灵活。用户可以根据现场的工作情况动态增加和减少设备驱动的节点，具有良好的可配置性。所以在一般情况下，用户只需要将所需服务在插件里实现就可以，每个插件实现一种服务，这样设计隔离了各个节点之间功能的耦合。

如果所需要的应用是由第三方软件提供，这时候需要第三方代理插件作为第三方软件的桥梁。这个桥梁需要按照第三方通讯接口 API 实现数据交换，同时把挢梁实现在插件里加载到适配器，就可跟第三方软件连接。

12. Neuron 集成 eKuiper 协同 作

eKuiper 是由 EMQ 发起并托管在 LF Edge 基金会（Edge X Foundry），对无限的数据流进行状态分析的软件或者框架。eKuiper 是独立第三方软件，因此 Neuron 和 eKuiper 集成方法可参考代理插件模式。集成 eKuiper 有以下特点：

• 使用 SQL 实现数据处理，内置函数快速处理 IoT，结构与非结构化数据处理。
• Neuron 将会默认付带有 eKuiper 流式处理引擎。
• Neuron 和 eKuiper 集成后，将会是一个功能强大而且非常轻巧的边缘数据流处理器，可对不同工业设备数据源直接进行处理。
• Neuron 的数据流和数据库处理和计算基本上已在本地边缘侧完成并且存入回本地数据库。
• Neuron 和 eKuiper 之间通讯采用 NNG pair0 作传输方式。数据结构采用 JSON 格式交换。

13. Neuron 的 data buffer 和 data 的元数据关系

这上面是 tagGroup 所需的资料，包括订阅管理、数据标签阵列，主要保存了数据名字、类型、起始位置、占用大小及读写功能等。下面是数据 buffer 的结构，数据是按照 tagGroup 次序排列，每次发送数据消息，都是依据 tagGroup 的排列次序。

14. Neuron 的 Plugin 的两种工作模式

① 客户端发出订阅请求，有 MQTT broker 转发。

② 收到转发数据订阅请求后，将订阅请求的 tag 组注册到设备对应的 configs 列表中，得到这个 tag 组 config 的 id，通过 response 接口通知 adapter 有关请求。

③ 将这些订阅请求 tag 组的 config 的 id 发送到 manager，manager 将这些请求交给 datatag manager 管理，datatag manager 会给这些订阅请求 tag 组加上对应的订阅者的 adapter id 对应的 pipe。

④ Manager 会发送一个订阅请求（包括 tag 组的 config 的 id）给设备关联的adapter，adapter 将这个请求通过 request 接口告诉 driver plugin。

⑤ Driver plugin 由 tag 组的 config 的 id 获取所有要读取的数据 tag，根据数据tag 读取的时间间隔 readInterval 建立一个间隔时间由少到多排列的数据 tag 的 id 队列，根据时间间隔来设定定时器。

⑥ 当定时时间到时，从队列头部取出请求的数据 tag 的 id，通过该 id 从 datatag table 中获取数据 tag 具体的配置信息，从设备读取配置信息制定要读取的数据，放进 neu_variable_t 中。

⑦ Driver plugin 将读取的数据的 neu_variable_t 通过 response 接口返回给adapter。

⑧ Adapter 将其转换为 databuf 放到 data message 中传给 manager。

⑨ Manager 调用 datatag manager 根据 databuf 中包含的 tag 组的 config 的 id 快速匹配到订阅者的 adapter，将 data message 发给所有订阅者的 adapter。

⑩ 订阅者的 adapter 再通过 requests 接口将数据发送给订阅者的 plugin，订阅者的 plugin 将请求的数据发送给订阅者应用。

⑪ 把数据发送回客户端。

15. Neuron 的数据流的汇聚/分散（gather/scatter）

我们主要是通过以上两个方式实现对数据流的汇聚和分散的。

16. Plugin 模式的两种实现方法

用户可以自行编写的 plugin 有两类，一类是半堆叠插件，另一类是全堆叠插件。

• 半堆叠插件只需编写应用层协议，数据规格和提供设备地址的定义。至于底层协议则调用 neuron 原生通讯库。大部分工业协议可用此方法。
• 全堆叠插件编写则用户需自行加上编写底层通讯库。有非常少部分插件，例如 BACnet，原生协议定义已括底层。又或者，用户想重新编写整个协议。

--

03

案例分享

1. 案例分享1——某油田生产数据采集方案

为将 Neuron 应用到油田服务中，发挥设备数据采集的作用，我们制定以下方案：

• 在场站端，通过工业数据采集网关将现场采用 Modbus-RTU、Modbus-TCP 协议的仪表数据转换成高可靠轻量化的 MQTT 协议，由边缘数据采集网关将实时数据通过专网将 MQTT 协议推送到厂区中心机房的数据采集汇聚平台。
• 部分冗余数据或者其他系统数据，可以通过部署在场站端的轻量化数据处理软件将数据进行过滤处理，将有意义的数据推送到厂区中心机房的数据采集汇聚平台。
• 厂区中心机房的数据采集汇聚平台将实时数据统一接入后存储到中心机房的数据库中，各种业务应用可以对接数据库拉取相关业务数据。
• 生产设备警报、场站内禁警报等需要实时处理的数据可以由数据采集汇聚平台通过 MQTT 协议推送到报警处理业务系统中，实现实时数据快速处理。
• 厂区数据中心平台也可以通过 MQTT 消息实现对现场设备的远程控制 和管理。

随着用户与应用场景的不断增加，我们逐渐发现最初多协议、单连接的设计开始无法适应多样的场景需求。以油田服务数据采集为例，要将所有油井数据汇聚到同一服务器上，需要 Neuron 同时提供多个连接到不同地区的油井。这是流程型工业生产，和离散型工业利用 SCADA 批量生产的情况有很大不同。

2. 案例分享2——智慧工厂设备数据采集及产品质量溯源

另一流程型生产的案例是半导体芯片的生产控制，面对生产环境的复杂性和对数据采集的高要求，主要需要解决以下几个痛点：

• 各类厂家、各类型号的 PLC 数据的实时采集能力。以完成现场 200 多套各个厂家的 PLC 设备共同完成生产控制任务。
• 多工序间配合控制功能。以实现工厂通过上层应用软件实现对 PLC 的指令下发，实现现场设备的上下料、工序间配合等控制工功能。
• 缺乏边缘端数据清洗及分析能力。对于生产工艺参数、生产过程数据需要高频采集存储，为产品质量溯源提供数据支撑，针对缺陷产品对客户提供质量分析报告。

针对以上痛点，EMQ 提出了以下解决方案：

• 采用 Neuron，可实现对现场各类 PLC 设备的数据实时采集，其速度>=100ms，上层应用直接调用 Neuron API 完成指令下发，实现工序间配合控制功能。
• 现场侧部署 eKuiper 边缘流式分析引擎，对 Neuron 采集的实时数据流根据规则设定进行数据过滤和处理，并将结果推送到 EMQ X Enterprise。通过 eKuiper 对数据的过滤，降低后端数据存储的压力，加强产品质量追溯能力。

EMQ X Enterprise 将多段 Neuron、eKuiper 采集的数据进行统一汇聚，数据推送到数据库及大数据系统做持久化存储。

今天的分享就到这里，谢谢大家。

分享嘉宾

张奉前

杭州映云科技 工业物联网产品总监

从事15年工业自动化及编写SCADA经验，曾参与国内外大型自动化项目，包括汽车零件电镀生产线、太阳能硅片生产线等等。

DataFun新媒体矩阵

关于DataFun

专注于大数据、人工智能技术应用的分享与交流。发起于2017年，在北京、上海、深圳、杭州等城市举办超过100+线下和100+线上沙龙、论坛及峰会，已邀请超过2000位专家和学者参与分享。其公众号 DataFunTalk 累计生产原创文章800+，百万+阅读，15万+精准粉丝。