View a markdown version of this page

批处理 HTTP 操作消息 - Amazon IoT Core
概述批量使用 HTTP 标头有效载荷示例限制批处理的错误操作使用批处理 HTTP 操作消息 Amazon CLI
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

批处理 HTTP 操作消息

您可以使用批处理在单个请求中发送多条 HTTP 操作消息。

概述

批处理使您能够将消息从 Amazon IoT Core 规则引擎批量发送到您的 HTTP 终端节点。此功能可以通过降低 HTTP 操作执行次数来帮助降低成本，并通过减少与建立新连接相关的开销来提高效率。

注意

批处理的 HTTP 操作按单个操作进行计量。根据 Amazon IoT Core 规则引擎向下游服务发出的出站批量负载的大小，以 5 KiB 为增量计量。有关更多信息，请参阅 Amazon IoT Core 价格页面。

当您在物联网规则操作的定义中启用批处理时，以下参数可供配置：

maxBatchOpenMs

传出消息等待其他消息创建批次的最大时间（以毫秒为单位）。设置越高，批处理 HTTP 操作的延迟越长。

最小值：5 毫秒。最大值：200 毫秒。

默认值：20 毫秒

支持替换模板：否

maxBatchSize

在单个 IoT 规则操作执行中批处理在一起的最大消息数。

最小值：2 条消息。最大值：10 条消息

默认值：10 条消息

支持替换模板：否

maxBatchSizeBytes

消息批次的最大大小，以字节为单位。

最小值：100 字节。最大值：131,072 字节

默认值：5120 字节

支持替换模板：否

batchAcrossTopics

是否允许将来自不同 MQTT 主题的消息批处理到单个 HTTP 请求中。默认情况下，只有来自同一主题的消息才会批处理在一起。对于来自多个设备主题的消息发往同一 HTTP 端点的路由用例，启用此参数。

默认值：false

支持替换模板：否

重要

当您指定多个批处理参数时，当达到第一个限制时，批处理即完成。例如，如果您指定 100 毫秒作为最大批处理打开时间，将 5 KiB 指定为最大批处理大小，而规则引擎在 100 毫秒内仅批处理 2 KiB，则将创建并发送 2 KiB 的批处理。

注意

消息始终在相同的账户、规则名称、目标 HTTP 端点 URL 和账单组的范围内进行批处理。无论batchAcrossTopics设置如何，这些属性中任何不同的消息都不会合并到同一个批次中。

批量使用 HTTP 标头

当您在 HTTP 操作中使用标头时，批处理请求使用添加到批处理的最后一条消息（不一定是您发布的最后一条消息）中的标头值。我们建议使用以下任一标头值：

  • 批处理中的所有消息都相同

  • 适用于所有消息（例如，身份验证凭证）

标头与 HTTP 请求一起发送，不是消息正文的一部分。

注意

启用批处理时：

  • 批处理请求自动包含Content-Type: application/json标头，因为批处理以 JSON 数组的形式发送。

  • 我们无法保证批处理中的最后一条消息是您发布的最后一条消息。这是最后一条进入批处理的消息。

有效载荷示例

以下示例显示了发送到您的 HTTP 端点的批处理消息负载的结构：

[
  {
    "user_id": "user1",
    "steps_today": 1000
  },
  {
    "user_id": "user2",
    "steps_today": 21000
  },
  {
    "user_id": "user8",
    "steps_today": 1500
  },
  ...
]

限制

以下是批处理的限制：

  • Amazon IoT Core 不能保证消息的整体顺序。批处理在每台主机上本地执行，这可能会导致批处理中的消息以与接收顺序不同的顺序进行处理。

  • Amazon IoT Core 不在接收方提供消息处理支持。您有责任确保将下游服务配置为批量接受和处理数据。

  • Cross-account 即使消息的目的地是相同的资源标识符（HTTP URL 或资源 ARN），也不支持批处理。

  • Amazon IoT Core 不能保证批量大小将满足您指定的配置。根据时间和消息流，批次可能小于您配置的限制。

  • 启用批处理后，不支持二进制有效载荷（非UTF-8 数据）。仅接受 UTF-8 文本有效载荷（例如 JSON）。要发送二进制数据，请在将其发送到 HTTP 操作之前 base64 对其进行编码，然后在接收端点对其进行解码。例如，您可以使用物联网规则中的编码功能对二进制负载进行编码。或者，您可以在物联网设备中对二进制负载进行编码并将其发布到 Amazon IoT Core。

批处理的错误操作

您将无法在错误操作定义中定义单独的批处理逻辑。但是，如果您在主操作中定义了批处理逻辑，则您的错误操作将支持批处理。

当批量请求失败时， Amazon IoT Core 规则引擎将遵循 HTTP 操作重试逻辑。在最后一次重试尝试之后，将对整个失败的批次调用错误操作。

以下是启用批处理的错误操作消息的示例：

{
    "ruleName": "FailedTopicRule",
    "topic": "topic/rulesengine",
    "payloadsWithMetadata": [
        {
            "id": 1,
            "cloudwatchTraceId": "bebd6d93-6d4a-899e-9e40-56e82252d2be",
            "clientId": "Test",
            "sourceIp": "10.0.0.0",
            "base64OriginalPayload": "eyJ1c2VyX2lkIjogInVzZXI1NjQ3IiwgInN0ZXBzX3RvZGF5IjogMTMzNjUsICJ0aW1lc3RhbXAiOiAiMjAyNS0xMC0wOVQwNzoyMjo1OC45ODQ3OTAxNzZaIn0="
        },
        {
            "id": 2,
            "cloudwatchTraceId": "af94d3b8-0b18-1dbf-2c7d-513f5cb9e2e1",
            "clientId": "Test",
            "sourceIp": "10.0.0.0",
            "base64OriginalPayload": "eyJ1c2VyX2lkIjogInVzZXI1NjQ3IiwgInN0ZXBzX3RvZGF5IjogMTMzNjUsICJ0aW1lc3RhbXAiOiAiMjAyNS0xMC0wOVQwNzoyMjo1OC45ODQ3OTAxNzZaIn0="
        },
        {
            "id": 3,
            "cloudwatchTraceId": "ca441266-c2ce-c916-6aee-b9e5c7831675",
            "clientId": "Test",
            "sourceIp": "10.0.0.0",
            "base64OriginalPayload": "eyJ1c2VyX2lkIjogInVzZXI1NjQ3IiwgInN0ZXBzX3RvZGF5IjogMTMzNjUsICJ0aW1lc3RhbXAiOiAiMjAyNS0xMC0wOVQwNzoyMjo1OC45ODQ3OTAxNzZaIn0="
        }
    ],
    "failures": [
        {
            "affectedIds": [
                1,
                2,
                3
            ],
            "failedAction": "HttpAction",
            "failedResource": "https://example.foobar.com/HttpAction",
            "errorMessage": "HttpAction failed to make a request to the specified endpoint. StatusCode: 500. Reason: Internal Server Error."
        },
        {
            "affectedIds": [
                3
            ],
            "failedAction": "S3Action",
            "failedResource": "amzn-s3-demo-bucket",
            "errorMessage": "Failed to put S3 object. The error received was The specified bucket does not exist"
        },
        {
            "affectedIds": [
                3
            ],
            "failedAction": "LambdaAction",
            "failedResource": "arn:aws:lambda:us-west-2:123456789012:function:dummy",
            "errorMessage": "Failed to invoke lambda function. Received Server error from Lambda. The error code is 403"
        }
    ]
}

启用后batchAcrossTopics，错误操作负载格式会发生变化。该topic字段从顶层移动到每个payloadsWithMetadata条目内部：

{
    "ruleName": "FailedTopicRule",
    "payloadsWithMetadata": [
        {
            "id": 1,
            "topic": "topic/sensor1",
            "cloudwatchTraceId": "bebd6d93-6d4a-899e-9e40-56e82252d2be",
            "clientId": "Test",
            "sourceIp": "10.0.0.0",
            "base64OriginalPayload": "eyJ1c2VyX2lkIjogInVzZXI1NjQ3IiwgInN0ZXBzX3RvZGF5IjogMTMzNjUsICJ0aW1lc3RhbXAiOiAiMjAyNS0xMC0wOVQwNzoyMjo1OC45ODQ3OTAxNzZaIn0="
        },
        {
            "id": 2,
            "topic": "topic/sensor2",
            "cloudwatchTraceId": "af94d3b8-0b18-1dbf-2c7d-513f5cb9e2e1",
            "clientId": "Test",
            "sourceIp": "10.0.0.0",
            "base64OriginalPayload": "eyJ1c2VyX2lkIjogInVzZXI1NjQ3IiwgInN0ZXBzX3RvZGF5IjogMTMzNjUsICJ0aW1lc3RhbXAiOiAiMjAyNS0xMC0wOVQwNzoyMjo1OC45ODQ3OTAxNzZaIn0="
        }
    ],
    "failures": [
        {
            "affectedIds": [1, 2],
            "failedAction": "HttpAction",
            "failedResource": "https://example.foobar.com/HttpAction",
            "errorMessage": "HttpAction failed to make a request to the specified endpoint. StatusCode: 500. Reason: Internal Server Error."
        }
    ]
}
注意

批处理操作失败还会生成更大的错误操作有效负载，这会因规模大小而增加错误操作失败的可能性。您可以使用该ErrorActionFailure指标监控错误操作失败。请参阅规则操作指标了解更多信息。

使用批处理 HTTP 操作消息 Amazon CLI

使用批处理创建或更新规则操作

  1. 使用相应的 Amazon CLI 命令创建或更新规则：

    • 要创建新规则，请使用 create-topic-ru le 命令：

      aws iot create-topic-rule --rule-name myrule --topic-rule-payload file://myrule.json

    • 要更新现有规则，请使用 replace-topic-r ule 命令：

      aws iot replace-topic-rule --rule-name myrule --topic-rule-payload file://myrule.json

  2. 通过在主题规则负载中将 enableBatching 参数设置为 true 来启用批处理功能：

    {
        "topicRulePayload": {
        "sql": "SELECT * FROM 'some/topic'", 
        "ruleDisabled": false,
        "awsIotSqlVersion": "2016-03-23", 
        "actions": [
            { 
                "http": { 
                    "url": "https://www.example.com/subpath",
                    "confirmationUrl": "https://www.example.com", 
                    "headers": [
                        { 
                            "key": "static_header_key", 
                            "value": "static_header_value" 
                        },
                        { 
                            "key": "substitutable_header_key", 
                            "value": "${value_from_payload}" 
                         }
                    ],
                    "enableBatching": true,
                    "batchConfig": {
                       "maxBatchOpenMs": 100,
                       "maxBatchSize": 5,
                       "maxBatchSizeBytes": 1024,
                       "batchAcrossTopics": true
                    }
                }
            }
      ]
}

  3. 配置批处理参数。您无需指定所有批处理参数。您可以选择指定 1、2、3 或全部 4 个批处理参数。如果您未指定批处理参数，则规则引擎将使用默认值更新该参数。有关批处理参数及其默认值的更多信息，请参阅 HTTP 参数。