集成 EMQX

本节将演示如何通过 Neuron 南向采集设备实际点位数据，通过北向 Sparkplug B 插件将数据上报到 EMQX， 再通过EMQX编解码功能解码后得到正确完整的数据结果，流程如图：

Sparkplug B

配置 Neuron

添加南向设备

通过南向驱动采集 Modbus TCP 模拟器点位值去模拟实际设备点位值，配置如下：

添加设备

image-20230419133807028

设备配置

image-20230419134424414

image-20230419134500446

创建组

image-20230419134630777

添加点位

image-20230419134741736

添加北向应用

添加应用

image-20230419134941116

应用配置

image-20230419135022571

image-20230419135247025

添加订阅

image-20230419135416517

配置 EMQX

我们可以通过安装 MQTTX 桌面客户端查看由 Neuron 转发到 EMQX 的数据。但如果直接订阅 Neuron 北向 SparkPlugB 上报到 EMQX 的数据，则会出现字符串乱码的情况，如图：

image-20230419140026304

所以可以通过EMQX规则引擎编解码的能力，编写一个相应的proto文件结合规则引擎将上报的数据进行解码后得到正确完整的数据结果。

创建编解码

image-20230419140540482

image-20230419140724126

bash

  1. ## 完整proto文件
  2. syntax = "proto2";
  4. //
  5. // To compile:
  6. // cd client_libraries/java
  7. // protoc --proto_path=../../ --java_out=src/main/java ../../sparkplug_b.proto 
  8. //
  10. message Payload {
  11.     /*
  12.         // Indexes of Data Types
  13.         // Unknown placeholder for future expansion.
  14.         Unknown         = 0;
  15.         // Basic Types
  16.         Int8            = 1;
  17.         Int16           = 2;
  18.         Int32           = 3;
  19.         Int64           = 4;
  20.         UInt8           = 5;
  21.         UInt16          = 6;
  22.         UInt32          = 7;
  23.         UInt64          = 8;
  24.         Float           = 9;
  25.         Double          = 10;
  26.         Boolean         = 11;
  27.         String          = 12;
  28.         DateTime        = 13;
  29.         Text            = 14;
  30.         // Additional Metric Types
  31.         UUID            = 15;
  32.         DataSet         = 16;
  33.         Bytes           = 17;
  34.         File            = 18;
  35.         Template        = 19;
  37.         // Additional PropertyValue Types
  38.         PropertySet     = 20;
  39.         PropertySetList = 21;
  40.     */
  42.     message Template {
  44.         message Parameter {
  45.             optional string name        = 1;
  46.             optional uint32 type        = 2;
  48.             oneof value {
  49.                 uint32 int_value        = 3;
  50.                 uint64 long_value       = 4;
  51.                 float  float_value      = 5;
  52.                 double double_value     = 6;
  53.                 bool   boolean_value    = 7;
  54.                 string string_value     = 8;
  55.                 ParameterValueExtension extension_value = 9;
  56.             }
  58.             message ParameterValueExtension {
  59.                 extensions              1 to max;
  60.             }
  61.         }
  63.         optional string version         = 1;          // The version of the Template to prevent mismatches
  64.         repeated Metric metrics         = 2;          // Each metric is the name of the metric and the datatype of the member but does not contain a value
  65.         repeated Parameter parameters   = 3;
  66.         optional string template_ref    = 4;          // Reference to a template if this is extending a Template or an instance - must exist if an instance
  67.         optional bool is_definition     = 5;
  68.         extensions                      6 to max;
  69.     }
  71.     message DataSet {
  73.         message DataSetValue {
  75.             oneof value {
  76.                 uint32 int_value                        = 1;
  77.                 uint64 long_value                       = 2;
  78.                 float  float_value                      = 3;
  79.                 double double_value                     = 4;
  80.                 bool   boolean_value                    = 5;
  81.                 string string_value                     = 6;
  82.                 DataSetValueExtension extension_value   = 7;
  83.             }
  85.             message DataSetValueExtension {
  86.                 extensions  1 to max;
  87.             }
  88.         }
  90.         message Row {
  91.             repeated DataSetValue elements  = 1;
  92.             extensions                      2 to max;   // For third party extensions
  93.         }
  95.         optional uint64   num_of_columns    = 1;
  96.         repeated string   columns           = 2;
  97.         repeated uint32   types             = 3;
  98.         repeated Row      rows              = 4;
  99.         extensions                          5 to max;   // For third party extensions
  100.     }
  102.     message PropertyValue {
  104.         optional uint32     type                    = 1;
  105.         optional bool       is_null                 = 2; 
  107.         oneof value {
  108.             uint32          int_value               = 3;
  109.             uint64          long_value              = 4;
  110.             float           float_value             = 5;
  111.             double          double_value            = 6;
  112.             bool            boolean_value           = 7;
  113.             string          string_value            = 8;
  114.             PropertySet     propertyset_value       = 9;
  115.             PropertySetList propertysets_value      = 10;      // List of Property Values
  116.             PropertyValueExtension extension_value  = 11;
  117.         }
  119.         message PropertyValueExtension {
  120.             extensions                             1 to max;
  121.         }
  122.     }
  124.     message PropertySet {
  125.         repeated string        keys     = 1;         // Names of the properties
  126.         repeated PropertyValue values   = 2;
  127.         extensions                      3 to max;
  128.     }
  130.     message PropertySetList {
  131.         repeated PropertySet propertyset = 1;
  132.         extensions                       2 to max;
  133.     }
  135.     message MetaData {
  136.         // Bytes specific metadata
  137.         optional bool   is_multi_part   = 1;
  139.         // General metadata
  140.         optional string content_type    = 2;        // Content/Media type
  141.         optional uint64 size            = 3;        // File size, String size, Multi-part size, etc
  142.         optional uint64 seq             = 4;        // Sequence number for multi-part messages
  144.         // File metadata
  145.         optional string file_name       = 5;        // File name
  146.         optional string file_type       = 6;        // File type (i.e. xml, json, txt, cpp, etc)
  147.         optional string md5             = 7;        // md5 of data
  149.         // Catchalls and future expansion
  150.         optional string description     = 8;        // Could be anything such as json or xml of custom properties
  151.         extensions                      9 to max;
  152.     }
  154.     message Metric {
  156.         optional string   name          = 1;        // Metric name - should only be included on birth
  157.         optional uint64   alias         = 2;        // Metric alias - tied to name on birth and included in all later DATA messages
  158.         optional uint64   timestamp     = 3;        // Timestamp associated with data acquisition time
  159.         optional uint32   datatype      = 4;        // DataType of the metric/tag value
  160.         optional bool     is_historical = 5;        // If this is historical data and should not update real time tag
  161.         optional bool     is_transient  = 6;        // Tells consuming clients such as MQTT Engine to not store this as a tag
  162.         optional bool     is_null       = 7;        // If this is null - explicitly say so rather than using -1, false, etc for some datatypes.
  163.         optional MetaData metadata      = 8;        // Metadata for the payload
  164.         optional PropertySet properties = 9;
  166.         oneof value {
  167.             uint32   int_value                      = 10;
  168.             uint64   long_value                     = 11;
  169.             float    float_value                    = 12;
  170.             double   double_value                   = 13;
  171.             bool     boolean_value                  = 14;
  172.             string   string_value                   = 15;
  173.             bytes    bytes_value                    = 16;       // Bytes, File
  174.             DataSet  dataset_value                  = 17;
  175.             Template template_value                 = 18;
  176.             MetricValueExtension extension_value    = 19;
  177.         }
  179.         message MetricValueExtension {
  180.             extensions  1 to max;
  181.         }
  182.     }
  184.     optional uint64   timestamp     = 1;        // Timestamp at message sending time
  185.     repeated Metric   metrics       = 2;        // Repeated forever - no limit in Google Protobufs
  186.     optional uint64   seq           = 3;        // Sequence number
  187.     optional string   uuid          = 4;        // UUID to track message type in terms of schema definitions
  188.     optional bytes    body          = 5;        // To optionally bypass the whole definition above
  189.     extensions                      6 to max;   // For third party extensions
  190. }

创建规则

SQL 语句

bash

  1. SELECT
  2.   schema_decode('neuron', payload, 'Payload') as SparkPlugB
  3. FROM
  4.   "spBv1.0/group1/DDATA/node1/modbus"

这里的关键点在于 schema_decode('neuron', payload, 'Payload')：

  • schema_decode 函数将 payload 字段的内容按照 protobuf_person Schema 来做解码;
  • as SparkPlugB 将解码后的值保存到变量 “SparkPlugB” 里;
  • 最后一个参数 Payload 指明了 payload 中的消息的类型是 protobuf schema 里定义的 ‘Payload’ 类型。

image-20230419141423179

然后使用以下参数添加动作：

  • 动作类型：消息重新发布
  • 目的主题：SparkPlugB/test

这个动作将解码之后的 “Payload” 以 JSON 的格式发送到 SparkPlugB/test 这个主题。

image-20230419141513303

通过 MQTTX 验证

再次通过 MQTTX 工具去订阅相关主题，经过 EMQX 规则引擎的编解码后，我们可以在 MQTTX 客户端看到正确解码后的数据，如图：

image-20230419141831882

如上图，可以看到解码前的原数据是乱码的，解码后得到完整正确的数据结果；至此，通过Neuron南向采集设备点位值，北向 SparkPlugB 上报到 EMQX，通过编解码功能解码得到完整的数据结果已完成。

拓展阅读：Sparkplug 协议

Neuron 上报数据到 EMQX 的 Topic 是根据 Sparkplug B 协议规范定义的，格式如下：

namespace/group_id/DDATA/edge_node_id/device_id，

协议内容如图：

image-20230419143059088

至于更多 Neuron 北向 Sparkplug B 插件相关标准的定义，可以参考 SparkPlug B 协议规范。