Insights Hub MindConnect MQTT – 本地私有云使用
摘要
本文主要介绍了如何在 Insights Hub 本地私有云环境（LPC）中使用 MindConnect MQTT 功能。 根据本文的操作步骤读者可以通过 MindConnect MQTT 上传 Time Series 数据到 Insights Hub 中。后续相关步骤可参考官网介绍。
详细信息
一、准备
- 一个 Developer 类型的 Insights Hub tenant。
- "MQTT.fx" 客户端可视化工具（本文使用 "v1.7.1" 这个版本）；"mosquitto" MQTT 命令行工具。
- OpenSSL 工具，参考官网。本文使用别人已经制作好的基于 Windows 的安装包。
- Postman 工具。
二、证书相关操作
2.1 生成 Tenant CA 证书并上传
2.1.1 生成证书
在这一步中我们需要在 Insights Hub 注册我们自己生成的 CA 证书。
1. 登录 Insights Hub，进入 Asset Manager
2. 然后左侧点击 MQTT Certificates
3. 针对 LPC 环境，使用者需要自行签发并上传 tenant 的 CA 证书
在填写了 CA 证书名字后，按照下面步骤生成证书
复制命令并填写相应内容（下面是准备好的命令，基于 Windows）
set TENANT=suppdev set COUNTRY_CODE=CN set CITY=Chengdu set ORGANIZATION=CTS set ORGANIZATIONAL_UNIT=CTS openssl genrsa -out %TENANT%.key 2048 openssl req -x509 -new -nodes -key %TENANT%.key -sha256 -days 3650 -out %TENANT%.pem -subj "/C=%COUNTRY_CODE%/ST=%CITY%/O=%ORGANIZATION%/OU=%ORGANIZATIONAL_UNIT%/CN=%TENANT%"
填写完过后将命令复制到命令行窗口并执行
2.1.2 上传 CA 证书
1. Tenant CA 生成过后，找到以 "tenant" 名字开头的证书，然后上传到 Insights Hub 中
2.2 生成 Verification 证书并上传
1. 上传过后，可以看到后续的 "Registration Code" 生成；同时，点击“问号”标志继续生成 Verification 证书
复制下面命令，填写对应参数
（下面是准备好的命令，基于 Windows）
set TENANT=suppdev set REGISTRATION_CODE=9041961a00224c0ea093077c3d6ac0f5 openssl genrsa -out verificationCert.key 2048 openssl req -new -key verificationCert.key -out verificationCert.csr -subj /CN=%REGISTRATION_CODE% openssl x509 -req -in verificationCert.csr -CA %TENANT%.pem -CAkey %TENANT%.key -CAcreateserial -out verificationCert.pem -days 7 -sha256
2. Verification 证书生成后，上传下面 "verificationCert.pem" 证书
3. 上传完成后，页面回自动跳回上一页面
至此，Tenant CA 证书生成步骤结束。
2.3 生成 Client 证书
下面已填好对应的信息
set TENANT=suppdev set DEVICE_NAME=kevin_mqtt_device_1 set COUNTRY_CODE=CN set CITY=Chengdu set ORGANIZATION=CTS openssl genrsa -out %DEVICE_NAME%.key 2048 openssl req -new -key %DEVICE_NAME%.key -out %DEVICE_NAME%.csr -subj "/C=%COUNTRY_CODE%/ST=%CITY%/O=%ORGANIZATION%/OU=IT/CN=%DEVICE_NAME%" openssl x509 -req -in %DEVICE_NAME%.csr -CA "%TENANT%.pem" -CAkey "%TENANT%.key" -CAcreateserial -out %DEVICE_NAME%.pem -days 365 -sha256 type %DEVICE_NAME%.pem "%TENANT%.pem" > "%DEVICE_NAME%"_chain.pem
执行命令，可看到后续生成的文件
三、为客户端（Client）生成 JWT
Client 连接 MC MQTT Broker 时需要提供 JWT。此节会介绍生成 JWT 的步骤。
打开 JWT.IO，看到如下界面：
3.1 HEADER
- alg。默认为：RS256。
- x5c。客户端证书与 Tenant CA 证书的组合。需要满足下面要求：
- 2 个证书必须要保持顺序，客户端证书在前，Tenant CA 证书在后。
- 所有证书必须要处于有效状态。
- 证书内容每一行默认的“换行符”必须去掉，需保证数组里的元素为“字符串”。
- typ。默认为：JWT。
{
"alg": "RS256",
"x5c": [
"MIIDNjCCAh4CFDvz+vdRGEfz3wGKa0+JdkOnWefHMA0GCSqGSIb3DQ...",
"MIIDezCCAmOgAwIBAgIUMoRljy56YvnC9bJQs9aFS3PVx8kwDQYJKoZIhvcNAQ..."
],
"typ": "JWT"
}
3.2 PAYLOAD
- jti。唯一标识，格式为 UUID，最大长度 36 个字符。
- iss。Token 签发信息。值为 "MQTT 的 Client ID"。
- sub。Subject 信息。值为 "MQTT 的 Client ID"。
- aud。Audience claim。值为 ["MQTTBroker"]。
- iat。Token 签发时间。时间戳格式，精度为“秒”。
- exp。Token 过期时间。时间戳格式，精度为“秒”。Note: 令牌必须在一小时内有效。因此，iat 和 exp 之间的差值必须最多为一小时。
- schemas。令牌自定义语义声明。值为 ["urn:siemens:mindsphere:v1"]。
- ten。Tenant 名字。
格式参考如下：
{
"jti": "d9f24ddd-447e-446c-aeaf-e4a976635e8c",
"iss": "suppdev_kevin_mqtt_device_1",
"sub": "suppdev_kevin_mqtt_device_1",
"aud": ["MQTTBroker"],
"iat": 1692074802,
"exp": 1692078402,
"schemas": ["urn:siemens:mindsphere:v1"],
"ten": "suppdev"
}
Note: Client ID 的格式为：{Tenant name}_{Device name}。且长度需要大于 16 个字符。
3.3 VERIFY SIGNATURE
这里需要填入 Client 的私钥
3.4 获取 JWT
将准备好的 HEADER、PAYLOAD、Client 私钥填入到 JWT.IO 网站的对应位置，然后左侧获取生成的 JWT。
四、通过 MQTT.fx 连接 MC MQTT Broker
打开 MQTT.fx 软件，新建一个连接
General
填入对应的信息到 General 这一页：
- Broker Address：地址为以 mindconnectmqtt. 开头，其余部分 URL 根据当前环境地址配置。
- Broker Port：使用 TLS 连接 Broker 的 Port。默认为 8883，根据当前实际所用端口配置。
- Client ID 的格式为：{Tenant name}_{Device name}。
User Credentials
- User Name。默认为 _CertificateBearer。
- Password。上述获取的 JWT。
SSL/TLS
这里选择 Message Broker 服务所用到的证书，需客户自行提供。
上述信息填写好后，保存并连接。成功连接后可看到右侧“绿色”标志
五、模型（Model）创建与初始化
5.1 模型（模板）结构定义
定义模型模板可以快速构建用于 MC MQTT 的 Asset 的结构。其中模型包含 Asset、Aspect、Asset 实例以及映射关系。模型模板定义过后，后期可以根据在实例化期间具体定义的值来进行初始化模型实例。
参考下例模型定义。其中：
- 以 <> 括起来的值需要替换为实际值。
- 以 ${} 括起来的值保持不变，这里作为一个变量供后续实例化时使用。
- 参数 externalId 需要特别注意，后续实例化模型时需要用到这个，请确保“值唯一”。这里使用的是 "SpaceShip"。
{
"id": "<requestId>",
"data": {
"externalId": "SpaceShip",
"typeModel": {
"aspectTypes": [
{
"id": "<tenantId>.wing",
"name": "${aspectTypeName}",
"category": "static",
"scope": "private",
"variables": [
{
"name": "temperature",
"dataType": "STRING",
"unit": "C/F",
"searchable": true,
"length": 5,
"qualityCode": true
}
],
"description": "wing aspect type description",
"referenceId": "287adc1a086840e0a6721dfd1170e97c"
}
],
"assetTypes": [
{
"id": "<tenantId>.spaceship",
"name": "wingAssetTypeName",
"parentTypeId": "core.basicasset",
"aspects": [
{
"name": "wingAspect",
"aspectTypeId": "<tenantId>.wing"
}
],
"description": "Hyperspace jump capable space ship",
"instantiable": true,
"scope": "private",
"referenceId": "82a4cc2a69cc42af80c1c6cf5dbefde5"
}
]
},
"instanceModel": {
"assets": [
{
"referenceId": "wingAssetReference",
"parentReferenceId": "root",
"typeId": "<tenantId>.spaceship",
"name": "wingAsset",
"description": "The ship of Han Solo and Chewbacca"
}
]
},
"mappingModel": {
"mappings": [
{
"dataPointId": "dp01",
"assetReferenceId": "wingAssetReference",
"aspectName": "wingAspect",
"variableName": "temperature",
"referenceId": "19e9048e78f540e7a9ba25e1249fea9b"
}
]
}
}
}
所以这里我们需要替换 <requestId>（请确保值“唯一”）、<tenantId>。修改过后的结构如下：
{
"id": "suppdev-request-20230816001",
"data": {
"externalId": "SpaceShip",
"typeModel": {
"aspectTypes": [
{
"id": "suppdev.wing",
"name": "${aspectTypeName}",
"category": "static",
"scope": "private",
"variables": [
{
"name": "temperature",
"dataType": "STRING",
"unit": "C/F",
"searchable": true,
"length": 5,
"qualityCode": true
}
],
"description": "wing aspect type description",
"referenceId": "287adc1a086840e0a6721dfd1170e97c"
}
],
"assetTypes": [
{
"id": "suppdev.spaceship",
"name": "wingAssetTypeName",
"parentTypeId": "core.basicasset",
"aspects": [
{
"name": "wingAspect",
"aspectTypeId": "suppdev.wing"
}
],
"description": "Hyperspace jump capable space ship",
"instantiable": true,
"scope": "private",
"referenceId": "82a4cc2a69cc42af80c1c6cf5dbefde5"
}
]
},
"instanceModel": {
"assets": [
{
"referenceId": "wingAssetReference",
"parentReferenceId": "root",
"typeId": "suppdev.spaceship",
"name": "wingAsset",
"description": "The ship of Han Solo and Chewbacca"
}
]
},
"mappingModel": {
"mappings": [
{
"dataPointId": "dp01",
"assetReferenceId": "wingAssetReference",
"aspectName": "wingAspect",
"variableName": "temperature",
"referenceId": "19e9048e78f540e7a9ba25e1249fea9b"
}
]
}
}
}
5.2 模型创建
定义好模型结构后，向 Broker 发送消息创建模型。
1. 向 Topic - tc/<tenantId>/<clientId>/o/amo_v3/m 发送创建模型消息。
打开 MQTT.fx，将准备好的模型结构消息体贴入然后点击 "Publish"。（注意：<tenantId> 和 <clientId> 要替换为实际值）。
2. 获取模型创建的状态。
同时，在向创建模型的 Topic 发送消息前可以先订阅 tc/<tenantId>/<clientId>/i/amo_v3/ms 或者 tc/<tenantId>/<clientId>/i/amo_v3/#，这样在创建过后可以收到模型创建结果的消息从而判断模型创建状态。
（由于之前已经发送过一次创建模型的消息，所以后续再以相同的 <requestId> 创建模型时，结果查询时会显示模型已存在）在 MQTT.fx 中切换到 "Subscribe" 标签页，然后订阅 Topic。
接着再次发送模型创建消息，可以收到如下返回消息
至此，模型创建完成。
5.3 实例化模型
1. 向 Topic - tc/<tenantId>/<clientId>/o/amo_v3/i 发送消息实例化上一步创建的模型
准备好消息体：
- <requestId> 上述创建模型用到的值。
- modelExternalId 上述创建模型用到的对应值。
- parameterization 为上述创建模型时以 ${} 括起来的参数。如果之前定义了多个参数，这里需以数组结构填入。
{
"id": "<requestId>",
"data": {
"modelExternalId": "SpaceShip",
"parameterization": {
"values": [
{
"name": "aspectTypeName",
"value": "wingAspectTypeName"
}
]
}
}
}
修改过后的消息体如下：
{
"id": "suppdev-request-20230816002",
"data": {
"modelExternalId": "SpaceShip",
"parameterization": {
"values": [
{
"name": "aspectTypeName",
"value": "wingAspectTypeName"
}
]
}
}
}
发送消息（在发送前，可以先订阅“获取实例化状态”的 Topics，这样发送实例化消息后可以接收到结果通知。参考下面订阅 topic）
2. 获取实例化状态
同时可以先在初始化模型时订阅 tc/<tenantId>/<clientId>/i/amo_v3/ip 或 tc/<tenantId>/<clientId>/i/amo_v3/#
由于已先订阅了此 Topic，当发送了实例化消息后可以收到多个通知，每个通知标识着不同的状态。
3. Asset Manager 中查看实例化过后的 Asset
当实例化完成后，回到 Asset Manager 中可查看到我们已经实例化过后的 Asset Agent 以及 Asset 信息
至此，模型实例化完成。
六、通过 MC MQTT 发送数据
本节将会介绍通过 MC MQTT 发送 Time Series 数据、Event、File 到 Insights Hub。（案例基于官网教程，请参考）
6.1 创建、初始化新的 Asset 模型
6.1.1 创建模型
模型定义消息体中的参数已在“第五章节”中介绍，下文模型定义的消息体将不再细述参数细节。
{
"id": "request-suppdev-20230817001",
"data": {
"externalId": "ElectricMechanicalDeviceModel",
"typeModel": {
"aspectTypes": [
{
"id": "suppdev.mechanicalAspectType",
"name": "mechanicalAspectType",
"category": "dynamic",
"scope": "private",
"variables": [
{
"name": "readTime",
"dataType": "TIMESTAMP",
"unit": "second",
"searchable": false,
"qualityCode": true
},
{
"name": "displacement",
"dataType": "INT",
"unit": "cm",
"searchable": true,
"qualityCode": true
},
{
"name": "rpm",
"dataType": "LONG",
"searchable": true,
"qualityCode": true
}
],
"description": "Mechanical properties of an device"
},
{
"id": "suppdev.electricalAspectType",
"name": "electricalAspectType",
"category": "dynamic",
"scope": "private",
"variables": [
{
"name": "overloaded",
"dataType": "BOOLEAN",
"searchable": true,
"qualityCode": true
},
{
"name": "current",
"dataType": "DOUBLE",
"unit": "A",
"searchable": true,
"qualityCode": true
},
{
"name": "errorMessage",
"dataType": "STRING",
"length": 255,
"searchable": true,
"qualityCode": true
}
],
"description": "Electrical properties of an device"
}
],
"assetTypes": [
{
"id": "suppdev.electricalAndMechanicalAssetType",
"name": "electricalAndMechanicalAssetType",
"parentTypeId": "core.basicasset",
"aspects": [
{
"name": "electricalAspect",
"aspectTypeId": "suppdev.electricalAspectType"
},
{
"name": "mechanicalAspect",
"aspectTypeId": "suppdev.mechanicalAspectType"
}
],
"description": "Asset Type with electrical and mechanical properties",
"instantiable": true,
"scope": "private"
}
]
},
"instanceModel": {
"assets": [
{
"referenceId": "electricalAndMechanicalAssetReference1",
"parentReferenceId": "root",
"typeId": "suppdev.electricalAndMechanicalAssetType",
"name": "electricalAndMechanicalAsset",
"description": "Asset with electrical and mechanical properties"
}
]
},
"mappingModel": {
"mappings": [
{
"dataPointId": "dataPoint1",
"assetReferenceId": "electricalAndMechanicalAssetReference1",
"aspectName": "electricalAspect",
"variableName": "current"
},
{
"dataPointId": "dataPoint2",
"assetReferenceId": "electricalAndMechanicalAssetReference1",
"aspectName": "mechanicalAspect",
"variableName": "displacement"
},
{
"dataPointId": "dataPoint3",
"assetReferenceId": "electricalAndMechanicalAssetReference1",
"aspectName": "mechanicalAspect",
"variableName": "rpm"
},
{
"dataPointId": "dataPoint4",
"assetReferenceId": "electricalAndMechanicalAssetReference1",
"aspectName": "electricalAspect",
"variableName": "errorMessage"
},
{
"dataPointId": "dataPoint5",
"assetReferenceId": "electricalAndMechanicalAssetReference1",
"aspectName": "electricalAspect",
"variableName": "overloaded"
},
{
"dataPointId": "dataPoint6",
"assetReferenceId": "electricalAndMechanicalAssetReference1",
"aspectName": "mechanicalAspect",
"variableName": "readTime"
}
]
}
}
}
消息体定义完成后，进入 MQTT.fx 向 Topic - tc/<tenantId>/<clientId>/o/amo_v3/m 发送创建模型消息。
6.1.2 初始化模型
初始化模型消息体参考如下代码：
{
"id": "request-suppdev-20230817002",
"data": {
"modelExternalId": "ElectricMechanicalDeviceModel",
"parameterization": {
"values": []
}
}
}
向 Topic - tc/<tenantId>/<clientId>/o/amo_v3/i 发送消息实例化模型。
初始化完成后，可以在 Asset Manager 中看到新生成的 Asset
6.2 发送 IoT Time Series 数据
6.2.1 TS 数据准备
根据如下的 TS 数据，客户端每 10 秒会向前 3 个数据点发送数据（dataPoint1, dataPoint2, dataPoint3）；每 20 秒向后三个数据点发送数据（dataPoint4, dataPoint5, dataPoint6）。
{
"timeseries": [
{
"timestamp": "2022-03-17T02:05:00.000Z",
"values": [
{
"dataPointId": "dataPoint1",
"value": 20.1,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint2",
"value": 555,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint3",
"value": 2147483648,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint4",
"value": "error",
"qualityCode": "0"
},
{
"dataPointId": "dataPoint5",
"value": "true",
"qualityCode": "0"
},
{
"dataPointId": "dataPoint6",
"value": "2022-02-10T01:05:20.020Z",
"qualityCode": "0"
}
]
},
{
"timestamp": "2022-03-17T02:05:10.000Z",
"values": [
{
"dataPointId": "dataPoint1",
"value": 34.14,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint2",
"value": 654,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint3",
"value": 4294967296,
"qualityCode": "0"
}
]
},
{
"timestamp": "2022-03-17T02:05:20.000Z",
"values": [
{
"dataPointId": "dataPoint1",
"value": 80.6,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint2",
"value": 565,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint3",
"value": 8589934592,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint4",
"value": "error",
"qualityCode": "0"
},
{
"dataPointId": "dataPoint5",
"value": "true",
"qualityCode": "0"
},
{
"dataPointId": "dataPoint6",
"value": "2022-02-10T01:10:20.020Z",
"qualityCode": "0"
}
]
},
{
"timestamp": "2022-03-17T02:05:30.000Z",
"values": [
{
"dataPointId": "dataPoint1",
"value": 43.13,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint2",
"value": 456,
"qualityCode": "0"
},
{
"dataPointId": "dataPoint3",
"value": 8589934595,
"qualityCode": "0"
}
]
}
]
}
6.2.1 发送、查看 TS 数据
1. 数据上传 Topic - tc/<tenantId>/<clientId>/o/mc_v3/ts
打开 MQTT.fx，将上述准备好的 TS 数据消息体发送到 topic。
2. 回到 Insights Hub Monitor 应用中查看刚才上传的数据。
将时间范围调整为上传的 TS 时间数据范围，然后数据显示模式调为：Raw。下面为上传过后的数据展示。
6.3 发送 Event
- Event id 必须要存在与请求 Pyaload 中。must be sent as id in the payload.
- Event type 必须要存在与 Event Management 服务中，否则 event 将会被丢弃。The event type must exist in Event Management. Otherwise, the event is dropped.
- Severity 可选。但是带有无效 severity 的 eventi 会被丢弃。Severity 目前可用的值有：20 (ERROR), 30 (WARNING), 40 (INFORMATION)。
6.3.1 创建自定义 Event Type
首先通过调用 Event Management API (POST - /api/eventmanagement/v3/eventTypes) 来创建自定义的 Event 类型。
参考如下格式：
{
"name": "FileUploadCompletedEvent",
"parentId": "core.connectivity.event.type.AgentBaseEvent",
"ttl": 35,
"scope": "LOCAL",
"fields": [
{
"name": "fileNumber",
"filterable": true,
"required": false,
"updatable": true,
"type": "INTEGER"
},
{
"name": "fileSize",
"filterable": true,
"required": false,
"updatable": true,
"type": "DOUBLE"
},
{
"name": "fileName",
"filterable": true,
"required": true,
"updatable": true,
"type": "STRING"
},
{
"name": "isValid",
"filterable": true,
"required": false,
"updatable": true,
"type": "BOOLEAN"
}
]
}
下面用 Postman 调用 API 来创建自定义 Event Type
6.3.2 发送 Event
Event Type 创建完后，便可以通过 MC MQTT 向 Asset 发送 Event。
发送 Event 消息体如下：
{
"events": [
{
"id": "e6e31c6f-7963-4f62-92a9-46b1b8b0649",
"correlationId": "eventCorrelationId",
"timestamp": "2022-03-17T15:45:02.302Z",
"severity": 20,
"type": "FileUploadCompletedEvent",
"description": "File upload test event",
"details": {
"fileName": "test2",
"fileSize": 11.2,
"isValid": "true",
"fileNumber": 15
}
}
]
}
向 Topic - tc/<tenantId>/<clientId>/o/mc_v3/e 发送创建 Event 消息。
消息发送完成后，回到 Insights Hub Monitor 应用中查看创建的消息：
6.4 发送 File
每个客户端可通过 MQTT 上传最大 75 KB 的文件。文件会被上传到对应的 Asset 中，例如设备日志文件或者其它较复杂的传感器结构定义之类的内容。上传后的文件可被对应 Asset 的父 Asset 引用。Insights Hub 不会解析这些文件的内容，使用者可以自行获取这些文件并解析。
Note: 文件内容需要经过 Base64 编码。
6.4.1 发送文件
1. 参考如下消息体
{
"file": {
"name": "samplefile",
"creationDate": "2022-03-17T15:46:02.302Z",
"content": "U2FtcGxlIEZpbGUgQ29udGVudA=="
}
}
2. 然后向 Topic - tc/<tenantId>/<clientId>/o/mc_v3/f 发送消息
3. 返送完成后回到 Monitor 中查看文件
注意：这里需要找到对应的 Agent Asset 来查看文件。
七、总结
至此，关于 MC MQTT 的介绍就结束了。更多的内容、细节描述可参考官方文档 来获取更多的介绍（包括本文所述的内容。但需要注意的是：Tenant CA 证书生成步骤需要参考本文中的步骤来生成，因为官方文档给出的步骤是基于公有云环境，与本地私有云（LPC）环境步骤有所差异）。