配置指南 - MQTT配置 - 《Thingsboard 二次开发中文文档》

扩展配置: mqtt-config.json
- 基本连接属性
客户端身份属性
映射
下一步

本指南将帮助您熟悉 ThingsBoard IoT 网关的 MQTT 扩展配置。使用常规配置启用此扩展。我们将在下面描述扩展配置文件。

扩展配置: mqtt-config.json

Extension configuration is a JSON file that contains information about how to connect and monitor a list of MQTT brokers. The root JSON element should contain “brokers” array. Each broker in the array is configured using following properties:扩展配置是一个 JSON 文件，包含关于如何连接和并监控 MQTT 代理列表的信息。根 JSON 元素应该包含 “代理” 数组。数组中的每个代理都使用以下属性配置:

基本连接属性

属性	描述	默认值
host	MQTT 代理地址	localhost
port	MQTT 代理端口	1883
ssl	启用/禁用加密连接的布尔标志位	false
truststore	用于加密连接的 Truststore 文件的路径	空
truststorePassword	Truststore 的密码	空
retryInterval	重连间隔(毫秒)	3000
clientId	MQTT 客户端ID。如果没有设置，则使用自动生成的UUID。	随机 UUID

例如:

{
  "brokers": [
    {
      "host": "localhost",
      "port": 1883,
      "ssl": false,
      "retryInterval": 3000
      ...
    }
  ]
}

客户端身份属性

MQTT 扩展支持以下客户端认证:

anonymous
basic
cert.PEM

匿名身份配置示例:

{
      ...
      "credentials": {
        "type": "anonymous"
      }
      ...
}

基本身份配置示例:

{
      ...
      "credentials": {
        "type": "basic",
        "username": "Your username",
        "password": "Your password"
      }
      ...
}

PEM 证书身份配置示例:

{
    ...
    "credentials": {
        "type": "cert.PEM",
        "caCert" : "Path to Your CA certificate file",
        "privateKey" : "Path to Your private key file",
        "cert" : "Path to Your certificate file"
    }
    ...
}

映射

本节将描述各种配置选项，允许指定 ThingsBoard 功能和现有/遗留设备协议之间的映射。

属性和时间序列数据上传

MQTT 主题监控的 “映射” 配置设置规则，并将数据转换为 ThingsBoard 键值格式。例如:

{
...
"mapping": [
        {
          "topicFilter": "sensors",
          "converter": {
            "type": "json",
            "filterExpression": "",
            "deviceNameJsonExpression": "${$.serialNumber}",
            "timeout": 60000,
            "attributes": [
              {
                "type": "string",
                "key": "model",
                "value": "${$.model}"
              }
            ],
            "timeseries": [
              {
                "type": "double",
                "key": "temperature",
                "value": "${$.temperature}"
              }
            ]
          }
        }
        ...
        ]
}

注意: 可以在数组中指定多个映射对象。
映射过程使用映射对象的 topicFilter 参数订阅 MQTT 主题。分析其他设备或应用程序发布到本主题的每条消息，以提取设备名称和设备数据 (属性或时间序列值)。默认情况下，网关使用 Json 转换器，但是您可以自定义这种行为并实现自己的转换器。有关更多详细信息，请参见转换器接口。
转换器基于 JsonPath 库，提供 JSON 结构的灵活映射和过滤能力。您可以根据路径和过滤器示例定义 filterExpression 。
deviceNameJsonExpression 映射参数用于从传入的 JSON 消息中提取设备名称。
timeout 映射参数是非活跃超时 (毫秒)选项。如果指定了此参数，一旦在特定超时内没有新的遥测或属性更新，网关将报告设备断开。
例如，如果您有以下消息:

{"serialNumber":"SN-001", "model":"T1000", "temperature":36.6}

“${$.serialNumber}“ 表达式将会返回“SN-001”设备名称。类似的, “Device ${$.serialNumber} (${$.model})”将会返回 “Device SN-001 (T1000)” 设备名称。
假如设备名称是 MQTT 主题的一部分，你可以使用 deviceNameTopicExpression 参数，这是一个基于 Java 的正则表达式。
举个例子，如果你有 “sensor/SN-001/temperature”主题, ”(?<=sensor\/)(.*?)(?=\/temperature)” 表示将会返回“SN-001” 设备名称.
类似的 JsonPath 映射规则会被用于属性与时序的值:

key - 常量属性或时间序列 ThingsBoard 键。
type - either boolean, long, double or string.

value - 用 ${} 包裹的基于 JsonPath 语法的表达式

设备连接和断开请求

ThingsBoard 允许向设备发送关于设备属性更新的 RPC 命令和通知。但是为了发送它们，平台需要知道目标设备是否连接，以及目前使用什么网关或会话来连接设备。如果您的设备不断发送遥测数据，ThingsBoard 已经知道如何推送通知。如果您的设备刚刚连接到 MQTT 代理并等待命令/更新，您需要向网关发送一条消息，并通知设备已连接到代理。
“connectRequests” 和 “disconnectRequests” 允许将 MQTT 消息转换为相应的事件，这些事件将通过网关传递给 ThingsBoard。例如:

{
...
    "connectRequests": [
      {
        "topicFilter": "sensors/connect",
        "deviceNameJsonExpression": "${$.serialNumber}"
      },
      {
        "topicFilter": "sensor/+/connect",
        "deviceNameTopicExpression": "(?<=sensor\/)(.*?)(?=\/connect)"
      }
    ]
...
}

TopicFilter 接受通配符 (‘+’ 和 ‘#’)。设备名称可以用 deviceNameTopicExpression 从主题中获取，也可以从传入的 json 消息中获取。
在这种情况下，以下消息是有效的:

mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensors/connect" -m '{"serialNumber":"SN-001"}'
mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensor/SN-001/connect" -m ''

注意: “connectRequests” 和 “disconnectRequests” 的配置结构和语法是相同的。

设备属性请求

ThingsBoard 允许提供设备属性并从设备应用程序中获取其中的一些属性。您可以将此视为设备的远程配置。您的设备能够从 ThingsBoard 请求客户端和共享属性。有关更多详细信息，请参见用户指南。
“attributeRequests” 配置允许配置相应属性请求和响应消息的格式。
例如:

{
...
    "attributeRequests": [
      {
        "topicFilter": "sensors/attributes",
        "deviceNameJsonExpression": "${$.serialNumber}",
        "attributeKeyJsonExpression": "${$.key}",
        "requestIdJsonExpression": "${$.requestId}",
        "clientScope": false,
        "responseTopicExpression": "sensors/${deviceName}/attributes/${responseId}",
        "valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
      },
      {
        "topicFilter": "sensors/+/attributes/+/request/+",
        "deviceNameTopicExpression": "(?<=sensors\/)(.*?)(?=\/attributes)",
        "attributeKeyTopicExpression": "(?<=attributes\/)(.*?)(?=\/request)",
        "requestIdTopicExpression": "(?<=request\/)(.*?)($)",
        "clientScope": false,
        "responseTopicExpression": "sensors/${deviceName}/attributes/${attributeKey}/response/${requestId}",
        "valueExpression": "${attributeValue}"
      }
    ]
...
}

下表列出了相应的配置属性及其描述。

属性	描述	强制	例子
topicFilter	允许通配符 (‘+’ 和 ‘#’)的主题过滤器	yes	sensors/+/attributes/+/request/+
deviceNameJsonExpression	从消息JSON中提取设备名称的 JsonPath 表达式	deviceNameJsonExpression 或 deviceNameTopicExpression 二选一	${$.serialNumber}
attributeKeyJsonExpression	从消息JSON中提取属性键的 JsonPath 表达式	attributeKeyJsonExpression 或 attributeKeyTopicExpression 二选一	${$.key}
requestIdJsonExpression	从消息JSON中提取请求 id 的 JsonPath 表达式	requestIdJsonExpression 或 requestIdTopicExpression 二选一	${$.requestId}
deviceNameTopicExpression	从主题中获取设备名称的正则表达式	deviceNameTopicExpression 或 deviceNameJsonExpression 二选一	(?<=sensors\/)(.*?)(?=\/attributes)
attributeKeyTopicExpression	从主题中获取属性键的正则表达式	attributeKeyTopicExpression 或 attributeKeyJsonExpression 二选一	(?<=attributes\/)(.*?)(?=\/request)
requestIdTopicExpression	从主题中获取请求id的正则表达式	requestIdTopicExpression 或 requestIdJsonExpression 二选一	(?<=request\/)(.*?)($)
clientScope	标识属性的范围：客户端或共享	yes	true 表示客户端属性，false 表示其他
responseTopicExpression	将要发布的响应对应的主题的表达式。您可以在此表达式中使用 ${deviceName}，${requestId}和${attributeKey}。	yes	sensors/${deviceName}/attributes/${attributeKey}/response/${requestId}
valueExpression	将要发布的响应消息的表示，你可以在表示式中使用 ${deviceName},${requestId} ${attributeKey} 和 ${attributeValue}	yes	${attributeValue}

在这种情况下，以下消息是有效的:

mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensors/attributes" -m '{"serialNumber":"SN-001", "key":"dataUploadFrequency", "requestId": 123}'
mosquitto_pub -h YOUR_MQTT_BROKER_HOST -p YOUR_MQTT_BROKER_PORT -t "sensors/SN-001/attributes/dataUploadFrequency/request/123" -m ''

设备属性更新

ThingsBoard 允许订阅设备应用程序中共享设备属性的更改。有关更多详细信息，请参见用户指南。
“attributeUpdates” 配置允许配置相应订阅消息的格式。
例如:

{
...
      "attributeUpdates": [
        {
          "deviceNameFilter": ".*",
          "attributeFilter": ".*",
          "topicExpression": "sensor/${deviceName}/${attributeKey}",
          "valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
        }
      ]
...
}

如您所见，您可以过滤掉设备和属性，并将此更新转发到特定主题。主题和值表达式支持以下内容
参数: ${deviceName},${requestId} ${attributeKey} 和 ${attributeValue}.

服务器端 RPC 命令

ThingsBoard 允许向直接或通过网关连接到 ThingsBoard 的设备发送 RPC 命令。
“serverSideRpc“ 允许配置请求和响应 MQTT 消息的格式。例如:

{
...
      "serverSideRpc": [
        {
          "deviceNameFilter": ".*",
          "methodFilter": "echo",
          "requestTopicExpression": "sensor/${deviceName}/request/${methodName}/${requestId}",
          "responseTopicExpression": "sensor/${deviceName}/response/${methodName}/${requestId}",
          "responseTimeout": 10000,
          "valueExpression": "${params}"
        },
        {
          "deviceNameFilter": ".*",
          "methodFilter": "no-reply",
          "requestTopicExpression": "sensor/${deviceName}/request/${methodName}/${requestId}",
          "valueExpression": "${params}"
        }
      ]
...
}

正如您可以使用 deviceNameFilter 和 methodFilter 为不同的设备/方法应用不同的映射规则一样。一旦网关从服务器接收到设备的 RPC 请求，它将基于 requestTopicExpression 和 valueExpression 发布相应的消息。如果您期望来自设备的请求得到回复，您还应该指定 responseTopicExpression 和 responseTimeout。网关将订阅 “response” 主题，并等待设备回复，直到检测到 “timeout”。
用 javascript 编写的设备应用程序示例:

var mqtt = require('mqtt');
var broker = 'mqtt://127.0.0.1:1883';
var device = 'SN-001';
var client  = mqtt.connect(broker, {
  // Will topic to report that device is disconnected 
  will: {topic: 'sensor/' + device + '/disconnect', payload: '', qos: 1}
});
client.on('connect', function () {
    console.log('connected');
    // Report that device is connected to the gateway
    client.publish('sensor/' + device + '/connect', '');
    // Subscribe to RPC requests topic
    client.subscribe('sensor/' + device + '/request/+/+');
});
client.on('message', function (topic, message) {
    console.log('request.topic: ' + topic);
    console.log('request.body: ' + message.toString());
    console.log('response.topic: ' + topic.replace('request','response'));
    // Publish the response that will simply echo the request.
    client.publish(topic.replace('request','response'), message);
});

需要从服务器发送的 RPC 请求 (rpc-request.json) 示例:

{
  "method": "echo",
  "params": {
    "message": "Hello!"
  },
  "timeout": 5000
}

使用 CURL 的 POST 命令示例:

$ SERVER_BASE_URL=http://localhost:8080
$ JWT_TOKEN=YOUR_TOKEN_HERE
$ DEVICE_ID=YOUR_DEVICE_ID_HERE
$ curl -v -X POST -d @rpc-request.json $SERVER_BASE_URL/api/plugins/rpc/twoway/$DEVICE_ID \
--header "Content-Type:application/json" \
--header "X-Authorization: Bearer $JWT_TOKEN"

下一步

探索例子:

Mosquitto connection example

探索与主要 ThingsBoard 功能相关的指南:

Data Visualization - 如何可视化收集的数据。
Device attributes - 如何使用设备属性。
Telemetry data collection - 如何采集遥测数据。
Using RPC capabilities - 如何向设备发送命令。
Rule Engine - 如何使用规则引擎分析来自设备的数据。