本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 在 Amazon SQS 中使用 API 本主题提供有关构建 Amazon SQS 端点、通过 GET 和 POST 方法发出查询 API 请求以及使用批处理 API 操作的信息。有关 Amazon SQS [操作](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_Operations.html)（包括参数、错误、示例和[数据类型](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_Types.html)）的详细信息，请参阅 [https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/)。要使用各种编程语言访问 Amazon SQS，您还可以使用包含以下自动功能的 [AWS SDK](https://aws.amazon.com/tools/#sdk)： + 对服务请求进行加密签名 + 重试请求 + 处理错误响应有关更多信息，请参阅 [将 Amazon SQS 与 AWS SDK 结合使用](sdk-general-information-section.md)。有关命令行工具信息，请参阅 [https://docs.aws.amazon.com/cli/latest/reference/sqs/index.html](https://docs.aws.amazon.com/cli/latest/reference/sqs/index.html)和[https://docs.aws.amazon.com/powershell/latest/reference/](https://docs.aws.amazon.com/powershell/latest/reference/)中的 Amazon SQS 部分。 **采用 AWS JSON 协议的 Amazon SQS API** Amazon SQS 使用 AWS JSON 协议作为指定 [AWS SDK 版本](sqs-json-faqs.md#json-protocol-getting-started)上所有 Amazon SQS API 的传输机制。AWSJSON 协议提供更高的吞吐量、更低的延迟和更快的应用程序到应用程序通信。AWS与 AWS 查询协议相比，JSON 协议在请求和响应的序列化与反序列化方面效率更高。如果您仍然倾向于在 SQS API 中使用 AWS 查询协议，请参阅 [亚马逊 SQS APIs 中使用的 AWS JSON 协议支持哪些语言？](sqs-json-faqs.md#json-protocol-supported-languages)了解支持 Amazon SQS AWS 查询协议的 AWS SDK 版本。 Amazon SQS 使用 AWS JSON 协议在 AWS SDK 客户端（例如 Java、Python、Golang、JavaScript）和 Amazon SQS 服务器之间进行通信。Amazon SQS API 操作的 HTTP 请求接受 JSON 格式的输入。将执行 Amazon SQS 操作，并将执行响应以 JSON 格式发送回 SDK 客户端。与 AWS 查询相比，AWS JSON 在客户端和服务器之间传输数据方面效率更高。 + AWS JSON 协议充当 Amazon SQS 客户端和服务器之间的中介。 + 服务器不理解创建 Amazon SQS 操作所用的编程语言，但它能理解 AWS JSON 协议。 + AWS JSON 协议在 Amazon SQS 客户端和服务器之间使用序列化（将对象转换为 JSON 格式）和反序列化（将 JSON 格式转换为对象）。有关 Amazon SQS 所使用 AWS JSON 协议的更多信息，请参阅 [亚马逊 SQS AWS JSON 协议 FAQs](sqs-json-faqs.md)。 AWS JSON 协议适用于指定的 [AWS SDK 版本](sqs-json-faqs.md#json-protocol-getting-started)。要查看不同语言变体的 SDK 版本和发布日期，请参阅《AWS SDK 和工具参考指南》**中的 [AWS SDK 和工具版本支持列表](https://docs.aws.amazon.com/sdkref/latest/guide/version-support-matrix.html)。 # 在 Amazon SQS 中使用 AWS JSON 协议发出查询 API 请求本主题介绍了如何构造 Amazon SQS 端点、发出 POST 请求以及解释响应。 **注意** AWS 大多数语言变体都支持 JSON 协议。有关受支持语言变体的完整列表，请参阅[亚马逊 SQS APIs 中使用的 AWS JSON 协议支持哪些语言？](sqs-json-faqs.md#json-protocol-supported-languages)。 ## 构造端点为了使用 Amazon SQS 队列，您必须构造一个端点。有关 Amazon SQS 端点的信息，请参阅 *Amazon Web Services 一般参考*中的以下页面： + [区域端点](https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints) + [Amazon Simple Queue Service 端点和配额](https://docs.aws.amazon.com/general/latest/gr/sqs-service) 每个 Amazon SQS 端点都是独立的。例如，如果有两个名为 *MyQueue* 的队列，其中一个队列具有终端节点 `sqs.us-east-2.amazonaws.com`，另一个队列具有终端节点 `sqs.eu-west-2.amazonaws.com`，则这两个队列不会相互共享任何数据。以下是一个提出创建队列请求的端点的示例。 ``` POST / HTTP/1.1 Host: sqs.us-west-2.amazonaws.com X-Amz-Target: AmazonSQS.CreateQueue X-Amz-Date: Content-Type: application/x-amz-json-1.0 Authorization: Content-Length: Connection: Keep-Alive { "QueueName":"MyQueue", "Attributes": { "VisibilityTimeout": "40" }, "tags": { "QueueType": "Production" } } ``` **注意** 队列名称和队 URLs 列区分大小写。 *`AUTHPARAMS`* 的结构取决于 API 请求的签名。有关更多信息，请参阅[《*亚马逊 Web Services 一般参考*》中的 “签署 AWS API 请求](https://docs.aws.amazon.com/general/latest/gr/signing_aws_api_requests.html)”。 ## 发出 POST 请求 Amazon SQS POST 请求在 HTTP 请求的正文中以表单的形式发送查询参数。以下是将 `X-Amz-Target` 设置为 `AmazonSQS.` 的 HTTP 标头以及将 `Content-Type` 设置为 `application/x-amz-json-1.0` 的 HTTP 标头的示例。 ``` POST / HTTP/1.1 Host: sqs.. X-Amz-Target: AmazonSQS.SendMessage X-Amz-Date: Content-Type: application/x-amz-json-1.0 Authorization: Content-Length: Connection: Keep-Alive { "QueueUrl": "https://sqs..///", "MessageBody": "This is a test message" } ``` 此 HTTP POST 请求将消息发送到 Amazon SQS 队列。 **注意** HTTP 标头 `X-Amz-Target` 和 `Content-Type` 均为必需项。根据客户端的 HTTP 版本，您的 HTTP 客户端可能会向 HTTP 请求添加其他项目。 # 解释 Amazon SQS JSON API 响应当您向 Amazon SQS 发送请求时，它会返回包含结果的 JSON 响应。响应结构取决于您使用的 API 操作。要了解这些响应的详细信息，请参阅： + 《Amazon Simple Queue Service API 参考》**的 [API 操作](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_Operations.html)中的特定 API 操作 + 这些区域有：[亚马逊 SQS AWS JSON 协议 FAQs](sqs-json-faqs.md) ## 成功的 JSON 响应结构如果请求成功，则主响应元素为 `x-amzn-RequestId`，其中包含请求的通用唯一标识符 (UUID) 以及其他附加的响应字段。例如，以下 [https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_CreateQueue.html](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_CreateQueue.html) 响应包含 `QueueUrl` 字段，后者又包含所创建队列的 URL。 ``` HTTP/1.1 200 OK x-amzn-RequestId: Content-Length: Date: Content-Type: application/x-amz-json-1.0 { "QueueUrl":"https://sqs.us-east-1.amazonaws.com/111122223333/MyQueue" } ``` ## JSON 错误响应结构如果请求失败，则 Amazon SQS 将返回主响应，包括 HTTP 标头和正文。在 HTTP 标头中，`x-amzn-RequestId` 包含请求的 UUID。`x-amzn-query-error` 包含两条信息：错误类型，以及错误是创建者错误还是使用者错误。在响应正文中，`"__type"` 表示其他错误详细信息，`Message` 以可读的格式指明错误情况。以下是 JSON 格式的错误响应示例： ``` HTTP/1.1 400 Bad Request x-amzn-RequestId: 66916324-67ca-54bb-a410-3f567a7a0571 x-amzn-query-error: AWS.SimpleQueueService.NonExistentQueue;Sender Content-Length: Date: Content-Type: application/x-amz-json-1.0 { "__type": "com.amazonaws.sqs#QueueDoesNotExist", "message": "The specified queue does not exist." } ``` # 亚马逊 SQS AWS JSON 协议 FAQs 本主题涵盖了有关在 Amazon SQS 中使用 AWS JSON 协议的常见问题。 ## 什么是 AWS JSON 协议，它与现有的 Amazon SQS API 请求和响应有何不同？ JSON 是在异构系统之间进行通信时最广为使用和接受的连接方法之一。亚马逊 SQS 使用 JSON 作为媒介在 S AWS DK 客户端（例如 Java、Python、Golang JavaScript）和亚马逊 SQS 服务器之间进行通信。Amazon SQS API 操作的 HTTP 请求接受 JSON 形式的输入。系统会执行 Amazon SQS 操作，然后将执行的响应以 JSON 的形式反过来共享给 SDK 客户端。与 AWS 查询相比，JSON 在客户端和服务器之间的数据传输方面效率更高。 + 亚马逊 SQS AWS JSON 协议充当亚马逊 SQS 客户端和服务器之间的中介。 + 服务器不理解创建 Amazon SQS 操作所用的编程语言，但它能理解 AWS JSON 协议。 + 亚马逊 SQS AWS JSON 协议在亚马逊 SQS 客户端和服务器之间使用序列化（将对象转换为 JSON 格式）和反序列化（将 JSON 格式转换为对象）。 ## 如何开始使用适用于亚马逊 SQS 的 AWS JSON 协议？要开始使用最新版本的 AWS SDK，以便更快地向 Amazon SQS 发送消息，请将您的 AWS 软件开发工具包升级到指定版本或任何后续版本。要详细了解 SDK 客户端，请参阅下表中的“指南”一列。以下是适用于亚马逊 SQS APIs 的 JS AWS ON 协议跨语言变体的软件开发工具包版本列表： | 语言 | SDK 客户端存储库 | 所需的 SDK 客户端版本 | 指南 | | --- | --- | --- | --- | | C\$1\$1 | [啊/ aws-sdk-cpp](https://github.com/aws/aws-sdk-cpp) | [1.11.98](https://github.com/aws/aws-sdk-cpp/releases/tag/1.11.198) | [AWS 适用于 C\$1\$1 的 SDK](https://aws.amazon.com/sdk-for-cpp/) | | Golang 1.x | [啊/ aws-sdk-go](https://github.com/aws/aws-sdk-go) | [v1.47.7](https://github.com/aws/aws-sdk-go/releases/tag/v1.47.7) | [AWS 适用于 Go 的 SDK](https://aws.amazon.com/sdk-for-go/) | | Golang 2.x | [aws/ 2 aws-sdk-go-v](https://github.com/aws/aws-sdk-go-v2) | [v1.28.0](https://github.com/aws/aws-sdk-go-v2/blob/release-2023-11-09/service/sqs/CHANGELOG.md#v1270-2023-11-09) | [AWS 适用于 Go V2 的 SDK](https://aws.github.io/aws-sdk-go-v2/docs/) | | Java 1.x | [啊/ aws-sdk-java](https://github.com/aws/aws-sdk-java) | [1.12.585](https://github.com/aws/aws-sdk-java/releases/tag/1.12.585) | [AWS 适用于 Java 的 SDK](https://aws.amazon.com/sdk-for-java/) | | Java 2.x | [aws/ 2 aws-sdk-java-v](https://github.com/aws/aws-sdk-java-v2) | [2.21.19](https://github.com/aws/aws-sdk-java-v2/releases/tag/2.21.19) | [AWS 适用于 Java 的 SDK](https://aws.amazon.com/sdk-for-java/) | | JavaScript v2.x | [啊/ aws-sdk-js](https://github.com/aws/aws-sdk-js) | [JavaScript on AWS](https://aws.amazon.com/developer/language/javascript/) | | JavaScript v3.x | [aws/ 3 aws-sdk-js-v](https://github.com/aws/aws-sdk-js-v3) | [v3.447.0](https://github.com/aws/aws-sdk-js-v3/releases/tag/v3.447.0) | [JavaScript on AWS](https://aws.amazon.com/developer/language/javascript/) | | .NET | [啊/ aws-sdk-net](https://github.com/aws/aws-sdk-net) | [3.7.681.0](https://github.com/aws/aws-sdk-net/releases/tag/3.7.681.0) | [AWS 适用于 .NET 的 SDK](https://aws.amazon.com/sdk-for-net/) | | PHP | [啊/ aws-sdk-php](https://github.com/aws/aws-sdk-php) | [3.285.2](https://github.com/aws/aws-sdk-php/releases/tag/3.285.2) | [AWS 适用于 PHP 的 SDK](https://aws.amazon.com/sdk-for-php/) | | Python-boto3 | [boto/boto3](https://github.com/boto/boto3) | [1.28.82](https://github.com/boto/boto3/releases/tag/1.28.82) | [AWS 适用于 Python (Boto3) 的 SDK](https://aws.amazon.com/sdk-for-python/) | | Python-botocore | [boto/botocore](https://github.com/boto/botocore/) | [1.31.82](https://github.com/boto/botocore/releases/tag/1.31.82) | [AWS 适用于 Python (Boto3) 的 SDK](https://aws.amazon.com/sdk-for-python/) | | awscli | [AWS CLI](https://github.com/aws/aws-cli) | [1.29.82](https://github.com/aws/aws-cli/releases/tag/1.29.82) | [AWS命令行界面](https://aws.amazon.com/cli/) | | Ruby | [啊/ aws-sdk-ruby](https://github.com/aws/aws-sdk-ruby) | [1.67.0](https://rubygems.org/gems/aws-sdk-sqs/versions/1.67.0) | [AWS 适用于 Ruby 的 SDK](https://aws.amazon.com/sdk-for-ruby/) | ## 为我的 Amazon SQS 工作负载启用 JSON 协议有什么风险？如果您使用软件开发工具包的自定义实现或自定义客户端和 AWS AWS 软件开发工具包的组合来与生成基于 AWS 查询（又名基于 XML）的响应的 Amazon SQS 进行交互，则可能与 JSON 协议不兼容。 AWS 如果您遇到任何问题，请联系 Supp AWS ort。 ## 如果我已经使用最新的 AWS SDK 版本，但我的开源解决方案不支持 JSON，该怎么办？您必须将 SDK 版本更改为当前所用版本之前的版本。有关[如何开始使用适用于亚马逊 SQS 的 AWS JSON 协议？](#json-protocol-getting-started)更多信息，请参阅。 AWS 中列出的软件开发工具包版本[如何开始使用适用于亚马逊 SQS 的 AWS JSON 协议？](#json-protocol-getting-started)使用适用于 Amazon SQS APIs 的 JSON 有线协议。如果您将 AWS 软件开发工具包更改为先前版本，则您的 Amazon SQS APIs 将使用该查询。 AWS ## 亚马逊 SQS APIs 中使用的 AWS JSON 协议支持哪些语言？ Amazon SQS 支持所有通用语言变体 (GA)。 AWS SDKs 目前，我们不支持 Kotlin、Rust 或 Swift。要详细了解其他语言变体，请参阅[用于在 AWS上进行构建的工具](https://aws.amazon.com/developer/tools/)。 ## 亚马逊 SQS 中使用的 AWS JSON 协议支持哪些区域 APIs 亚马逊 SQS 在所有提供亚马逊 SQS 的[AWS 地区](https://docs.aws.amazon.com/general/latest/gr/sqs-service.html)都支持 AWS JSON 协议。 ## 使用 JS AWS ON 协议升级到适用于 Amazon SQS 的指定 AWS 软件开发工具包版本时，我可以期待哪些延迟改善？ AWS 与查询协议相比，JSON 协议在请求和响应的序列化和反序列化方面效率更高。 AWS 根据对 5 KB 消息负载的 AWS 性能测试，适用于 Amazon SQS 的 JSON 协议可将 end-to-end消息处理延迟减少多达 23%，并减少应用程序客户端 CPU 和内存使用量。 ## AWS 查询协议会被弃用吗？ AWS 将继续支持查询协议。只要你的 AWS SDK 版本设置了除[如何开始使用 Amazon SQS 的 JS AWS ON 协议中列出的版本之外的任何先前版本，你就可以继续使用 AWS 查询协议](#json-protocol-getting-started)。 ## 在哪里可以找到有关 AWS JSON 协议的更多信息？您可以在 *Smithy* 文档的 [AWS JSON 1.0 协议](https://smithy.io/2.0/aws/protocols/aws-json-1_0-protocol.html)中找到有关 JSON 协议的更多信息。有关使用 AWS JSON 协议的 Amazon SQS API 请求的更多信息，请参阅[在 Amazon SQS 中使用 AWS JSON 协议发出查询 API 请求](sqs-making-api-requests-json.md)。 # 在 Amazon SQS 中使用 AWS 查询协议发出查询 API 请求本主题介绍了如何构造 Amazon SQS 端点、发出 GET 和 POST 请求以及解释响应。 ## 构造端点为了使用 Amazon SQS 队列，您必须构造一个端点。有关 Amazon SQS 端点的信息，请参阅 *Amazon Web Services 一般参考*中的以下页面： + [区域端点](https://docs.aws.amazon.com/general/latest/gr/rande.html#sqs_region) + [Amazon Simple Queue Service 端点和配额](https://docs.aws.amazon.com/general/latest/gr/sqs-service) 每个 Amazon SQS 端点都是独立的。例如，如果有两个名为 *MyQueue* 的队列，其中一个队列具有终端节点 `sqs.us-east-2.amazonaws.com`，另一个队列具有终端节点 `sqs.eu-west-2.amazonaws.com`，则这两个队列不会相互共享任何数据。以下是一个提出创建队列请求的端点的示例。 ``` https://sqs.eu-west-2.amazonaws.com/ ?Action=CreateQueue &DefaultVisibilityTimeout=40 &QueueName=MyQueue &Version=2012-11-05 &AUTHPARAMS ``` **注意** 队列名称和队 URLs 列区分大小写。 *`AUTHPARAMS`* 的结构取决于 API 请求的签名。有关更多信息，请参阅*亚马逊 Web Services 一般参考*中的[签署 AWS API 请求](https://docs.aws.amazon.com/general/latest/gr/signing_aws_api_requests.html)。 ## 提出 GET 请求 Amazon SQS GET 请求的结构是一个 URL，其中包含以下部分： + **端点** – 作为请求操作对象的资源（[队列名称和 URL](sqs-queue-message-identifiers.md#queue-name-url)），例如：`https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue` + **操作** – 要对端点执行的[操作](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_Operations.html)。端点与操作之间用问号 (`?`) 分隔，例如：`?Action=SendMessage&MessageBody=Your%20Message%20Text` + **参数** - 任何请求参数。各参数以和号 (`&`) 分隔，例如：`&Version=2012-11-05&AUTHPARAMS` 以下是一个 GET 请求的示例，该请求向 Amazon SQS 队列发送一条消息。 ``` https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue ?Action=SendMessage&MessageBody=Your%20message%20text &Version=2012-11-05 &AUTHPARAMS ``` **注意** 队列名称和队 URLs 列区分大小写。由于 GET 请求是 URLs，因此必须对所有参数值进行 URL 编码。由于不允许使用空格 URLs，因此每个空格都以 URL 编码为。`%20`示例的其余部分没有进行 URL 编码，以方便您阅读。 ## 发出 POST 请求 Amazon SQS POST 请求在 HTTP 请求的正文中以表单的形式发送查询参数。下面是一个将 `Content-Type` 设置为 `application/x-www-form-urlencoded` 的 HTTP 标头的示例。 ``` POST /123456789012/MyQueue HTTP/1.1 Host: sqs.us-east-2.amazonaws.com Content-Type: application/x-www-form-urlencoded ``` 该标头后跟一个 `[form-urlencoded](https://www.w3.org/MarkUp/html-spec/html-spec_8.html#SEC8.2)` GET 请求，该请求向 Amazon SQS 队列发送一条消息。各参数以和号 (`&`) 分隔。 ``` Action=SendMessage &MessageBody=Your+Message+Text &Expires=2020-10-15T12%3A00%3A00Z &Version=2012-11-05 &AUTHPARAMS ``` **注意** 仅 `Content-Type` HTTP 标头是必需的。`AUTHPARAMS` 对于 GET 请求是相同的。根据客户端的 HTTP 版本，您的 HTTP 客户端可能会向 HTTP 请求添加其他项目。 # 解释 Amazon SQS XML API 响应当您向 Amazon SQS 发送请求时，它会返回包含请求结果的 XML 响应。要了解这些响应的结构和详情，请参阅《Amazon Simple Queue Service API 参考》**中的特定 [API 操作](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_Operations.html)。 ## 成功的 XML 响应结构如果请求成功，则主要响应元素将以操作命名并附加上 `Response`（例如，`ActionNameResponse`)。此元素包含以下子元素： + **`ActionNameResult`** – 包含一个特定于操作的元素。例如，`CreateQueueResult` 元素包含 `QueueUrl` 元素，后者又包含所创建队列的 URL。 + **`ResponseMetadata`** – 包含 `RequestId`，后者又包含请求的通用唯一标识符 (UUID)。以下是 XML 格式的成功响应的示例： ``` https://sqs.us-east-2.amazonaws.com/770098461991/queue2 cb919c0a-9bce-4afe-9b48-9bdf2412bb67 ``` ## XML 错误响应结构如果请求不成功，则 Amazon SQS 将始终返回主要响应元素 `ErrorResponse`。此元素包含一个 `Error` 元素和一个 `RequestId` 元素。 `Error` 元素包含以下子元素： + **`Type`** – 指定错误是创建者错误还是使用者错误。 + **`Code`** – 指定错误类型。 + **`Message`** – 以可读格式指定错误情况。 + **`Detail`** –（可选）指定有关错误的其他详细信息。 `RequestId` 元素包含请求的 UUID。下面是 XML 格式的错误响应的示例： ``` Sender InvalidParameterValue Value (quename_nonalpha) for parameter QueueName is invalid. Must be an alphanumeric String of 1 to 80 in length. 42d59b56-7407-4c4a-be0f-4c88daeea257 ``` # 对 Amazon SQS 请求进行身份验证身份验证是用于识别和验证发送请求的当事方的过程。在身份验证的第一个阶段， AWS 将验证创建者的身份以及创建者是否[已注册使用 AWS](https://aws.amazon.com/)（有关更多信息，请参阅[步骤 1：创建 AWS 账户和 IAM 用户](sqs-setting-up.md#sqs-creating-aws-account)）。接下来，请 AWS 遵守以下程序： 1. 创建者（发件人）获取必要的凭证。 1. 创建者向使用者（接收方）发送请求和凭证。 1. 使用者使用证书来验证创建者是否发送了该请求。 1. 将出现以下情况之一： + 如果身份验证成功，使用者将处理该请求。 + 如果身份验证失败，使用者将拒绝请求并返回错误。 ## 使用 HMAC-SHA 的基本身份验证过程使用查询 API 访问 Amazon SQS 时，必须提供以下项来对请求进行身份验证： + 用于识别您的**AWS 访问密钥 ID** AWS 账户， AWS 用于查找您的私有访问密钥。 + **HMAC-SHA 请求签名**，使用您的私有访问密钥（只有您知道的共享密钥）计算得出，有关更多信息， AWS请参阅）。[RFC2104](http://www.faqs.org/rfcs/rfc2104.html)[AWS 开发工具包](https://aws.amazon.com/code/)可处理签名过程；但是，如果您通过 HTTP 或 HTTPS 提交查询请求，则必须在每个查询请求中包含一个签名。 1. 派生签名版本 4 签名密钥。有关更多信息，请参阅[使用 Java 派生签名密钥](https://docs.aws.amazon.com/general/latest/gr/signature-v4-examples.html#signature-v4-examples-java)。 **注意** Amazon SQS 支持签名版本 4，与之前的版本相比，该版本提供了SHA256基于改进的安全性和性能。创建使用 Amazon SQS 的新应用程序时，应使用 Signature Version 4。 1. 对请求签名必须采用 Base64 编码。下面的示例 Java 代码将执行此操作： ``` package amazon.webservices.common; // Define common routines for encoding data in AWS requests. public class Encoding { /* Perform base64 encoding of input bytes. * rawData is the array of bytes to be encoded. * return is the base64-encoded string representation of rawData. */ public static String EncodeBase64(byte[] rawData) { return Base64.encodeBytes(rawData); } } ``` + 请求的 **时间戳（或到期时间）**。在请求中使用的时间戳必须是 `dateTime` 对象，并包含[完整的日期以及小时、分钟和秒](http://www.w3.org/TR/xmlschema-2/#dateTime)。例如 `2007-01-31T23:59:59Z`。尽管没有强制要求，但还是建议您使用协调世界时（格林威治标准时间）时区提供该对象。 **注意** 确保您的服务器时间设置正确。如果您指定时间戳（而不是过期），则请求将在指定时间后 15 分钟自动过期（ AWS 服务器上AWS 不会处理时间戳比当前时间早于 15 分钟的请求）。如果使用 .NET，则不得发送过于具体的时间戳（因为对如何降低额外时间精度的解释不同）。在这种情况下，应手动构造精度不超过 1 毫秒的 `dateTime` 对象。 ## 第 1 部分：来自用户的请求以下是使用 HMAC-SHA 请求签名对 AWS 请求进行身份验证时必须遵循的流程。 ![\[描述用户请求的图表。\]](http://docs.aws.amazon.com/zh_cn/AWSSimpleQueueService/latest/SQSDeveloperGuide/images/sqs-basic-authentication-process-hmac-user.png) 1. 构造对的请求 AWS。 1. 使用您的秘密访问密钥计算密钥哈希消息验证码 (HMAC-SHA) 签名。 1. 在请求中包含签名和您的访问密钥 ID，然后将请求发送到 AWS。 ## 第 2 部分：来自的回应 AWS AWS 开始以下过程作为响应。 ![\[描述来自 AWS的响应的图表。\]](http://docs.aws.amazon.com/zh_cn/AWSSimpleQueueService/latest/SQSDeveloperGuide/images/sqs-basic-authentication-process-hmac-aws.png) 1. AWS 使用访问密钥 ID 来查找您的私有访问密钥。 1. AWS 使用与计算请求中发送的签名相同的算法，根据请求数据和私有访问密钥生成签名。 1. 将出现以下情况之一： + 如果 AWS 生成的签名与您在请求中发送的签名相匹配，则 AWS 认为该请求是真实的。 + 如果比较失败，则该请求将被丢弃，并 AWS 返回错误。 # Amazon SQS 批处理操作 Amazon SQS 支持批处理操作，帮助您降低成本，并通过单次操作处理最多 10 条消息。这些批处理操作包括： + `[SendMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessageBatch.html)` + `[DeleteMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteMessageBatch.html)` + `[ChangeMessageVisibilityBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ChangeMessageVisibilityBatch.html)` 使用批处理操作，您可以在一次 API 调用中执行多个操作，这有助于优化性能并降低成本。您可以使用查询 API 或任何支持 Amazon SQS 批处理操作的 AWS 软件开发工具包来利用批处理功能。 **重要详细信息** + **消息大小限制：**在一次 `SendMessageBatch` 调用中发送的所有消息的总大小不能超过 1048576 字节（1 MiB）。 + **权限：**您无法显式设置 `SendMessageBatch`、`DeleteMessageBatch` 或 `ChangeMessageVisibilityBatch` 的权限。相反，设置 `SendMessage`、`DeleteMessage` 或 `ChangeMessageVisibility` 操作的权限会同时设置其对应的批处理版本的权限。 + **控制台支持：**Amazon SQS 控制台不支持批处理操作。您必须使用查询 API 或 S AWS DK 来执行批量操作。 ## 批处理消息操作为了进一步优化成本和效率，请考虑以下批处理消息操作的最佳实践： + **批处理 API 操作：**使用[ Amazon SQS 批处理 API 操作](#sqs-batch-api-actions)来发送、接收和删除消息，并通过单次操作更改多条消息的可见性超时。这样做可以减少 API 调用的次数和相关成本。 + **客户端缓冲和长轮询数：**将长轮询与适用于 Java 的 AWS SDK中的[缓冲异步客户端](sqs-client-side-buffering-request-batching.md)一起使用，从而将客户端缓冲与请求批处理功能相结合。这种方法有助于最大限度地减少请求数量并优化对大量消息的处理。 **注意** Amazon SQS 缓冲异步客户端目前不支持 FIFO 队列。 # 启用客户端缓冲和请求批处理功能，并将其与 Amazon SQS 结合使用 [适用于 Java 的 AWS SDK](https://aws.amazon.com/sdkforjava/) 包括可访问 Amazon SQS 的 `AmazonSQSBufferedAsyncClient`。该客户端支持使用客户端缓冲进行简单的请求批处理。客户端进行的调用首先被缓冲，然后作为批处理请求发送到 Amazon SQS。客户端缓冲最多允许缓冲 10 个请求并将这些请求作为一个批处理请求发送，从而减少使用 Amazon SQS 的成本并减少发送的请求数。`AmazonSQSBufferedAsyncClient` 会缓冲同步和异步调用。批量请求和对[长轮询](sqs-short-and-long-polling.md)的支持还有助于提高吞吐量。有关更多信息，请参阅 [利用水平扩缩和操作批处理，借助 Amazon SQS 来提高吞吐量](sqs-throughput-horizontal-scaling-and-batching.md)。由于 `AmazonSQSBufferedAsyncClient` 实施与 `AmazonSQSAsyncClient` 相同的接口，因此从 `AmazonSQSAsyncClient` 迁移到 `AmazonSQSBufferedAsyncClient` 通常只需要对现有代码进行少量的更改。 **注意** Amazon SQS 缓冲异步客户端目前不支持 FIFO 队列。 ## 使用亚马逊 SQSBuffered AsyncClient 在开始之前，请完成 [设置 Amazon SQS](sqs-setting-up.md) 中的步骤。 ### AWS 适用于 Java 的 SDK 1.x 对于 AWS 适用于 Java 的 SDK 1.x，你可以`AmazonSQSBufferedAsyncClient`根据以下示例创建一个新的： ``` // Create the basic Amazon SQS async client final AmazonSQSAsync sqsAsync = new AmazonSQSAsyncClient(); // Create the buffered client final AmazonSQSAsync bufferedSqs = new AmazonSQSBufferedAsyncClient(sqsAsync); ``` 在创建新的 `AmazonSQSBufferedAsyncClient` 之后，您可以使用它将多个请求发送到 Amazon SQS（就像使用 `AmazonSQSAsyncClient` 所做的那样），例如： ``` final CreateQueueRequest createRequest = new CreateQueueRequest().withQueueName("MyQueue"); final CreateQueueResult res = bufferedSqs.createQueue(createRequest); final SendMessageRequest request = new SendMessageRequest(); final String body = "Your message text" + System.currentTimeMillis(); request.setMessageBody( body ); request.setQueueUrl(res.getQueueUrl()); final Future sendResult = bufferedSqs.sendMessageAsync(request); final ReceiveMessageRequest receiveRq = new ReceiveMessageRequest() .withMaxNumberOfMessages(1) .withQueueUrl(queueUrl); final ReceiveMessageResult rx = bufferedSqs.receiveMessage(receiveRq); ``` ### 配置亚马逊 SQSBuffered AsyncClient `AmazonSQSBufferedAsyncClient` 预配置了适用于大多数使用案例的设置。您可以进一步配置 `AmazonSQSBufferedAsyncClient`，例如： 1. 使用必需的配置参数来创建 `QueueBufferConfig` 类的实例。 1. 将该实例提供给 `AmazonSQSBufferedAsyncClient` 构造函数。 ``` // Create the basic Amazon SQS async client final AmazonSQSAsync sqsAsync = new AmazonSQSAsyncClient(); final QueueBufferConfig config = new QueueBufferConfig() .withMaxInflightReceiveBatches(5) .withMaxDoneReceiveBatches(15); // Create the buffered client final AmazonSQSAsync bufferedSqs = new AmazonSQSBufferedAsyncClient(sqsAsync, config); ``` **QueueBufferConfig 配置参数** | 参数 | 默认值 | 说明 | | --- | --- | --- | | longPoll | true | 如果 `longPoll` 设置为 `true`，`AmazonSQSBufferedAsyncClient` 会在使用消息时尝试使用长轮询。 | | longPollWaitTimeoutSeconds | 20 秒 | 在返回空接收结果前，`ReceiveMessage` 调用在服务器上阻塞以等待消息显示在队列中的最长时间（以秒为单位）。如果禁用长轮询，则此设置不起作用。 | | maxBatchOpenMs | 200 毫秒 | 传出调用等待其他要一起对同类型的消息进行批处理的调用的最长时间（以毫秒为单位）。设置的时间越长，则执行等量工作所需的批处理次数就越少（但是，批处理中的首次调用必须等待更长的时间）。如果将此参数设置为 `0`，则提交的请求不会等待其他请求，从而有效地禁用批处理。 | | maxBatchSize | 每批 10 个请求 | 在一个请求中一起进行批处理的消息的最大数量。该设置越大，则执行等量请求所需的批处理就越少。 Amazon SQS 允许的最大值为每批 10 个请求。 | | maxBatchSizeBytes | 1 MiB | 客户端尝试向 Amazon SQS 发送的消息批处理的最大大小（以字节为单位）。 Amazon SQS 允许的最大值为 1 MiB。 | | maxDoneReceiveBatches | 10 个批处理 | `AmazonSQSBufferedAsyncClient` 在客户端预取和存储的接收批处理的最大数量。设置的值越高，则可满足越多的接收请求而不必调用 Amazon SQS（但是，预取的消息越多，则消息在缓冲区中停留的时间就越长，从而导致它们的可见性超时过期）。 `0` 表示所有消息预取操作将被禁用，消息只能按需使用。 | | maxInflightOutboundBatches | 5 个批处理 | 可以同时处理的最大活跃出站批处理数量。设置的值越高，发送出站批处理的速度就越快（受限于其他配额，例如 CPU 或带宽），并且 `AmazonSQSBufferedAsyncClient` 使用的线程就越多。 | | maxInflightReceiveBatches | 10 个批处理 | 可以同时处理的最大活跃接收批处理数量。设置的值越高，可接收的消息就越多（受限于其他配额，例如 CPU 或带宽），并且 `AmazonSQSBufferedAsyncClient` 使用的线程就越多。 `0` 表示所有消息预取操作将被禁用，消息只能按需使用。 | | visibilityTimeoutSeconds | –1 | 如果此参数设置为正值（非零值），则此处设置的可见性超时将覆盖在使用的消息所在的队列上设置的可见性超时。 `-1` 表示为队列选择默认设置。不能将可见性超时设置为 `0`。 | ### AWS 适用于 Java 的 SDK 2.x 对于 AWS 适用于 Java 的 SDK 2.x，你可以`SqsAsyncBatchManager`根据以下示例创建一个新的： ``` // Create the basic Sqs Async Client SqsAsyncClient sqs = SqsAsyncClient.builder() .region(Region.US_EAST_1) .build(); // Create the batch manager SqsAsyncBatchManager sqsAsyncBatchManager = sqs.batchManager(); ``` 在创建新的 `SqsAsyncBatchManager` 之后，您可以使用它将多个请求发送到 Amazon SQS（就像使用 `SqsAsyncClient` 所做的那样），例如： ``` final String queueName = "MyAsyncBufferedQueue" + UUID.randomUUID(); final CreateQueueRequest request = CreateQueueRequest.builder().queueName(queueName).build(); final String queueUrl = sqs.createQueue(request).join().queueUrl(); System.out.println("Queue created: " + queueUrl); // Send messages CompletableFuture sendMessageFuture; for (int i = 0; i < 10; i++) { final int index = i; sendMessageFuture = sqsAsyncBatchManager.sendMessage( r -> r.messageBody("Message " + index).queueUrl(queueUrl)); SendMessageResponse response= sendMessageFuture.join(); System.out.println("Message " + response.messageId() + " sent!"); } // Receive messages with customized configurations CompletableFuture receiveResponseFuture = customizedBatchManager.receiveMessage( r -> r.queueUrl(queueUrl) .waitTimeSeconds(10) .visibilityTimeout(20) .maxNumberOfMessages(10) ); System.out.println("You have received " + receiveResponseFuture.join().messages().size() + " messages in total."); // Delete messages DeleteQueueRequest deleteQueueRequest = DeleteQueueRequest.builder().queueUrl(queueUrl).build(); int code = sqs.deleteQueue(deleteQueueRequest).join().sdkHttpResponse().statusCode(); System.out.println("Queue is deleted, with statusCode " + code); ``` ### 正在配置 SqsAsyncBatchManager `SqsAsyncBatchManager` 预配置了适用于大多数使用案例的设置。您可以进一步配置 `SqsAsyncBatchManager`，例如：通过 `SqsAsyncBatchManager.Builder` 创建自定义配置： ``` SqsAsyncBatchManager customizedBatchManager = SqsAsyncBatchManager.builder() .client(sqs) .scheduledExecutor(Executors.newScheduledThreadPool(5)) .overrideConfiguration(b -> b .maxBatchSize(10) .sendRequestFrequency(Duration.ofMillis(200)) .receiveMessageMinWaitDuration(Duration.ofSeconds(10)) .receiveMessageVisibilityTimeout(Duration.ofSeconds(20)) .receiveMessageAttributeNames(Collections.singletonList("*")) .receiveMessageSystemAttributeNames(Collections.singletonList(MessageSystemAttributeName.ALL))) .build(); ``` **`BatchOverrideConfiguration` 参数** | 参数 | 默认值 | 说明 | | --- | --- | --- | | maxBatchSize | 每批 10 个请求 | 在一个请求中一起进行批处理的消息的最大数量。该设置越大，则执行等量请求所需的批处理就越少。 Amazon SQS 允许的最大值为每批次 10 个请求。 | | sendRequestFrequency | 200 毫秒 | 传出调用等待其他要一起对同类型的消息进行批处理的调用的最长时间（以毫秒为单位）。设置的时间越长，则执行等量工作所需的批处理次数就越少（但是，批处理中的首次调用必须等待更长的时间）。如果将此参数设置为 `0`，则提交的请求不会等待其他请求，从而有效地禁用批处理。 | | receiveMessageVisibilityTimeout | –1 | 如果此参数设置为正值（非零值），则此处设置的可见性超时将覆盖在使用的消息所在的队列上设置的可见性超时。 `1` 表示为队列选择默认设置。不能将可见性超时设置为 `0`。 | | receiveMessageMinWaitDuration | 50 毫秒 | `receiveMessage` 调用等待可用消息被获取的最短时间（以毫秒为单位）。设置的值越高，则执行等量请求所需的批量操作就越少。 | # 利用水平扩缩和操作批处理，借助 Amazon SQS 来提高吞吐量 Amazon SQS 支持高吞吐量消息传递。有关吞吐量限制的详细信息，请参阅 [Amazon SQS 消息配额](quotas-messages.md)。要最大限度提高吞吐量： + 通过增加创建者和使用者的实例数量，实现水平[扩展](#horizontal-scaling)。 + 使用[操作批处理](#request-batching)在单次请求中发送或接收多条消息，从而减少 API 调用开销。 ## 横向扩展由于您通过 HTTP 请求-响应协议来访问 Amazon SQS，因此，*请求延迟*（启动请求和接收响应之间的时间间隔）会限制您可以通过单一连接利用单一线程达到的吞吐量。例如，如果从基于 Amazon EC2 的客户端到同一区域内的 Amazon SQS 的延迟时间平均为 20 毫秒，则通过单一连接利用单一线程达到的最大吞吐量平均为 50 TPS。 *水平扩展* 涉及到增加消息创建者（发出 `[SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html)` 请求）和使用者（发出 `[ReceiveMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html)` 和 `[DeleteMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteMessage.html)` 请求）的数量，以提高整个队列的吞吐量。可以通过三种方式进行水平扩展： + 增加每个客户端的线程数量 + 添加更多客户端 + 增加每个客户端的线程数量并添加更多客户端在添加更多客户端后，基本上可以实现队列吞吐量的线性增长。例如，如果将客户端数量翻倍，吞吐量也会翻倍。 ## 操作批处理 *批处理* 可在与服务的每次往返操作中执行更多的工作（例如，当您通过单个 `SendMessageBatch` 请求发送多条消息时）。Amazon SQS 批处理操作包括 `[SendMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessageBatch.html)`、`[DeleteMessageBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteMessageBatch.html)` 和 `[ChangeMessageVisibilityBatch](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ChangeMessageVisibilityBatch.html)`。要在不更改创建者或使用者的情况下利用批处理，您可以使用 [Amazon SQS 缓冲异步客户端](sqs-client-side-buffering-request-batching.md)。 **注意** 由于 `[ReceiveMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html)` 一次可以处理 10 条消息，因此没有 `ReceiveMessageBatch` 操作。批处理会在一个批处理请求中的多条消息之间分配批处理操作的延迟时间，而不是接受单一消息（例如，`[SendMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SendMessage.html)` 请求）的整个延迟时间。由于每次往返操作都会执行更多工作，因此，批处理请求可以更高效地使用线程和连接，从而提高吞吐量。可以将批处理与水平扩展结合使用来提供吞吐量，所需的线程、连接和请求的数量比单独的消息请求所需的数量更少。您可以使用 Amazon SQS 批处理操作一次性发送、接收或删除多达 10 条消息。由于 Amazon SQS 按请求收费，因此，批处理可以大幅降低您的成本。批处理会为您的应用程序带来一些复杂性（例如，您的应用程序必须先积累消息然后才能发送，或者有时候必须花费较长的时间等待响应）。但是，批处理在以下情况下仍然会很有效： + 您的应用程序在短时间内生成很多消息，因此，延迟时间从来不会很长。 + 消息使用者从队列中自行获取消息，这与需要发送消息来响应其无法控制的事件的典型消息创建者不同。 **重要** 即使批处理中的个别消息失败了，批处理请求也可能会成功。发出批处理请求后，始终检查各条消息是否失败，并在必要时重试操作。 ## 单一操作和批处理请求的有效 Java 示例 ### 先决条件将 `aws-java-sdk-sqs.jar`、`aws-java-sdk-ec2.jar` 和 `commons-logging.jar` 程序包添加到 Java 生成类路径中。以下示例说明了 Maven 项目的 `pom.xml` 文件中的这些依赖关系。 ``` com.amazonaws aws-java-sdk-sqs LATEST com.amazonaws aws-java-sdk-ec2 LATEST commons-logging commons-logging LATEST ``` ### SimpleProducerConsumer.java 以下 Java 代码示例将实施一个简单的创建者-使用者模式。主线程会生成大量创建者和使用者线程，这些线程会在指定时间内处理 1KB 消息。此示例包括发出单一操作请求的创建者和使用者，以及发出批处理请求的创建者和使用者。 ``` /* * Copyright 2010-2024 Amazon.com, Inc. or its affiliates. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"). * You may not use this file except in compliance with the License. * A copy of the License is located at * * https://aws.amazon.com/apache2.0 * * or in the "license" file accompanying this file. This file is distributed * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either * express or implied. See the License for the specific language governing * permissions and limitations under the License. * */ import com.amazonaws.AmazonClientException; import com.amazonaws.ClientConfiguration; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.*; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import java.math.BigInteger; import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.Scanner; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; /** * Start a specified number of producer and consumer threads, and produce-consume * for the least of the specified duration and 1 hour. Some messages can be left * in the queue because producers and consumers might not be in exact balance. */ public class SimpleProducerConsumer { // The maximum runtime of the program. private final static int MAX_RUNTIME_MINUTES = 60; private final static Log log = LogFactory.getLog(SimpleProducerConsumer.class); public static void main(String[] args) throws InterruptedException { final Scanner input = new Scanner(System.in); System.out.print("Enter the queue name: "); final String queueName = input.nextLine(); System.out.print("Enter the number of producers: "); final int producerCount = input.nextInt(); System.out.print("Enter the number of consumers: "); final int consumerCount = input.nextInt(); System.out.print("Enter the number of messages per batch: "); final int batchSize = input.nextInt(); System.out.print("Enter the message size in bytes: "); final int messageSizeByte = input.nextInt(); System.out.print("Enter the run time in minutes: "); final int runTimeMinutes = input.nextInt(); /* * Create a new instance of the builder with all defaults (credentials * and region) set automatically. For more information, see Creating * Service Clients in the AWS SDK for Java Developer Guide. */ final ClientConfiguration clientConfiguration = new ClientConfiguration() .withMaxConnections(producerCount + consumerCount); final AmazonSQS sqsClient = AmazonSQSClientBuilder.standard() .withClientConfiguration(clientConfiguration) .build(); final String queueUrl = sqsClient .getQueueUrl(new GetQueueUrlRequest(queueName)).getQueueUrl(); // The flag used to stop producer, consumer, and monitor threads. final AtomicBoolean stop = new AtomicBoolean(false); // Start the producers. final AtomicInteger producedCount = new AtomicInteger(); final Thread[] producers = new Thread[producerCount]; for (int i = 0; i < producerCount; i++) { if (batchSize == 1) { producers[i] = new Producer(sqsClient, queueUrl, messageSizeByte, producedCount, stop); } else { producers[i] = new BatchProducer(sqsClient, queueUrl, batchSize, messageSizeByte, producedCount, stop); } producers[i].start(); } // Start the consumers. final AtomicInteger consumedCount = new AtomicInteger(); final Thread[] consumers = new Thread[consumerCount]; for (int i = 0; i < consumerCount; i++) { if (batchSize == 1) { consumers[i] = new Consumer(sqsClient, queueUrl, consumedCount, stop); } else { consumers[i] = new BatchConsumer(sqsClient, queueUrl, batchSize, consumedCount, stop); } consumers[i].start(); } // Start the monitor thread. final Thread monitor = new Monitor(producedCount, consumedCount, stop); monitor.start(); // Wait for the specified amount of time then stop. Thread.sleep(TimeUnit.MINUTES.toMillis(Math.min(runTimeMinutes, MAX_RUNTIME_MINUTES))); stop.set(true); // Join all threads. for (int i = 0; i < producerCount; i++) { producers[i].join(); } for (int i = 0; i < consumerCount; i++) { consumers[i].join(); } monitor.interrupt(); monitor.join(); } private static String makeRandomString(int sizeByte) { final byte[] bs = new byte[(int) Math.ceil(sizeByte * 5 / 8)]; new Random().nextBytes(bs); bs[0] = (byte) ((bs[0] | 64) & 127); return new BigInteger(bs).toString(32); } /** * The producer thread uses {@code SendMessage} * to send messages until it is stopped. */ private static class Producer extends Thread { final AmazonSQS sqsClient; final String queueUrl; final AtomicInteger producedCount; final AtomicBoolean stop; final String theMessage; Producer(AmazonSQS sqsQueueBuffer, String queueUrl, int messageSizeByte, AtomicInteger producedCount, AtomicBoolean stop) { this.sqsClient = sqsQueueBuffer; this.queueUrl = queueUrl; this.producedCount = producedCount; this.stop = stop; this.theMessage = makeRandomString(messageSizeByte); } /* * The producedCount object tracks the number of messages produced by * all producer threads. If there is an error, the program exits the * run() method. */ public void run() { try { while (!stop.get()) { sqsClient.sendMessage(new SendMessageRequest(queueUrl, theMessage)); producedCount.incrementAndGet(); } } catch (AmazonClientException e) { /* * By default, AmazonSQSClient retries calls 3 times before * failing. If this unlikely condition occurs, stop. */ log.error("Producer: " + e.getMessage()); System.exit(1); } } } /** * The producer thread uses {@code SendMessageBatch} * to send messages until it is stopped. */ private static class BatchProducer extends Thread { final AmazonSQS sqsClient; final String queueUrl; final int batchSize; final AtomicInteger producedCount; final AtomicBoolean stop; final String theMessage; BatchProducer(AmazonSQS sqsQueueBuffer, String queueUrl, int batchSize, int messageSizeByte, AtomicInteger producedCount, AtomicBoolean stop) { this.sqsClient = sqsQueueBuffer; this.queueUrl = queueUrl; this.batchSize = batchSize; this.producedCount = producedCount; this.stop = stop; this.theMessage = makeRandomString(messageSizeByte); } public void run() { try { while (!stop.get()) { final SendMessageBatchRequest batchRequest = new SendMessageBatchRequest().withQueueUrl(queueUrl); final List entries = new ArrayList(); for (int i = 0; i < batchSize; i++) entries.add(new SendMessageBatchRequestEntry() .withId(Integer.toString(i)) .withMessageBody(theMessage)); batchRequest.setEntries(entries); final SendMessageBatchResult batchResult = sqsClient.sendMessageBatch(batchRequest); producedCount.addAndGet(batchResult.getSuccessful().size()); /* * Because SendMessageBatch can return successfully, but * individual batch items fail, retry the failed batch items. */ if (!batchResult.getFailed().isEmpty()) { log.warn("Producer: retrying sending " + batchResult.getFailed().size() + " messages"); for (int i = 0, n = batchResult.getFailed().size(); i < n; i++) { sqsClient.sendMessage(new SendMessageRequest(queueUrl, theMessage)); producedCount.incrementAndGet(); } } } } catch (AmazonClientException e) { /* * By default, AmazonSQSClient retries calls 3 times before * failing. If this unlikely condition occurs, stop. */ log.error("BatchProducer: " + e.getMessage()); System.exit(1); } } } /** * The consumer thread uses {@code ReceiveMessage} and {@code DeleteMessage} * to consume messages until it is stopped. */ private static class Consumer extends Thread { final AmazonSQS sqsClient; final String queueUrl; final AtomicInteger consumedCount; final AtomicBoolean stop; Consumer(AmazonSQS sqsClient, String queueUrl, AtomicInteger consumedCount, AtomicBoolean stop) { this.sqsClient = sqsClient; this.queueUrl = queueUrl; this.consumedCount = consumedCount; this.stop = stop; } /* * Each consumer thread receives and deletes messages until the main * thread stops the consumer thread. The consumedCount object tracks the * number of messages that are consumed by all consumer threads, and the * count is logged periodically. */ public void run() { try { while (!stop.get()) { try { final ReceiveMessageResult result = sqsClient .receiveMessage(new ReceiveMessageRequest(queueUrl)); if (!result.getMessages().isEmpty()) { final Message m = result.getMessages().get(0); sqsClient.deleteMessage(new DeleteMessageRequest(queueUrl, m.getReceiptHandle())); consumedCount.incrementAndGet(); } } catch (AmazonClientException e) { log.error(e.getMessage()); } } } catch (AmazonClientException e) { /* * By default, AmazonSQSClient retries calls 3 times before * failing. If this unlikely condition occurs, stop. */ log.error("Consumer: " + e.getMessage()); System.exit(1); } } } /** * The consumer thread uses {@code ReceiveMessage} and {@code * DeleteMessageBatch} to consume messages until it is stopped. */ private static class BatchConsumer extends Thread { final AmazonSQS sqsClient; final String queueUrl; final int batchSize; final AtomicInteger consumedCount; final AtomicBoolean stop; BatchConsumer(AmazonSQS sqsClient, String queueUrl, int batchSize, AtomicInteger consumedCount, AtomicBoolean stop) { this.sqsClient = sqsClient; this.queueUrl = queueUrl; this.batchSize = batchSize; this.consumedCount = consumedCount; this.stop = stop; } public void run() { try { while (!stop.get()) { final ReceiveMessageResult result = sqsClient .receiveMessage(new ReceiveMessageRequest(queueUrl) .withMaxNumberOfMessages(batchSize)); if (!result.getMessages().isEmpty()) { final List messages = result.getMessages(); final DeleteMessageBatchRequest batchRequest = new DeleteMessageBatchRequest() .withQueueUrl(queueUrl); final List entries = new ArrayList(); for (int i = 0, n = messages.size(); i < n; i++) entries.add(new DeleteMessageBatchRequestEntry() .withId(Integer.toString(i)) .withReceiptHandle(messages.get(i) .getReceiptHandle())); batchRequest.setEntries(entries); final DeleteMessageBatchResult batchResult = sqsClient .deleteMessageBatch(batchRequest); consumedCount.addAndGet(batchResult.getSuccessful().size()); /* * Because DeleteMessageBatch can return successfully, * but individual batch items fail, retry the failed * batch items. */ if (!batchResult.getFailed().isEmpty()) { final int n = batchResult.getFailed().size(); log.warn("Producer: retrying deleting " + n + " messages"); for (BatchResultErrorEntry e : batchResult .getFailed()) { sqsClient.deleteMessage( new DeleteMessageRequest(queueUrl, messages.get(Integer .parseInt(e.getId())) .getReceiptHandle())); consumedCount.incrementAndGet(); } } } } } catch (AmazonClientException e) { /* * By default, AmazonSQSClient retries calls 3 times before * failing. If this unlikely condition occurs, stop. */ log.error("BatchConsumer: " + e.getMessage()); System.exit(1); } } } /** * This thread prints every second the number of messages produced and * consumed so far. */ private static class Monitor extends Thread { private final AtomicInteger producedCount; private final AtomicInteger consumedCount; private final AtomicBoolean stop; Monitor(AtomicInteger producedCount, AtomicInteger consumedCount, AtomicBoolean stop) { this.producedCount = producedCount; this.consumedCount = consumedCount; this.stop = stop; } public void run() { try { while (!stop.get()) { Thread.sleep(1000); log.info("produced messages = " + producedCount.get() + ", consumed messages = " + consumedCount.get()); } } catch (InterruptedException e) { // Allow the thread to exit. } } } } ``` ### 监控示例运行中的数量指标 Amazon SQS 会针对发送、接收和删除的消息自动生成容量指标。可以通过队列的**监控**选项卡或者通过 [CloudWatch 控制台](https://console.aws.amazon.com/cloudwatch/home)访问上述指标和其他指标。 **注意** 队列启动后，这些指标最多可能需要花费 15 分钟才可用。 # 将 Amazon SQS 与 AWS SDK 结合使用 AWS 软件开发工具包（SDK）适用于许多常用编程语言。每个软件开发工具包都提供 API、代码示例和文档，使开发人员能够更轻松地以其首选语言构建应用程序。 | SDK 文档 | 代码示例 | | --- | --- | | [适用于 C\$1\$1 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-cpp) | [适用于 C\$1\$1 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/cpp) | | [AWS CLI](https://docs.aws.amazon.com/cli) | [AWS CLI 代码示例](https://docs.aws.amazon.com/code-library/latest/ug/cli_2_code_examples.html) | | [适用于 Go 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-go) | [适用于 Go 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/gov2) | | [适用于 Java 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-java) | [适用于 Java 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2) | | [适用于 JavaScript 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-javascript) | [适用于 JavaScript 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3) | | [适用于 Kotlin 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-kotlin) | [适用于 Kotlin 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/kotlin) | | [适用于 .NET 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-net) | [适用于 .NET 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3) | | [适用于 PHP 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-php) | [适用于 PHP 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/php) | | [AWS Tools for PowerShell](https://docs.aws.amazon.com/powershell) | [AWS Tools for PowerShell 代码示例](https://docs.aws.amazon.com/code-library/latest/ug/powershell_5_code_examples.html) | | [适用于 Python (Boto3) 的 AWS SDK](https://docs.aws.amazon.com/pythonsdk) | [适用于 Python (Boto3) 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python) | | [适用于 Ruby 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-ruby) | [适用于 Ruby 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/ruby) | | [适用于 Rust 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-rust) | [适用于 Rust 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1) | | [适用于 SAP ABAP 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-sapabap) | [适用于 SAP ABAP 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap) | | [适用于 Swift 的 AWS SDK](https://docs.aws.amazon.com/sdk-for-swift) | [适用于 Swift 的 AWS SDK 代码示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/swift) | **示例可用性** 找不到所需的内容？通过使用此页面底部的**提供反馈**链接请求代码示例。