如何使用 Protocol Buffers 在 AWS IoT Core 构建智能应用程序 官方博客

利用 AWS IoT Core 搭建智能应用程序的 Protocol Buffers

作者 Jos Gardiazabal 和 Donato Azevedo日期 2023 年 3 月 21 日分类 高级 (300) AWS IoT Core 技术如何做永久链接

关键要点

Protocol BuffersProtobuf是一种高效的数据序列化方式，优于 JSON。AWS IoT Core 允许连接大量 IoT 设备，并支持通过 Protobuf 整合高效消息传递。使用 Protobuf 的自动化代码生成可以简化开发流程，并提高数据交换的安全性和一致性。通过 AWS IoT Core 的 Rules Engine，可以轻松实现 Protobuf 消息的过滤和转发。

引言

Protocol Buffers或称 Protobuf是一种平台中立的数据序列化方式，旨在以高效、快速的方式处理结构化数据，相比于 JSON Protobuf 更小、更快，且能够自动生成所需的编程语言绑定。

AWS IoT Core 是一种托管服务，能够连接数十亿个 IoT 设备，并将数万亿条消息路由到 AWS 服务，从而使应用程序可以无缝扩展到数百万个设备。通过将 AWS IoT Core 与 Protobuf 集成，您可以利用 Protobuf 的数据序列化协议和自动生成的代码绑定。

IoT 中的灵活性与安全性：Protobuf 代码生成的优势

使用 Protobuf 的代码生成器进行软件开发的一个重要优势在于其易用性和安全性。您可以编写架构来描述应用程序各组件之间交换的消息。代码生成器如 protoc 或其他工具解析该架构，并在您选择的编程语言中实现编码和解码函数。Protobuf 的代码生成器得到了良好的维护和广泛应用，从而产生坚固且经过考验的代码。

自动化代码生成 使开发者无需编写编码和解码函数，并确保了不同编程语言之间的兼容性。与最新发布的 AWS IoT Core 的规则引擎支持 Protobuf 消息格式 结合使用后，您便可以在设备上使用 C 编写的生产者应用程序，与用 Python 编写的 AWS Lambda 函数消费者共同使用生成的绑定。

使用 Protocol Buffers 而非 JSON 进行 AWS IoT Core 的其他好处包括：

优势描述架构和验证架构由发送者和接收者共同强制执行，确保正确集成。编码和解码由自动生成的代码处理，避免了错误。适应性架构是可变的，可以在保持向前和向后兼容的同时更改消息内容。带宽优化Protobuf 消息的大小相较于 JSON 更小，随着时间的推移，这提供了更好的设备自主性和较少的带宽使用。高效解码解码 Protobuf 的效率高于 JSON。这意味着接收方功能运行的时间更短，Protobuf 在相同消息负载下的性能可以超过 JSON 6 倍。

本文将带您了解如何部署一个示例应用程序，该应用程序使用 Protobuf 格式向 AWS IoT Core 发布消息。这些消息随后将通过 AWS IoT Core 规则引擎 进行选择性过滤。

Protocol Buffers 简述

消息架构是 Protobuf 的关键元素。架构示例如下：

protobufsyntax = proto3

import google/protobuf/timestampproto

message Telemetry{ enum MsgType { MSGTYPENORMAL = 0 MSGTYPEALERT = 1 } MsgType msgType = 1 string instrumentTag = 2 googleprotobufTimestamp timestamp = 3 double value = 4}

架构的第一行定义了您所使用的 Protocol Buffers 的版本。本文将使用 proto3 版本语法，而 proto2 也得到了支持。

接下来一行指示将描述一个名为 Telemetry 的新消息定义。该消息包含四个字段：

msgType 字段：类型为 MsgType，仅能接受枚举值 MSGTYPENORMAL 或 MSGTYPEALERT。instrumentTag 字段：类型为 string，标识发送遥测数据的测量仪器。timestamp 字段：类型为 googleprotobufTimestamp，表示测量的时间。value 字段：类型为 double，包含测量的值。

欲了解所有支持的数据类型及语法的更多信息，请参见 完整文档。

以 JSON 格式描述的 Telemetry 消息如下所示：

json{ msgType MSGTYPEALERT instrumentTag Temperature001 timestamp 1676059669 value 725}

相同的协议缓冲消息以 base64 编码显示如下：

0801120F54656D70657261747572652D3030311A060895C89A9F06210000000000205240

请注意，JSON 表示的消息大小为 115 字节，而 Protobuf 的消息大小仅为 36 字节。

一旦定义了架构，便可使用 protoc 来：

在您选择的编程语言中创建绑定。创建一个 FileDescriptorSet，用于 AWS IoT Core 解码接收到的消息。

在 AWS IoT Core 中使用 Protocol Buffers

Protobuf 可以通过多种方式与 AWS IoT Core 一起使用。最简单的方法是将消息作为 二进制负载 发布，接收应用程序则解码该负载。这已被 AWS IoT Core 规则引擎支持，并适用于任何二进制负载，不仅限于 Protobuf。

然而，当您想要解码 Protobuf 消息以进行过滤和转发时，将获得最大的价值。被过滤的消息可以作为 Protobuf 转发，或解码为 JSON 兼容于只能理解 JSON 格式的应用程序。

最新发布的 AWS IoT 规则引擎支持 Protobuf 消息格式 允许您以最少的努力以受管理的方式实现这一点。在接下来的部分中，我们将指导您如何部署和运行一个示例应用程序。

前提条件

要运行此示例应用程序，您必须具备以下条件：

一台安装了 protoc 的计算机。请参考您操作系统的 官方安装说明。安装 AWS CLI (安装说明)一个具备有效凭证的 AWS 账户，且对 Amazon S3、AWS IAM、AWS IoT Core 和 AWS CloudFormation 拥有完全权限。Python 37 或更高版本。

示例应用程序：过滤和转发 Protobuf 消息为 JSON

要部署和运行示例应用程序，我们将执行 7 个简单步骤：

下载示例代码并安装 Python 依赖。配置您的 IOTENDPOINT 和 AWSREGION 环境变量。使用 protoc 生成 Python 绑定和消息描述符。使用 Python 和 Protobuf 生成的代码绑定运行模拟设备。使用 AWS CloudFormation 创建 AWS 资源并上传 Protobuf 文件描述符。检查与之匹配、过滤并将 Protobuf 消息重新发布为 JSON 的 AWS IoT 规则。验证转换后的消息是否被重新发布。

步骤 1：下载示例代码并安装 Python 依赖

要运行示例应用程序，请先下载代码并安装其依赖项：

首先，从我们的 AWS GitHub 存储库下载并解压示例应用程序 https//githubcom/awssamples/awsiotcoreprotobufsample如果您将其下载为 ZIP 文件，请将其解压。在解压后的示例应用程序文件夹中，运行以下命令安装所需的 Python 依赖项：

bashpip install r requirementstxt

上述命令将安装两个必需的 Python 依赖项：boto3AWS 的 Python SDK和 protobuf。

步骤 2：配置您的 IOTENDPOINT 和 AWSREGION 环境变量

我们的模拟 IoT 设备将连接到 AWS IoT Core 端点以发送 Protobuf 格式的消息。

如果您在 Linux 或 Mac 上运行，请运行以下命令。确保将 ltAWSREGIONgt 替换为您选择的 AWS 区域。

bashexport AWSREGION=ltAWSREGIONgtexport IOTENDPOINT=(aws iot describeendpoint endpointtype iotDataATS query endpointAddress region AWSREGION output text)

步骤 3：使用 protoc 生成 Python 绑定和消息描述符

解压后的示例应用程序包含一个名为 msgproto 的文件，类似于我们之前介绍的架构示例。

运行以下命令以生成您的模拟设备将使用的代码绑定：

bashprotoc pythonout= msgprotoprotoc includeimports o filedescriptordesc msgproto

运行完这些命令后，您应该会在当前文件夹中看到两个新文件：

filedescriptordesc 和 msgpb2py。

步骤 4：运行模拟设备，使用 Python 和 Protobuf 生成的代码绑定

解压后的示例应用程序包含一个名为 simulatedevicepy 的文件。

西游加速器网站

要启动模拟设备，请运行以下命令：

bashpython3 simulatedevicepy

在 AWS 控制台中使用MQTT 测试客户端确认消息正在发送到 AWS IoT Core。

访问 AWS IoT Core 服务控制台 https//consoleawsamazoncom/iot 确保您位于正确的 AWS 区域。在 测试 中，选择 MQTT 测试客户端。在 主题过滤器 下，填入 test/telemetryall。展开 附加配置 部分，在 MQTT 负载显示 选项中选择 显示原始负载。点击 订阅，观察 Protobuf 格式的消息如何到达 AWS IoT Core MQTT 代理。

步骤 5：使用 AWS CloudFormation 创建 AWS 资源并上传 Protobuf 文件描述符

解压后的示例应用程序中包含一个名为 supportinfrastructuretemplateyaml 的 AWS CloudFormation 模板。

该模板定义了一个 Amazon S3 Bucket、一个 AWS IAM 角色及一个 AWS IoT 规则。

运行以下命令将 CloudFormation 模板部署到您的 AWS 账户中。确保将 ltYOURBUCKETNAMEgt 和 ltAWSREGIONgt 替换为您选择的 S3 Bucket 的唯一名称和 AWS 区域。

bashaws cloudformation createstack stackname IotBlogPostSample templatebody file//supportinfrastructuretemplateyaml capabilities CAPABILITYNAMEDIAM parameters ParameterKey=FileDescriptorBucketNameParameterValue=ltYOURBUCKETNAMEgt region=ltAWSREGIONgt

AWS IoT Core 对 Protobuf 格式消息的支持需要我们用 protoc 生成的文件描述符。为了使其可用，我们需要将其上传到新创建的 S3 Bucket 中。运行以下命令上传文件描述符，确保将 ltYOURBUCKETNAMEgt 替换为您在部署 CloudFormation 模板时使用的相同名称。

bashaws s3 cp filedescriptordesc s3//ltYOURBUCKETNAMEgt/msg/filedescriptordesc

步骤 6：检查与之匹配、过滤并将 Protobuf 消息重新发布为 JSON 的 AWS IoT 规则

假设您想要过滤 msgType 为 MSGTYPEALERT 的消息，因为这些表明可能存在危险的操作条件。CloudFormation 模板创建了一个 AWS IoT 规则，该规则解码模拟设备发送到 AWS IoT Core 的 Protobuf 格式消息，然后选择那些标记为警报的消息并以 JSON 格式重新发布，以便其他 MQTT 主题响应者可以订阅。要检查 AWS IoT 规则，请执行以下步骤：

访问 AWS IoT Core 服务控制台 https//consoleawsamazoncom/iot在左侧菜单中，点击 消息路由 下的 规则。列表中将包含名为 ProtobufAlertRule 的 AWS IoT 规则，点击以查看详细信息。在 SQL 语句下，注意 SQL 语句，我们将在稍后讨论每个元素的含义。在操作下，注意单个操作以 重新发布到 AWS IoT 主题。

sqlSELECT VALUE decode(encode( base64) proto ltYOURBUCKETNAMEgt msg/filedescriptordesc msg Telemetry)FROM test/telemetryallWHERE decode(encode( base64) proto ltYOURBUCKETNAMEgt msg/filedescriptordesc msg Telemetry)msgType = MSGTYPEALERT

此 SQL 语句执行以下操作：

SELECT VALUE decode() 指出整个解码后的 Protobuf 负载将以 JSON 负载的形式重新发布到目标 AWS IoT 主题。如果希望仍以 Protobuf 格式转发消息，可以将其替换为简单的 SELECT 。WHERE decode()msgType = MSGTYPEALERT 将解码传入的 Protobuf 格式消息，仅包含 msgType 字段值为 MSGTYPEALERT 的消息将被转发。

步骤 7：验证转换后的消息是否被重新发布

如果您点击该 AWS IoT 规则中的单个操作，您将注意到它将消息重新发布到 topic/telemetryalerts 主题。

要订阅该主题并查看是否重新发布了 JSON 格式的消息，请执行以下步骤：

访问 AWS IoT Core 服务控制台 https//consoleawsamazoncom/iot在 测试 中，选择 MQTT 测试客户端。在 主题过滤器 下，填入 test/telemetryalerts。展开 附加配置 部分，确保选中 自动格式化 JSON 负载 选项。点击 订阅，观察 msgType MSGTYPEALERT 的 JSON 转换消息如何到达。

如果您检查模拟设备的代码，您将注意到大约 20 的模拟消息为 MSGTYPEALERT，每 5 秒发送一次消息。您可能需要等待以查看警报消息的到来。

清理

运行完该示例后，要清理环境，请运行以下命令：

bash

从 Amazon S3 Bucket 删除文件描述符对象

aws s3 rm s3///msg/filedescriptordesc

从 IoT 服务角色中分离所有策略

aws iam detachrolepolicy rolename IoTCoreServiceSampleRole policyarn (aws iam listattachedrolepolicies rolename IoTCoreServiceSampleRole query AttachedPolicies[0]PolicyArn output text)

删除 AWS CloudFormation 堆栈

aws cloudformation deletestack stackname IotBlogPostSample

结论

如前所述，在 AWS IoT Core 中使用 Protobuf 是编写 SQL 语句般简单。Protobuf 消息在成本节省带宽使用减少、设备自主性提升和任何支持 protoc 的编程语言的开发便捷性方面，相较于 JSON 具有明显优势。

有关使用 AWS IoT Core 规则引擎解码 Protobuf 格式消息的额外细节，请查阅 AWS IoT Core 文档。

示例代码可以在 GitHub 存储库中找到：https//githubcom/awssamples/awsiotcoreprotobufsample。

decode 函数在向 Amazon Kinesis Data Firehose 转发数据时特别有用，因为其将接受 JSON 输入，无需您编写 AWS Lambda