通过

你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

MQTT 路由消息的扩充

  • 项目

借助扩充支持，可以在将消息发送到事件网格自定义主题之前，将最多 20 个自定义键值属性添加到消息。 这些扩充使你能够：

  • 将上下文数据添加到消息。 例如，使用客户端名称或命名空间名称扩充消息可为终结点提供有关消息源的信息。
  • 减少终结点上的计算负载。 例如，使用 MQTT 发布请求的有效负载格式指示符或内容类型扩充消息会告知终结点如何处理消息的有效负载，而无需首先尝试多个分析器。
  • 根据添加的数据通过事件网格事件订阅筛选路由消息。 例如，扩充客户端属性可以根据不同属性的值筛选要路由到终结点的消息。

配置

扩充键：

扩充键是一个需要符合以下要求的字符串：

  • 仅包含小写字母数字：仅限 (a-z) 和 (0-9)
  • 不能为 specversion、id、time、type、source、subject、datacontenttype、dataschema、data 或 data_base64。
  • 不能以 azsp 开头。
  • 不能重复。
  • 不能超过 20 个字符。

扩充值：

扩充值可以是静态扩充的静态字符串，也可以是表示客户端属性或用于动态扩充的 MQTT 消息属性的受支持值之一。 扩充值不能超过 128 个字符。 以下列表包含支持的值：

客户端属性

  • ${client.authenticationName}：发布客户端的名称。
  • ${client.attributes.x}：发布客户端的属性，其中 x 表示属性键名称。

MQTT 属性

  • ${mqtt.message.userProperties.x}：MQTTv5 PUBLISH 数据包中的用户属性，其中 x 表示用户属性键名称
    • Type：string
    • 如果用户属性包含特殊字符 ${mqtt.message.userProperties['x']}，请改用以下变量格式。 仍需按如下所示转义撇号和反斜杠：“PN\t”变为“PN\t”。
  • ${mqtt.message.topicName}：MQTT PUBLISH 数据包中的主题。
    • Type：string
  • ${mqtt.message.responseTopic}：MQTTv5 PUBLISH 数据包中的响应主题。
    • Type：string
  • ${mqtt.message.correlationData}：MQTTv5 PUBLISH 数据包中的关联数据。
    • 类型：二进制
  • ${mqtt.message.pfi}：MQTTv5 PUBLISH 数据包中的有效负载格式指示符。
    • 类型：整数

Azure 门户配置

使用以下步骤配置路由扩充：

  1. 在 Azure 门户中转到你的命名空间。
  2. 在“路由”下，选中“启用路由”。
  3. 在路由主题下，选择已创建的事件网格主题，所有 MQTT 消息将路由到其中。
  4. 在“消息扩充”下，选择“+添加扩充”。
  5. 最多添加 20 个键值对，并相应地选择其类型。
  6. 选择“应用”。

Screenshot showing the routing enrichment configuration through the portal.

有关路由配置的详细信息，请参阅通过 Azure 门户配置路由。

Azure CLI 配置

在创建/更新命名空间期间使用命令和有效负载来配置路由扩充：

az resource create --resource-type Microsoft.EventGrid/namespaces --id /subscriptions/<Subscription ID>/resourceGroups/<Resource Group>/providers/Microsoft.EventGrid/namespaces/<Namespace Name> --is-full-object --api-version 2023-06-01-preview --properties @./resources/NS.json

NS.json

{
  "properties": {
    "topicSpacesConfiguration": {
        "state": "Enabled",
        "routeTopicResourceId": "/subscriptions/<Subscription ID>/resourceGroups/<Resource Group>/providers/Microsoft.EventGrid/topics/<Event Grid Topic name>",
        "routingEnrichments": {
            "static": [
                {
                    "key": "namespaceid",
                    "value": "123",
                    "valueType": "string"
                }
            ],
            "dynamic": [
                {
                    "key": "clientname",
                    "value": "${client.authenticationName}"
                },
                {
                    "key": "clienttype",
                    "value": "${client.attributes.type}"
                },
                {
                    "key": "address",
                    "value": "${mqtt.message.userProperties['client.address']}"
                },
                {
                    "key": "region",
                    "value": "${mqtt.message.userProperties.location}"
                },
                {
                    "key": "mqtttopic",
                    "value": "${mqtt.message.topicName}"
                },
                {
                    "key": "mqttresponsetopic",
                    "value": "${mqtt.message.responseTopic}"
                },
                {
                    "key": "mqttcorrelationdata",
                    "value": "${mqtt.message.correlationData}"
                },
                {
                    "key": "mqttpfi",
                    "value": "${mqtt.message.pfi}"
                }
            ]
        }
    }
},
"location": "eastus2euap",
"tags": {},
}

有关路由配置的详细信息，请参阅通过 Azure CLI 配置路由。

示例输出

以下 CloudEvent 是在应用前面的扩充配置后，包含 PFI=0 的 MQTTv5 消息的示例输出：

{
    "specversion": "1.0",
	"id": "9aeb0fdf-c01e-0131-0922-9eb54906e20", // unique id stamped by the service
	"time": "2019-11-18T15:13:39.4589254Z", // timestamp when messages was received by the service
	"type": "MQTT.EventPublished", // set type for all MQTT messages enveloped by the service
	"source": "testnamespace", // namespace name
	"subject": "campus/buildings/building17", // topic of the MQTT publish request
	"namespaceid": "123", // static enrichment
	"clientname": "client1", // dynamic enrichment of the name of the publishing client
	"clienttype": "operator", // dynamic enrichment of an attribute of the publishing client
	"address": "1 Microsoft Way, Redmond, WA 98052", // dynamic enrichment of a user property in the MQTT publish request
	"region": "North America", // dynamic enrichment of another user property in the MQTT publish request
	"mqtttopic": "campus/buildings/building17", // dynamic enrichment of the topic of the MQTT publish request
	"mqttresponsetopic": "campus/buildings/building17/response", // dynamic enrichment of the response topic of the MQTT publish request
	"mqttcorrelationdata": "cmVxdWVzdDE=", // dynamic enrichment of the correlation data of the MQTT publish request encoded in base64
	"mqttpfi": 0, // dynamic enrichment of the payload format indicator of the MQTT publish request
	"datacontenttype": "application/octet-stream", //content type of the MQTT publish request
	"data_base64": 
    {
	    IlRlbXAiOiAiNzAiLAoiaHVtaWRpdHkiOiAiNDAiCg==
	}
}

处理特殊情况：

  • 未指定的客户端属性/用户属性：如果动态扩充指向不存在的客户端属性/用户属性，则扩充将包含指定的键，该键的值为空字符串。 例如 emptyproperty: ""。
  • 数组：客户端属性中的数组和重复的用户属性将转换为逗号分隔的字符串。 例如：如果扩充的客户端属性设置为 "array": "value1", "value2", "value3"，则生成的扩充属性将是 array: value1,value2,value3。 另举一例：如果同一 MQTT 发布请求包含用户属性 > "userproperty1": "value1", "userproperty1": "value2"，则生成的扩充属性将是 userproperty1: value1,value2。

后续步骤：

使用以下文章详细了解路由：

快速入门：

概念：