使用 IoT 中心消息路由将设备到云消息发送到不同的终结点

注意

本文中提到的某些功能(例如云到设备消息传递、设备孪生、设备管理)仅在 IoT 中心的标准层中提供。 有关基本和标准 IoT 中心层的详细信息,请参阅如何选择合适的 IoT 中心层

消息路由使你能够以自动、可缩放以及可靠的方式将消息从设备发送到云服务。 消息路由可用于:

  • 发送设备遥测消息以及事件(即设备生命周期事件、设备孪生更改事件、数字孪生体更改事件和设备连接状态事件)到内置终结点和自定义终结点。 了解有关路由终结点

  • 在将数据路由到各个终结点之前对数据进行筛选,筛选方法是通过应用丰富的查询。 消息路由允许你查询消息属性和消息正文以及设备孪生标记和设备孪生属性。 深入了解如何使用消息路由中的查询

IoT 中心需要这些服务终结点的写入权限,以便使用消息路由。 如果通过 Azure 门户配置终结点,则为你添加必要权限。 请确保将服务配置为支持预期吞吐量。 例如,如果使用事件中心作为自定义终结点,则必须为该事件中心配置吞吐量单位,以便它可以处理你计划通过 IoT 中心消息路由发送的事件流入量。 同样,使用服务总线队列作为终结点时,必须配置最大大小,以确保队列可以容纳所有流入的数据,直到它被使用者传出。 在首次配置 IoT 解决方案时,可能需要监视其他终结点,并针对实际负载进行任何必要的调整。

IoT 中心为所有设备到云的消息传送定义了格式,以便实现跨协议互操作性。 如果某条消息与多个路由匹配,而这些路由指向同一终结点,则 IoT 中心仅向该终结点传递一次消息。 因此无需在服务总线队列或主题中配置重复数据删除。 使用本教程了解如何配置消息路由

路由终结点

IoT 中心有一个默认的内置终结点(消息/事件),此终结点与事件中心兼容。 可以通过将订阅中的其他服务链接到中心来创建要将消息路由到的自定义终结点

每条消息都路由到与它的路由查询匹配的所有终结点。 换句话说,消息可以路由到多个终结点。

如果自定义终结点具有防火墙配置,请考虑使用 Microsoft 受信任的第一方例外情况

IoT 中心目前支持以下终结点:

  • 内置终结点
  • 存储容器
  • 服务总线队列和服务总线主题
  • 事件中心
  • Cosmos DB(预览版)

内置终结点作为路由终结点

可以使用标准事件中心集成和 SDK 接收来自内置终结点(消息/事件)的设备到云的消息。 在创建一个路由后,数据将停止流向内置终结点,除非创建了到该终结点的路由。 即使未创建路由,也必须启用回退路由,以便将消息路由到内置终结点。 如果使用门户或 CLI 创建中心,则默认情况下会启用回退。

Azure 存储作为路由终结点

有两个存储服务 IoT 中心可将消息路由到:Azure Blob 存储Azure Data Lake Storage Gen2 (ADLS Gen2) 帐户。 Azure Data Lake Storage 帐户是在 Blob 存储之上构建的启用分层命名空间的存储帐户。 这两个存储服务都使用 blob 作为其存储。

IoT 中心支持以 Apache Avro 格式和 JSON 格式将数据写入 Azure 存储。 默认值为 AVRO。 使用 JSON 编码时,必须在消息系统属性中将 contentType 设置为 application/json,将 contentEncoding 设置为 UTF-8。 这两个值都不区分大小写。 如果未设置内容编码,则 IoT 中心将以 base 64 编码格式写入消息。

只有在配置 Blob 存储终结点时才能设置编码格式,不能编辑现有终结点的编码格式。 若要切换现有终结点的编码格式,则需要删除该终结点并使用所需格式重新创建。 一个有用的策略可能是创建具有所需编码格式的新自定义终结点,并将并行路由添加到该终结点。 通过这种方式,可以在删除现有终结点之前验证数据。

可以使用 IoT 中心的创建或更新 REST API(具体说来就是 RoutingStorageContainerPropertiesAzure 门户Azure CLIAzure PowerShell)选择编码格式。 下图说明如何在 Azure 门户中选择编码格式。

Blob 存储终结点编码

IoT 中心将在消息达到特定大小或在经过一定的时间后,对消息进行批处理并将数据写入存储。 IoT 中心默认为以下文件命名约定:

{iothub}/{partition}/{YYYY}/{MM}/{DD}/{HH}/{mm}

你可以使用任何文件命名约定,但必须使用所有列出的令牌。 如果没有要写入的数据,IoT 中心会写入到一个空 blob。

我们建议列出 blob 或文件,然后循环访问它们,以确保在未进行有关分区的任何假设的情况下读取所有 blob 或文件。 在 Microsoft 发起的故障转移或 IoT 中心手动故障转移期间,分区范围可能发生变化。 可以使用 List Blobs API 枚举 blob 列表,或使用 List ADLS Gen2 API 枚举文件列表。 请将以下示例作为指南。

public void ListBlobsInContainer(string containerName, string iothub)
{
    var storageAccount = CloudStorageAccount.Parse(this.blobConnectionString);
    var cloudBlobContainer = storageAccount.CreateCloudBlobClient().GetContainerReference(containerName);
    if (cloudBlobContainer.Exists())
    {
        var results = cloudBlobContainer.ListBlobs(prefix: $"{iothub}/");
        foreach (IListBlobItem item in results)
        {
            Console.WriteLine(item.Uri);
        }
    }
}

若要创建与 Azure Data Lake Gen2 兼容的存储帐户,请创建新的 V2 存储帐户,并在“高级”选项卡的“分层命名空间”字段上选择“启用”,如下图所示:

选择 Azure Date Lake Gen2 存储

服务总线队列和服务总线主题作为路由终结点

用作 IoT 中心终结点的服务总线队列和主题不能启用“会话”或“重复项检测”。 如果启用了其中任一选项,该终结点将在 Azure 门户中显示为“无法访问”。

事件中心作为路由终结点

除了与事件中心兼容的内置终结点外,还可以将数据路由到事件中心类型的自定义终结点。

Azure Cosmos DB 作为路由终结点(预览版)

可以将数据直接从 IoT 中心发送到 Azure Cosmos DB。 Cosmos DB 是一种完全托管的超大规模多型号数据库服务。 它的延迟非常低,可用性很高,因此非常适合需要大量下游数据分析的连接解决方案和制造之类的方案。

IoT 中心支持将内容以 JSON(如果在消息的 content-type 中进行了指定)格式或 Base64 编码二进制文件形式写入 Cosmos DB。 若要设置到 Cosmos DB 的路由,必须执行以下操作:

从预配的 IoT 中心转到“中心设置”,然后单击“消息路由”。 转到“自定义终结点”选项卡,单击“添加”,然后选择“Cosmos DB”。 下图显示了添加终结点的操作:

屏幕截图显示如何添加 Cosmos DB 终结点。

输入终结点名称。 你应能够从可供选择的 Cosmos DB 帐户列表以及数据库和集合中进行选择。

由于 Cosmos DB 是超大规模数据存储,因此写入到其中的所有数据/文档都必须包含一个表示逻辑分区的字段。 分区键属性名称在容器级别定义,设置后无法更改。 每个逻辑分区的最大大小为 20 GB。 若要有效地支持大规模方案,可以为 Cosmos DB 终结点启用综合分区键,并根据估计的数据量对其进行配置。 例如,在制造方案中,逻辑分区可能会在一个月内达到 20 GB 的最大限制。 在这种情况下,可以定义综合分区键,它是设备 ID 和月份的组合。 此键将自动添加到每个新 Cosmos DB 记录的分区键字段,确保每月为每个设备创建逻辑分区。

 可以根据系统设置选择任何受支持的身份验证类型来访问数据库。

注意

如果使用系统托管标识向 CosmosDB 进行身份验证,则需要通过 CLI 分配“Cosmos DB 内置数据参与者”角色。 目前不支持从门户进行角色设置。 有关各种角色的更多详细信息,请参阅为 Azure Cosmos DB 配置基于角色的访问。 若要了解如何通过 CLI 分配角色,请参阅管理 Azure Cosmos DB SQL 角色资源

选择所有详细信息后,请单击“创建”,完成自定义终结点的设置。

读取已路由的数据

可以按照此教程配置一个路由。

使用以下教程了解如何从终结点读取消息。

回退路由

回退路由将所有不满足任何现有路由上的查询条件的消息发送到与事件中心兼容的内置事件中心(消息/事件)。 如果已启用消息路由,则可以启用此回退路由功能。 在创建一个路由后,数据将停止流向内置终结点,除非创建了到该终结点的路由。 如果没有到内置终结点的路由并且已启用回退路由,则仅与路由上的任何查询条件不匹配的消息将被发送到内置终结点。 此外,如果已删除现有路由,必须启用回退路由才能接收内置终结点处的所有数据。

可以在 Azure 门户>“消息路由”边栏选项卡中启用/禁用回退路由。 还可以将 Azure 资源管理器用于 FallbackRouteProperties 来为回退路由使用自定义终结点。

非遥测事件

除了设备遥测数据之外,消息路由还支持发送设备孪生更改事件、设备生命周期事件、数字孪生体更改事件和设备连接状态事件。 例如,如果使用数据源创建一个设置为到设备孪生更改事件的路由,IoT 中心会将消息发送到包含设备孪生更改的终结点。 同样,如果创建路由时将数据源设置为“设备生命周期事件”,则 IoT 中心会发送一条消息,指示是否删除或创建了设备或模块。 有关设备生命周期事件的详细信息,请参阅设备和模块生命周期通知

IoT 中心还集成了 Azure 事件网格来发布设备事件以支持基于这些事件的工作流的实时集成和自动化。 请参阅消息路由和事件网格之间的主要区别来了解哪种更适合你的方案。

设备连接状态事件的限制

设备连接状态事件适用于使用 MQTT 或 AMQP 协议进行连接的设备,或者通过 WebSocket 使用这些协议之一进行连接的设备。 仅使用 HTTPS 发出的请求不会触发设备连接状态通知。 若要让 IoT 中心开始发送设备连接状态事件,在打开连接后,设备必须调用云到设备的接收消息操作或设备到云的发送遥测数据操作。 在 Azure IoT SDK 之外,在 MQTT 中,这些操作相当于对相应的消息主题的 SUBSCRIBE 或 PUBLISH 操作。 通过 AMQP,这些操作相当于在相应的链接路径上附加或传输消息。

IoT 中心不报告每个单独的设备连接和断开连接,而是发布在定期的 60 秒快照时获取的当前连接状态。 接收具有不同序列号或不同连接状态事件的相同连接状态事件均意味着设备连接状态在 60 秒时段内发生了变化。

测试路由

在创建新路由或编辑现有路由时,应通过示例消息来测试路由查询。 可以测试单个路由或一次测试所有路由,并且在测试期间,不会有消息被路由到终结点。 可以使用 Azure 门户、Azure 资源管理器、Azure PowerShell 和 Azure CLI 来进行测试。 测试结果有助于确定示例消息是否与查询相匹配,或者测试是否因为示例消息或查询语法错误而无法运行。 若要了解详细信息,请参阅测试路由测试所有路由

延迟

使用内置终结点路由设备到云遥测消息时,在创建第一个路由后,端到端延迟略微增大。

在大多数情况下,延迟的平均增量小于 500 毫秒。 但是,你体验到的延迟可能会有所不同,并且可能会更高,具体取决于你的 IoT 中心和解决方案体系结构的层级。 可以使用路由:消息/事件的消息延迟或 d2c.endpoints.latency.builtIn.events IoT 中心指标来监视延迟 。 在创建第一个路由后创建或删除任何路由不会影响端到端延迟。

监视和故障排除

IoT 中心提供了多个与路由和终结点相关的指标,使你能够大致了解你的中心的运行状况和已发送的消息数。 有关按功能类别细分的所有 IoT 中心指标的列表,请参阅监视数据参考中的指标。 可以使用 IoT 中心资源日志中的“路由”类别,跟踪发生在路由查询和终结点运行状况评估期间的由 IoT 中心察觉到的错误。 若要详细了解如何使用 IoT 中心的指标和资源日志,请参阅监视 IoT 中心

可以使用 REST API Get Endpoint Health 获取终结点的运行状况状态

通过路由故障排除指南获取更多详细信息以及对路由故障排除的支持。

后续步骤