本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 使用 Amazon SNS FIFO 主题的消息排序和重复数据删除策略 本主题提供有关 Amazon SNS FIFO（先进先出）主题的特征和功能以及它们如何与 [Amazon SQS FIFO 队列](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html)集成的信息。您将学习如何将这些服务结合使用，以确保严格的消息排序和重复数据删除，这对于需要数据一致性的应用程序来说至关重要。本内容涵盖了 Amazon SNS FIFO 主题有益的特定用例，提供了对消息顺序和唯一性至关重要的场景的见解。 您还将了解消息排序、消息分组的技术细节，以及它们如何影响消息传送。消息重复数据删除主题说明了防止重复消息的机制，确保每条消息只处理一次。此外，您还将了解消息筛选、安全性和耐久性，这些对于维护消息收发系统的完整性和可靠性非常重要。本内容还包括有关消息归档和重播的信息，提供了管理消息历史记录的策略。还提供了实用的代码示例，协助您在自己的应用程序中实现这些特征，让您亲身体验 Amazon SNS FIFO 主题及其与 Amazon SQS FIFO 队列的集成。 # Amazon SNS 中的高吞吐量 FIFO 主题 Amazon SNS 中的高吞吐量 FIFO 主题可以有效地管理高消息吞吐量，同时保持严格的消息顺序，从而确保处理大量消息的应用程序的可靠性和可扩展性。该解决方案非常适合要求高吞吐量和有序消息传递的场景。为了提高使用高吞吐量 FIFO 主题的消息吞吐量，建议增加消息组的数量。有关高吞吐量消息配额的更多信息，请参阅《Amazon Web Services 一般参考》**中的 [Amazon SNS service quotas](https://docs.aws.amazon.com/general/latest/gr/sns.html)。 ## Amazon SNS FIFO 主题的高吞吐量使用案例 以下使用案例重点介绍了高吞吐量 FIFO 队列的不同应用，展示了它们在各行业和场景中的有效性： + **实时数据处理**：处理实时数据流（例如事件处理或遥测数据摄取）的应用程序可以利用高吞吐量 FIFO 主题来处理不断涌入的消息，并在保证消息顺序的同时，进行准确分析。 + **电子商务订单处理**：在电子商务平台中，保持客户交易顺序至关重要，高吞吐量 FIFO 主题可以确保订单按顺序交付并且不会出现延迟，即使在购物高峰期也是如此。 + **金融服务**：金融机构在处理高频交易或交易数据时，依靠高吞吐量 FIFO 主题以最小的延迟处理市场数据和交易，同时遵守严格的消息顺序监管要求。 + **媒体流式传输**：流媒体平台和媒体分发服务利用高吞吐量 FIFO 主题来管理媒体文件和流媒体内容的分发，确保用户获得流畅的播放体验，同时保持正确的内容分发顺序。 ## Amazon SNS FIFO 主题高吞吐量的分区和数据分发 对于高吞吐量主题，Amazon SNS 可以跨分区分发 FIFO 主题数据。分区是为主题分配的容量，可在 AWS 区域内的多个可用区中自动进行复制。您无需管理分区。相反，Amazon SNS 会根据入口速率自动代表您管理分区。 对于 FIFO 主题，在以下情况下，Amazon SNS 会修改主题中的分区数量： + 如果当前发布速率接近或超过现有分区所能支持的速率，则会分配其他分区，直到主题达到区域配额。有关更多信息，请参阅《Amazon Web Services 一般参考》**中的 [Amazon SNS service quotas](https://docs.aws.amazon.com/general/latest/gr/sns.html)。 + 如果当前分区的利用率较低，则分区的数量可能会减少。 分区管理在后台自动进行，对程序是透明的。您的主题和消息始终可用。 **注意** 如果您突然大幅增加主题的流量，同时发送量是平时的数倍，则可能会出现临时[发布](https://docs.aws.amazon.com/sns/latest/api/API_Publish.html) API 节流。这种节流在重复数据删除窗口期可能会一直持续，而主题则可以纵向扩展以适应增加的流量。 ## 按消息组分发数据 IDs 将消息发布到 FIFO 主题时，Amazon SNS 使用每条消息的消息组 ID 的值作为内部哈希函数的输入。哈希函数的输出值决定了哪个分区处理消息，一个或多个消息组 IDs 可以由给定分区处理。 **注意** Amazon SNS 经过优化，可以在 FIFO 主题的分区中均匀分配项目，无论分区数量如何。 AWS 建议您使用可 IDs 包含大量不同值的消息组。 ## 在 Amazon SNS FIFO 主题上启用高吞吐量 默认情况下，Amazon SNS FIFO 主题配置为主题级重复数据删除，这由 [https://docs.aws.amazon.com/sns/latest/api/API_CreateTopic.html](https://docs.aws.amazon.com/sns/latest/api/API_CreateTopic.html) 设置为 `Topic` 的主题属性控制，吞吐量配额更受限制，请参阅《Amazon Web Services 一般参考》中的 [Amazon SNS service quotas](https://docs.aws.amazon.com/general/latest/gr/sns.html)。** 要为 Amazon SNS FIFO 主题启用高吞吐量，请将 `FifoThroughputScope` 属性更新为 `MessageGroup`。此更改可以通过控制台或使用 AWS CLI 和 SDK 完成，也可以在主题创建期间进行设置，Amazon SNS 建议这样做，以获得最佳的客户体验并减少主题被限制的机会。 **重要** 将主题的 `FifoThroughputScope` 更改为 `MessageGroup` 后，就无法将其恢复为 `Topic` 吞吐量。 ## 为任何已订阅的 Amazon SQS FIFO 队列启用高吞吐量模式 向启用了高吞吐量的 Amazon SNS FIFO 主题发布消息并且订阅了一个或多个 Amazon SQS FIFO 队列时，建议您在 Amazon SQS FIFO 队列上启用高吞吐量，以便 Amazon SNS FIFO 高吞吐量主题顺畅传输。有关更多信息，请参阅《Amazon Simple Queue Service 开发人员指南》**中的 [FIFO 队列的高吞吐量](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/high-throughput-fifo.html)。 # Amazon SNS FIFO 主题示例使用案例 以下示例介绍了一个由汽车零部件制造商使用 Amazon SNS FIFO 主题和 Amazon SQS 队列构建的电子商务平台。该平台包含四个无服务器应用程序： + 库存经理使用价格管理应用程序为每件存货设置价格。在该公司，产品价格可能会因汇率波动、市场需求、销售策略的变化而变化。价格管理应用程序使用 AWS Lambda 函数，它会在价格变化时将价格更新发布到 Amazon SNS FIFO 主题。 + 批发应用程序为汽车车身修理厂和汽车制造商可以在其中批量购买公司汽车零部件的网站提供后端服务。为了获取价格变化通知，批发应用程序为其 Amazon SQS FIFO 队列订阅价格管理应用程序的 Amazon SNS FIFO 主题。 + 零售应用程序为另一个网站提供后端，车主和汽车改装爱好者可以通过该网站为他们的车辆购买单独的汽车零部件。为了获取价格变化通知，零售应用程序也会为其 Amazon SQS FIFO 队列订阅价格管理应用程序的 Amazon SNS FIFO 主题。 + 一种分析应用程序，可汇总价格更新并将其存储到 Amazon S3 存储桶中，从而使 Amazon Athena 能够出于商业智能 (BI) 目的查询存储桶。为了获取价格变化通知，分析应用程序为其 Amazon SQS 标准队列订阅价格管理应用程序的 Amazon SNS FIFO 主题。与其他应用程序不同，分析应用程序不需要对价格更新进行严格排序。 ![\[汽车零部件制造商使用 Amazon SNS FIFO 主题和 Amazon SQS 队列创建的电子商务平台示例，展示了不同的无服务器应用程序（例如价格管理、批发、零售和分析）如何利用这些服务进行有序的消息传送和重复数据删除。此设置可确保批发和零售应用程序以正确的顺序接收价格更新，而分析应用程序以商业智能目的聚合数据，而无需严格排序消息。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-usecase.png) 为了使批发和零售应用程序以正确的顺序接收价格更新，价格管理应用程序必须使用严格排序的消息分发系统。使用 Amazon SNS FIFO 主题和 Amazon SQS FIFO 队列可以按顺序处理消息，而不会出现重复。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息排序详细信息](fifo-topic-message-ordering.md)。有关实现此使用案例的代码片段，请参阅 [适用于 FIFO 主题的 Amazon SNS 代码示例](fifo-topic-code-examples.md)。 # FIFO 主题的 Amazon SNS 消息排序详细信息 Amazon SNS FIFO 主题始终按照消息发布到主题的确切顺序，向订阅的 Amazon SQS 队列传送消息，并且只传送一次。订阅 Amazon SQS FIFO 队列后，该队列的使用者将按照消息传送到队列的确切顺序接收消息，且不会出现重复消息。但是，订阅了 Amazon SQS 标准队列后，该队列的使用者可能会多次收到乱序消息。这可以进一步将订阅者与发布者分开，从而在消息使用和成本优化方面为订阅者提供更大的灵活性，如基于 [Amazon SNS FIFO 主题示例使用案例](fifo-example-use-case.md)的下图所示。 ![\[Amazon SNS FIFO（先进先出）主题中的消息传送系统示例，重点介绍了如何以严格的顺序将消息一致地传递到 Amazon SQS FIFO 队列。这与 Amazon SQS 标准队列的行为形成鲜明对比，在标准队列中，消息可能不按顺序送达，而且不止一次。该示例展示了三种不同的订阅用户类型（分析功能、批发应用程序和零售应用程序），演示了每种订阅用户是如何根据他们订阅的队列类型按严格顺序或按最大努力顺序接收消息的。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-ordering-1.png) 请注意，不存在订阅者的隐含排序。以下示例显示消息 **m1** 首先传输给批发订阅者，然后传输给零售订阅者，再传输给分析订阅者。消息 **m2** 首先传输给零售订阅者，然后传输给批发订阅者，最后传输给分析订阅者。尽管这两条消息以不同的顺序传递给订阅者，但是每个 Amazon SQS FIFO 订阅者都会保留消息顺序。每个订阅者都与任何其他订阅者隔离感知。 ![\[示例：Amazon SNS FIFO 主题和各种类型的订阅用户（包括 Amazon SQS FIFO 和标准队列）如何处理消息的排序和传送。它显示消息被发布到主题并传送到不同类型的队列，确保 FIFO 队列的传送顺序和标准队列的最大努力排序。此设置支持电子商务平台中的场景，在这些场景中，不同的组件需要按特定顺序可靠地传送消息才能进行精确处理。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-ordering-2.png) 如果 Amazon SQS 队列订阅者无法访问，它可能会不同步。例如，假设批发应用程序队列所有者错误地更改了 [Amazon SQS 队列策略](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-using-identity-based-policies.html)，以防止 Amazon SNS 服务委托人将消息传递到队列。在这种情况下，将价格更新传输到批发队列失败，而零售和分析队列的价格更新成功，从而导致订阅者不同步。批发应用程序队列所有者更正队列策略后，Amazon SNS 将继续向订阅队列传递消息。在队列配置不正确时，发布到主题的任何消息都将被删除，除非对应的订阅已配置了[死信队列](sns-dead-letter-queues.md)。 ![\[Amazon SNS FIFO 设置中的消息传递行为示例，在该设置中，消息通过 Amazon SQS 队列发布到不同的订阅用户类型（批发、零售和分析）。它重点介绍了队列策略配置错误对不同订阅用户队列间消息传送同步的影响。该示例演示了批发订阅用户的消息传输为何因策略错误而失败，但零售和分析订阅用户的消息传输却继续成功，这强调了正确的队列配置对于保持同步数据交付的重要性。这种情况凸显了 FIFO 主题在正常情况下确保按顺序交付和精确一次交付的能力以及配置错误的后果。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-ordering-3.png) 您可以让多个应用程序（或同一应用程序中的多个线程）并行向 SNS FIFO 主题发布消息。执行此操作时，您可以有效地将消息排序委托给 Amazon SNS 服务。要确定已建立的消息序列，您可以检查序列号。 序列号是 Amazon SNS 为每条消息分配的大型、非连续的数字。序列号的长度为 128 位，并且每个[消息组](fifo-message-grouping.md)的序列号会继续增加。序列号作为消息正文的一部分传递给订阅的 Amazon SQS 队列。但是，如果启用[原信息传输](sns-large-payload-raw-message-delivery.md)，则传输到 Amazon SQS 队列的消息不包括序列号或任何其他 Amazon SNS 消息元数据。 ![\[多个 Lambda 函数的示例将消息发布到 Amazon SNS FIFO（先入先出）主题，然后该主题将这些消息传送到 Amazon SQS FIFO 队列，从而保持严格的消息处理顺序。此设置用于确保按照在应用程序的不同组件之间发送消息的确切顺序进行处理，序列号表示组中每条消息的顺序。这种类型的配置对于必须严格维护操作和消息顺序以确保一致性的应用程序至关重要。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-ordering-4.png) Amazon SNS FIFO 主题定义消息组上下文中的排序。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息分组](fifo-message-grouping.md)。 # FIFO 主题的 Amazon SNS 消息分组 属于同一组的消息按相对于组的严格顺序逐个处理。 向 Amazon SNS FIFO 主题发布消息时，需设置消息组 ID。组 ID 是指定消息属于特定消息组的强制令牌。SNS FIFO 主题将组 ID 传递给订阅的 Amazon SQS FIFO 队列。SNS FIFO 主题或 SQS FIFO 队列 IDs 中的群组数量没有限制。消息组 ID 不会传递给 Amazon SQS 标准队列。 消息组和订阅之间没有关联性。因此，发布到任何消息组的消息都会传输到所有已订阅队列，但须遵守附加到订阅的任何筛选策略。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息传输](fifo-message-delivery.md) 和 [FIFO 主题的 Amazon SNS 消息筛选](fifo-message-filtering.md)。 在[汽车零部件价格管理示例使用案例](fifo-example-use-case.md)中，平台中销售的每个产品都有一个专用的消息组。用同一个 Amazon SNS FIFO 主题处理所有价格更新。价格更新的顺序保留在单个汽车零部件产品的上下文中，但不是跨多个产品。下图演示了工作原理。请注意，对于消息组 ID 为 **product-214** 的产品，**m1** 消息始终在 **m4** 消息之前得到处理。此顺序将在整个工作流程中保留，这些工作流程使用 Amazon SNS FIFO 和 Amazon SQS FIFO。同样，对于消息组 ID 为 **product-799** 的产品，**m2** 消息始终在 **m3** 消息之前得到处理。但是，使用 Amazon SQS 标准队列时，将无法保证消息顺序，也不存在消息组。**product-214** 和 **product-799** 消息组彼此独立，因此它们的消息排序方式之间没有任何关系。 ![\[消息排序和重复数据删除在涉及 AWS 服务 不同和消息组的 Amazon SNS FIFO 主题场景中如何工作的示例。 IDs它显示了从 Lambda 函数通过 Amazon SNS FIFO 主题流向各种类型的 Amazon SQS 队列（FIFO 和标准）的消息流，在 FIFO 队列中保持严格的顺序，同时演示了标准队列中的潜在混乱。此设置用于强调消息排序在诸如电子商务平台价格更新之类的应用程序中的重要性，突出显示每个消息组如何在不同的消费者服务中独立维持其顺序。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-grouping.png) ## 按消息组分发数据 IDs 以提高性能 为了优化传输吞吐量，Amazon SNS FIFO 主题并行传送来自不同消息组的消息，同时严格维护每个消息组内的消息顺序。每个消息组每秒最多可以传送 300 条消息。因此，要实现单个主题的高吞吐量，请使用大量不同的消息组 IDs。通过利用一组不同的消息组，Amazon SNS FIFO 主题可自动在更多并行分区中分发消息。 **注意** Amazon SNS FIFO 主题经过优化，可以跨消息组均匀分发消息 IDs，无论组的数量如何。 AWS 建议您使用大量不同的消息组 IDs 以优化性能。 当以高吞吐量发布到您的 Amazon SNS FIFO 主题并且订阅了一个或多个 Amazon SQS FIFO 队列时，建议您在队列上启用高吞吐量。有关更多信息，请参阅《Amazon Simple Queue Service 开发人员指南》**中的 [FIFO 队列的高吞吐量](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/high-throughput-fifo.html)。 # FIFO 主题的 Amazon SNS 消息传输 Amazon SNS FIFO（先入先出）主题支持向 Amazon SQS 标准队列和 FIFO 队列交付，让客户在集成需要近乎实时数据一致性的分布式应用程序时具有灵活性和控制力。 对于需要保持严格的消息排序或重复数据删除的工作负载，Amazon SNS FIFO 主题与作为传输端点订阅的 [Amazon SQS FIFO 队列](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html)相结合，可在操作和事件顺序至关重要或不能容忍重复数据时增强应用程序之间的消息传输。 对于允许尽力而为的订购和至少一次交付的工作负载，订阅 [Amazon SQS 标准队列](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues.html)到 Amazon SNS FIFO 主题后，除了可以在不使用 FIFO 的工作负载之间共享队列外，还可以降低成本。 **注意** 要将来自 Amazon SNS FIFO 主题的消息扇出到 AWS Lambda 函数，则需要额外的步骤。首先，为 Amazon SQS FIFO 队列或标准队列订阅主题。然后，配置队列以触发函数。有关更多信息，请参阅 *AWS 计算博客*上的[作为事件源的 SQS FIFO](https://aws.amazon.com/blogs/compute/new-for-aws-lambda-sqs-fifo-as-an-event-source/) 博文。 SNS FIFO 主题无法将消息传输到客户管理的终端节点，例如电子邮件地址、移动应用程序、用于收发短信 (SMS) 的电话号码或 HTTP(S) 终端节点。这些终端节点类型不能保证保留严格的消息排序。尝试将客户管理的终端节点订阅到 SNS FIFO 主题会导致错误。 SNS FIFO 主题支持与标准主题相同的消息筛选功能。有关更多信息，请参阅 *AWS 计算博客*上的 [FIFO 主题的 Amazon SNS 消息筛选](fifo-message-filtering.md) 和[使用 Amazon SNS 消息筛选功能简化您的发布/订阅消息收发](https://aws.amazon.com/blogs/compute/simplify-pubsub-messaging-with-amazon-sns-message-filtering/)。 # FIFO 主题的 Amazon SNS 消息筛选 Amazon SNS FIFO 主题支持消息筛选。使用消息筛选可通过从发布者系统卸载消息路由逻辑，从订阅者系统卸载消息筛选逻辑来简化您的架构。 当您为 Amazon SQS FIFO 队列或标准队列订阅 SNS FIFO 主题时，您可以使用消息筛选来指定订阅者接收消息的子集，而不是所有消息。每个订阅者都可以将其自己的筛选策略设置为订阅属性。根据筛选策略的范围，筛选策略与入站消息属性或消息正文进行匹配。如果筛选策略匹配，则主题会向订阅者传输消息的副本。如果没有匹配项，则主题不会传输消息的副本。 在[汽车零部件价格管理示例使用案例](fifo-example-use-case.md)中，假设已设置以下 Amazon SNS 筛选策略且筛选策略范围是 `MessageBody`： + 对于批发队列，筛选策略 `{"business":["wholesale"]}` 匹配包含名为 `business` 的键且在一组值中具有 `wholesale` 的每条消息。在下图中，消息 **m1** 中的键之一是值为 `wholesale` 的 `business`。消息 **m3** 中的键之一是值为 `["wholesale,retail"]` 的 `business`。因此，**m1** 和 **m3** 均匹配筛选策略的条件，并且这两条消息都会传输到批发队列中。 + 对于零售队列，筛选策略 `{"business":["retail"]}` 匹配包含名为 `business` 的键且在一组值中具有 `retail` 的每条消息。在图中，消息 **m2** 中的键之一是值为 `retail` 的 `business`。消息 **m3** 中的键之一是值为 `["wholesale,retail"]` 的 `business`。因此，**m2** 和 **m3** 均匹配筛选策略的条件，并且这两条消息都会传输到零售队列中。 + 对于分析队列，我们希望 Amazon Athena 接收所有记录，因此不应用任何筛选策略。 ![\[Amazon SNS FIFO 主题如何利用筛选策略根据消息内容将消息路由到不同的 Amazon SQS 队列的示例。“business”键设置为“wholesale”的消息 m1 和 m3 将传送到批发队列，而带有“retail”的消息 m2 则指定指向零售队列的路由。这演示了有针对性的消息传送，确保每个订阅用户功能仅接收相关的消息，从而优化了处理效率和数据相关性。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-filtering.png) SNS FIFO 主题支持各种匹配运算符，包括属性字符串值、属性数值和属性键。有关更多信息，请参阅 [Amazon SNS 消息筛选](sns-message-filtering.md)。 SNS FIFO 主题不会向订阅的终端节点传输重复的消息。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息重复数据删除](fifo-message-dedup.md)。 # FIFO 主题的 Amazon SNS 消息重复数据删除 Amazon SNS FIFO 主题和 Amazon SQS FIFO 队列支持消息重复数据删除，只要满足以下条件，即可提供一次性消息传输和处理： + 订阅的 Amazon SQS FIFO 队列存在，并具有允许 Amazon SNS 服务委托人向队列传输消息的权限。 + Amazon SQS FIFO 队列使用者处理消息，并在可见性超时到期之前将其从队列中删除。 + Amazon SNS 订阅主题没有[消息筛选](fifo-message-filtering.md)功能。配置消息筛选时，Amazon SNS FIFO 主题支持最多一次性传输，因为可以根据订阅筛选策略筛选出消息。 + 没有阻止确认邮件传输的网络中断。 **注意** 当主题属性 `FifoThroughputScope` 设置为 `Topic` 时，消息重复数据删除将应用于整个 Amazon SNS FIFO 主题。当主题属性 `FifoThroughputScope` 设置为 `MessageGroup` 时，消息重复数据删除将应用于每个单独的[消息组](fifo-message-grouping.md)。 当您将消息发布到 Amazon SNS FIFO 主题时，该消息必须包含重复数据删除 ID。此 ID 包含在 Amazon SNS FIFO 主题传输给订阅的 Amazon SQS FIFO 队列的消息中。 如果具有特定重复数据删除 ID 的消息成功发布到 Amazon SNS FIFO 主题，则在五分钟重复数据删除间隔内使用相同重复数据删除 ID 发布的任何消息都将被接受，但不会传输。Amazon SNS FIFO 主题会在主题属性 `FifoThroughputScope` 配置的重复数据删除范围内继续跟踪该消息重复数据删除 ID，即使在消息被传输到订阅的端点之后也是如此。 如果保证消息正文对于每个已发布的消息都是唯一的，您可以为 Amazon SNS FIFO 主题和订阅的 Amazon SQS FIFO 队列启用基于内容的重复数据删除。Amazon SNS 使用消息正文生成一个唯一的哈希值，以用作每个消息的重复数据删除 ID，因此您无需在发送每条消息时设置重复数据删除 ID。 **注意** 消息属性不包括在哈希计算中。 如果为 Amazon SNS FIFO 主题启用基于内容的重复数据删除，并且发布了带有重复数据删除 ID 的消息，则发布的重复数据删除 ID 将覆盖生成的基于内容的重复数据删除 ID。 在[汽车零部件价格管理示例使用案例](fifo-example-use-case.md)中，公司必须为每次价格更新设置一个通用唯一的重复数据删除 ID。这是因为即使在批发和零售的消息属性不同时，消息正文也可以是相同的。但是，如果公司将业务类型（批发或零售）与产品 ID 和产品价格一起添加到消息正文中，它们可以在 Amazon SNS FIFO 主题和订阅的 Amazon SQS FIFO 队列中启用基于内容的复制。 ![\[使用汽车零件价格管理示例，消息重复数据删除在 Amazon SNS FIFO（先入先出）主题环境中的工作原理。它显示了如何防止发布到 Amazon SNS FIFO 主题的重复消息（m1）多次传送到订阅用户系统（批发、零售和分析队列）。这种重复数据删除可确保只处理唯一的消息，从而提高不同订阅用户功能之间消息处理的效率和准确性。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-dedup.png) 除了消息排序和重复数据删除以外，Amazon SNS FIFO 主题还支持使用 AWS KMS 进行消息服务器端加密 (SSE) 以及 AWS PrivateLink 保护通过 VPC 端点的消息隐私。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息安全性](fifo-message-security.md)。 # FIFO 主题的 Amazon SNS 消息安全性 您可以使用[AWS Key Management Service（AWS KMS）](https://aws.amazon.com/kms/)[客户](https://docs.aws.amazon.com/kms/latest/developerguide/concepts.html#master_keys)主密钥（CMK）为 Amazon SNS FIFO 主题和 Amazon SQS FIFO 队列启用加密。 + 您可以创建新的加密 FIFO 主题和队列，也可以为现有的 FIFO 主题和队列启用加密。 + 只有消息正文会被加密。消息属性、资源元数据和资源指标不会加密。 **注意** 将加密添加到现有 FIFO 主题或队列中不会加密任何积压的消息，从主题或队列中删除加密会使积压的消息加密。 SNS FIFO 主题会立即解密消息，然后将消息传递到订阅的终端节点。SQS FIFO 队列在将消息返回到使用者应用程序之前解密消息。有关更多信息，请参阅 *AWS 计算博客*上的 [Amazon SNS 数据加密](sns-data-encryption.md) 和[使用 AWS KMS 加密发布到 Amazon SNS 的消息](https://aws.amazon.com/blogs/compute/encrypting-messages-published-to-amazon-sns-with-aws-kms/)。 此外，SNS FIFO 主题和 SQS FIFO 队列支持[接口 VPC 终端节点](https://docs.aws.amazon.com/vpc/latest/userguide/vpce-interface.html)由 AWS PrivateLink 支持的消息隐私。使用接口终端节点，您可以将消息从 Amazon Virtual Private Cloud (Amazon VPC) 子网发送到 FIFO 主题和队列，而无需遍历公有 Internet。此模型将您的消息收发保持在 AWS 基础设施和网络中，从而增强应用程序的整体安全性。当您使用 AWS PrivateLink 时，无需设置 Internet 网关、网络地址转换 (NAT) 或 Virtual Private Network (VPN)。有关更多信息，请参阅 *AWS 安全性博客*上的 [使用 VPC 端点保护 Amazon SNS 流量](sns-internetwork-traffic-privacy.md) 和[使用 AWS PrivateLink 保护发布到 Amazon SNS 的消息](https://aws.amazon.com/blogs/security/securing-messages-published-to-amazon-sns-with-aws-privatelink)。 SNS FIFO 主题还支持跨可用区的死信队列和消息存储。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息持久性](fifo-message-durability.md)。 # FIFO 主题的 Amazon SNS 消息持久性 Amazon SNS FIFO 主题和 Amazon SQS 队列具有持久性。这两种资源类型都以冗余方式跨多个可用区存储消息，并提供死信队列以处理异常情况。 在 Amazon SNS 中，当 Amazon SNS 主题由于客户端或服务器端错误无法访问订阅的 Amazon SQS 队列时，消息传输将失败。 + 当 Amazon SNS FIFO 主题具有过时的订阅元数据时，会发生客户端错误。两个常见的客户端错误发生于 Amazon SQS 队列所有者执行以下操作之一时： + 删除队列。 + 以防止 Amazon SNS 服务委托人向其传输消息的方式更改队列策略。 Amazon SNS 不会重试传输由于客户端错误而失败的消息。 + 在以下情况下可能会发生服务器端错误： + Amazon SQS 服务不可用。 + Amazon SQS 无法处理来自 Amazon SNS 服务的有效请求。 在发生服务器端错误时，Amazon SNS FIFO 主题会最多重试失败的传输 100015 次（时间超过 23 天）。有关更多信息，请参阅 [Amazon SNS 消息传输重试](sns-message-delivery-retries.md)。 对于任何类型的错误，Amazon SNS 都可以将消息搁置到 Amazon SQS 死信队列中，以免数据丢失。 在 Amazon SQS 中，当使用者应用程序无法接收消息、处理消息并从队列中删除消息时，消息处理将失败。当接收请求的最大数量失败时，Amazon SQS 可以将消息搁置到死信队列中，以免数据丢失。 在[汽车零部件价格管理示例使用案例](fifo-example-use-case.md)，公司可以为每个 Amazon SNS FIFO 主题订阅以及每个订阅的 Amazon SQS 队列分配一个 Amazon SQS 死信队列 (DLQ)。这可以保护公司免受任何价格更新损失。 ![\[一个示例，演示如何将死信队列（DLQ）与 Amazon SNS FIFO 主题和 Amazon SQS 队列集成，以确保汽车零部件价格管理系统中的消息可靠性。它显示了用于批发、零售和分析目的的每个 Amazon SNS FIFO 订阅与相应的 Amazon SQS FIFO 或标准队列配对的设置，每个队列都配备了自己的类型匹配的 DLQ，以防止因处理失败而导致的消息丢失。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-fifo-dlq.png) 与 Amazon SNS 订阅关联的死信队列必须是与订阅队列类型相同的 Amazon SQS 队列。例如，Amazon SQS FIFO 队列的 Amazon SNS FIFO 订阅必须将 Amazon SQS FIFO 队列用作死信队列。同样，Amazon SQS 标准队列的 Amazon SNS FIFO 订阅必须使用 Amazon SQS 标准队列作为死信队列。有关更多信息，请参阅 *AWS 计算博客*上的 [Amazon SNS 死信队列](sns-dead-letter-queues.md) 和[使用 DLQ 为 Amazon SNS、Amazon SQS、AWS Lambda 设计持久的无服务器应用程序](https://aws.amazon.com/blogs/compute/designing-durable-serverless-apps-with-dlqs-for-amazon-sns-amazon-sqs-aws-lambda/ )博客文章。 为了延长持久性以帮助从下游故障中恢复，主题所有者还可以使用 FIFO 主题将消息归档长达 365 天。然后，主题订阅用户可以将这些消息重播到已订阅的端点，以恢复因下游应用程序故障而丢失的消息，或者复制现有应用程序的状态。有关更多信息，请参阅[FIFO 主题的 Amazon SNS 消息归档与重播功能](fifo-message-archiving-replay.md)。 # FIFO 主题的 Amazon SNS 消息归档与重播功能 ## 什么是消息归档与重播功能？ Amazon SNS 提供无代码消息归档与重播功能，专门针对 FIFO（先进先出）主题设计。此功能允许主题所有者将消息直接存储在主题归档中长达 365 天，并在需要时重播给订阅用户。消息归档与重播功能对于恢复丢失的消息以及通过复制状态跨区域或系统同步应用程序至关重要。 可通过 AWS API CloudFormation、SDK 和访问此功能 AWS 管理控制台。 **关键用例** + **消息恢复** - 通过将消息重播到订阅用户的端点，恢复因下游应用程序故障而丢失的消息。 + **状态复制** - 通过重播从特定时间戳开始的消息，在新环境中复制现有系统的状态。 + **错误更正** - 在中断期间重新发送错过的消息，以确保正确处理所有事件。 ### 消息归档与重播功能的组成部分 管理 Amazon SNS FIFO 主题的消息存档和重播，包括设置保留期、使用监控存档消息 CloudWatch、通过订阅属性启动重播以及了解修改和启动重播所需的权限。 **消息归档** + 主题所有者启用归档功能并设置消息保留期，最长可达 365 天。有关更多信息，请参阅[适用于 FIFO 主题所有者的 Amazon SNS 消息归档](message-archiving-and-replay-topic-owner.md) + CloudWatch 指标有助于监控存档的邮件。 **消息重播** + 订阅用户启动重播，选择将消息重新处理到订阅端点的时间窗口。有关更多信息，请参阅[FIFO 主题订阅用户的 Amazon SNS 消息重播](message-archiving-and-replay-subscriber.md)。 + 您可以使用 `ReplayPolicy` 功能通过订阅属性管理重播。 **相关权限** + **`SetSubscriptionAttributes`** – 需要此权限才能使用订阅上的 `ReplayPolicy` 属性配置或修改重播设置。 + **`Subscribe`** – 需要此权限才能附加新的订阅并启动重播。 + **`GetTopicAttributes`** – 允许查看主题的属性，但重播启动主要围绕订阅管理展开。 # 适用于 FIFO 主题所有者的 Amazon SNS 消息归档 通过消息归档，您可以归档发布到您的主题的所有消息的单个副本。您可以通过对主题启用消息归档策略将已发布的消息存储在主题中，该策略将为链接到该主题的所有订阅启用消息归档。消息可以归档至少一天，最多 365 天。 设置归档策略时需支付额外费用。有关定价信息，请参阅 [Amazon SNS 定价](https://aws.amazon.com/sns/pricing/)。 ## 使用创建邮件存档策略 AWS 管理控制台 使用此选项，通过 AWS 管理控制台创建新的消息归档策略。 1. 登录 [Amazon SNS 控制台](https://console.aws.amazon.com/sns/home)。 1. 选择一个主题或创建一个新主题。要了解有关创建主题的更多信息，请参阅[创建 Amazon SNS 主题](sns-create-topic.md)。 **注意** Amazon SNS 消息存档和重播仅适用于 application-to-application (A2A) FIFO 主题。 1. 在**编辑主题**页面上，展开**归档策略**部分。 1. 启用**归档策略**特征，然后输入要在主题中归档消息的**天数**。 1. 选择**保存更改**。 **查看、编辑和停用消息归档主题策略** + 在**主题详细信息**页面上，**保留策略**显示归档策略的状态，包括设置该策略对应的天数。选择**归档策略**选项卡以查看以下消息归档详细信息： + **状态** - 应用归档策略后，归档与重播功能状态显示为**活动**。当归档策略设置为空的 JSON 对象时，归档与重播功能状态显示为**不活动**。 + **消息保留期** - 指定的消息保留天数。 + **归档开始日期** - 订阅用户可以重播消息的起始日期。 + **JSON 预览** - 归档策略的 JSON 预览。 + （可选）要**编辑**归档策略，请转到主题摘要页面并选择**编辑**。 + （可选）要**停用**归档策略，请转至主题摘要页面并选择**编辑**。停用**归档策略**并选择**保存更改**。 + （可选）要使用归档策略**删除**主题，必须先按照前面所述停用归档策略。 **重要** 为避免意外删除消息，您不能在具有有效消息归档策略的情况下删除主题。必须先停用主题的消息归档策略，然后才能删除该主题。当您停用消息归档策略时，Amazon SNS 会删除所有已归档的消息。删除主题时，订阅将被删除，并且任何传输中的消息都可能无法传送。 ## 使用 API 创建消息归档策略 要使用 API 创建消息归档策略，您需要将属性 `ArchivePolicy` 添加到主题中。您可以使用 API 操作 `CreateTopic` 和 `SetTopicAttributes` 设置 `ArchivePolicy`。`ArchivePolicy` 只有一个值 `MessageRetentionPeriod`，它表示 Amazon SNS 保留消息的天数。要为主题激活消息归档，请将 `MessageRetentionPeriod` 设置为大于零的整数值。例如，要将归档中的消息保留 30 天，请将 `ArchivePolicy` 设置为： ``` { "ArchivePolicy": { "MessageRetentionPeriod": "30" } } ``` 要对主题禁用消息归档并清除归档，请取消设置 `ArchivePolicy`，如下所示： ``` {} ``` ## 使用 SDK 创建消息归档策略 要使用 S AWS DK，必须使用您的凭据对其进行配置。有关更多信息，请参阅《工具参考指南》[`config`和《*工具参考指南》中的共享 AWS SDKs 和`credentials`*文件](https://docs.aws.amazon.com/sdkref/latest/guide/file-format.html)。 以下代码示例展示如何为 Amazon SNS 主题设置 `ArchivePolicy`，以便将发布到该主题的所有消息保留 30 天。 ``` // Specify the ARN of the Amazon SNS topic to set the ArchivePolicy for. String topicArn = "arn:aws:sns:us-east-2:123456789012:MyArchiveTopic.fifo"; // Set the MessageRetentionPeriod to 30 days for the ArchivePolicy. String archivePolicy = "{\"MessageRetentionPeriod\":\"30\"}"; // Set the ArchivePolicy for the Amazon SNS topic SetTopicAttributesRequest request = new SetTopicAttributesRequest() .withTopicArn(topicArn) .withAttributeName("ArchivePolicy") .withAttributeValue(archivePolicy); sns.setTopicAttributes(request); ``` ## 使用创建邮件存档策略 CloudFormation 要使用创建存档策略， CloudFormation 请参阅*AWS CloudFormation 用户指南[https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sns-topic.html](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sns-topic.html)*中的。 ## 授予对加密归档的访问权限 必须先完成以下步骤，然后订阅用户才能开始重播来自加密主题的消息。由于过去的消息会被重播，因此需要为 Amazon SNS 预调配对 KMS 密钥的 `Decrypt` 访问权限，此密钥用于加密归档中的消息。 1. 当您使用 KMS 密钥加密消息并将其存储在主题中时，必须授予 Amazon SNS 通过密钥策略对这些消息解密的能力。有关更多信息，请参阅[向 Amazon SNS 授予解密权限](#message-archiving-and-replay-topic-decrypt-permissions)。 1. AWS KMS 为亚马逊 SNS 启用。有关更多信息，请参阅[配置 AWS KMS 权限](sns-key-management.md#sns-what-permissions-for-sse)。 **重要** 向 KMS 密钥策略添加新部分时，不要更改策略中任何已存在的部分。如果对主题启用了加密但禁用或删除了 KMS 密钥，或未针对 Amazon SNS 正确地配置 KMS 密钥策略，则 Amazon SNS 无法向您的订阅用户重播消息。 ### 向 Amazon SNS 授予解密权限 要让 Amazon SNS 访问您主题的归档中的加密消息并将其重播到已订阅的端点，您必须启用 Amazon SNS 服务原则来解密这些消息。 以下是允许 Amazon SNS 服务主体在重播主题内的历史消息时解密存储的消息所需的示例策略。 ``` { "Sid": "Allow SNS to decrypt archived messages", "Effect": "Allow", "Principal": { "Service": "sns.amazonaws.com" }, "Action": [ "kms:Decrypt", "kms:GenerateDataKey" ], "Resource": "*" } ``` ## 使用 Amazon 监控邮件存档指标 CloudWatch 您可以使用以下指标在 Amaz CloudWatch on 上监控存档的邮件。为了收到工作负载异常的通知并帮助避免影响，您可以根据这些指标配置 Amazon CloudWatch 警报。有关更多详细信息，请参阅[Amazon SNS 中的日志记录和监控](sns-logging-monitoring.md)。 | 指标 | 说明 | | --- | --- | | **ApproximateNumberOfMessagesArchived** | 以 60 分钟的分辨率向主题所有者提供在主题归档中归档的消息总数。 | | **ApproximateNumberOfBytesArchived** | 以 60 分钟的分辨率向主题所有者提供对主题归档中的所有消息归档的总字节数。 | | **NumberOfMessagesArchiveProcessing** | 以 1 分钟的分辨率向主题所有者提供在间隔期间保存到主题存档的消息数。 | | **NumberOfBytesArchiveProcessing** | 以 1 分钟的分辨率向主题所有者提供在间隔期间保存到主题存档的总字节数。 | `GetTopicAttributes` API 有一个 `BeginningArchiveTime` 属性，它表示订阅用户可以开始重播的最早时间戳。以下是此 API 操作的示例响应： ``` { "ArchivePolicy": { "MessageRetentionPeriod": "" }, "BeginningArchiveTime": "", ... } ``` # FIFO 主题订阅用户的 Amazon SNS 消息重播 借助 Amazon SNS 重播，主题订阅用户可以从主题数据存储中检索归档的消息，并将其重新传输（或重播）到订阅的端点。 + 创建订阅后即可立即重播消息。 + 重播的消息保留了与原始消息相同的内容、`MessageId` 和 `Timestamp`。 + 该消息包含一个 `Replayed` 属性，用于指示这是一条重播的消息。 + 要仅重播指定消息，可以在订阅中添加筛选策略。 有关筛选消息的更多信息，请参阅[筛选重播的消息](#message-archiving-and-replay-subscription-filtering)。 ## 使用创建消息重播策略 AWS 管理控制台 使用此选项，通过 AWS 管理控制台创建新的重播策略。 1. 登录 [Amazon SNS 控制台](https://console.aws.amazon.com/sns/home)。 1. 选择一个主题订阅或创建一个新的主题订阅。要了解有关创建订阅的更多信息，请参阅[在 Amazon SNS 主题中创建订阅](sns-create-subscribe-endpoint-to-topic.md)。 1. 要启动消息重播，请转到**重播**下拉列表并选择**开始重播**。 1. 从**重播时间范围**模式中，进行以下选择： 1. **选择重播开始日期和时间**-选择要开始重播存档邮件的**日期**（YYYY/MM/DD格式）和**时间**（24 小时 hh: mm: ss 格式）。开始时间应晚于近似归档时间的开始时间。 1. **（可选）选择重播结束日期和时间**-选择要停止重播存档邮件的**日期**（YYYY/MM/DD格式）和**时间**（24 小时 hh: mm: ss 格式）。 1. 选择**开始重播**。 1. （可选）要**停止**消息重播，请转到**订阅详细信息**页面，然后从**重播**下拉列表中选择**停止重播**。 1. （可选）要使用**监控**此工作流程中的邮件重播指标 CloudWatch，请参阅[使用 Amazon 监控消息重播指标 CloudWatch](#message-archiving-and-replay-subscription-cloudwatch)。 **查看和编辑消息重播策略** 您可以从**订阅详细信息**页面执行以下操作： + 要**查看**消息重播状态，**重播状态**字段将显示以下值： + **已完成** - 重播已成功重新传送所有消息，现在正在传送新发布的消息。 + **进行中** – 重播当前正在重播所选消息。 + **失败** – 重播无法完成。 + **待处理** - 重播启动时的默认状态。 + （可选）要**修改**消息重播策略，请转到**订阅详细信息**页面，然后从**重播**下拉列表中选择**开始重播**。开始某个重播将取代现有的重播。 ## 使用 API 向订阅添加重播策略 要重播归档的消息，请使用属性 `ReplayPolicy`。`ReplayPolicy` 可以与 `Subscribe` 和 `SetSubscriptionAttributes` API 操作一起使用。此策略包含以下值： + **`StartingPoint`**（必需）- 表示从何处开始重播消息。 + **`EndingPoint`**（可选）- 表示何时停止重播消息。如果省略 `EndingPoint`，则重播将继续，直到赶上当前时间。 + **`PointType`**（必需）- 设置起点和终点的类型。目前，`PointType` 唯一支持的值是 `Timestamp`。 例如，要从下游故障中恢复并重新发送 2023 年 10 月 1 日的两小时时段内的所有消息，请使用 `SetSubscriptionAttributes` API 操作设置 `ReplayPolicy`，如下所示： ``` { "PointType":"Timestamp", "StartingPoint":"2023-10-01T10:00:00.000Z", "EndingPoint":"2023-10-01T12:00:00.000Z" } ``` 要重播截至 2023 年 10 月 1 日发送到该主题的所有消息，并继续接收与您的主题有关的所有新发布的消息，请使用 `SetSubscriptionAttributes` API 操作对您的订阅设置 `ReplayPolicy`，如下所示： ``` { "PointType":"Timestamp", "StartingPoint":"2023-10-01T00:00:00.000Z" } ``` 为了验证消息是否已重播，将在每条重播的消息中添加布尔属性 `Replayed`。 ## 使用 SDK 向订阅添加重播策略 要使用 S AWS DK，必须使用您的凭据对其进行配置。有关更多信息，请参阅《工具参考指南》[`config`和《*工具参考指南》中的共享 AWS SDKs 和`credentials`*文件](https://docs.aws.amazon.com/sdkref/latest/guide/file-format.html)。 以下代码示例显示了如何对订阅设置 `ReplayPolicy`，以便从 Amazon SNS FIFO 主题的归档中重新传送 2023 年 10 月 1 日 2 小时时段内的消息。 ``` // Specify the ARN of the Amazon SNS subscription to initiate the ReplayPolicy on. String subscriptionArn = "arn:aws:sns:us-east-2:123456789012:MyArchiveTopic.fifo:1d2a3e9d-7f2f-447c-88ae-03f1c68294da"; // Set the ReplayPolicy to replay messages from the topic's archive // for a 2 hour time period on October 1st 2023 between 10am and 12pm UTC. String replayPolicy = "{\"PointType\":\"Timestamp\",\"StartingPoint\":\"2023-10-01T10:00:00.000Z\",\"EndingPoint\":\"2023-10-01T12:00:00.000Z\"}"; // Set the ArchivePolicy for the Amazon SNS topic SetSubscriptionAttributesRequest request = new SetSubscriptionAttributesRequest() .withSubscriptionArn(subscriptionArn) .withAttributeName("ReplayPolicy") .withAttributeValue(replayPolicy); sns.setSubscriptionAttributes(request); ``` ## 理解 EndingPoint 当您向 Amazon SNS 订阅应用 `ReplayPolicy` 时，`EndingPoint` 值是可选的。如果未提供 `EndingPoint`，则重播将从指定的 `StartingPoint` 开始并持续到当前时间，包括处理任何新发布的消息。一旦赶上，订阅将作为常规订阅运行，在新消息发布后接收这些消息。 如果指定了 `EndingPoint`，则该服务将从 `StartingPoint` 一直到 `EndingPoint` 重播消息，然后停止。**此操作实际上会暂停订阅。**订阅暂停期间，新发布的消息将不会传送到已订阅的端点。 要恢复消息传送，请在不提供 `EndingPoint` 的情况下应用新的 `ReplayPolicy`，并将 `StartingPoint` 设置为继续接收消息的所需时间点。例如，要从之前的重播结束的地方恢复订阅，请将新的 `StartingPoint` 设置为先前提供的 `EndingPoint`。 ## 筛选重播的消息 Amazon SNS 消息筛选可让您控制 Amazon SNS 向您的订阅用户端点重播的消息。当消息筛选和消息归档都已启用时，Amazon SNS 会首先从主题的数据存储中检索消息，然后根据订阅的 `FilterPolicy` 应用消息。当存在匹配项时，消息将传送到订阅的端点，否则消息将被筛选掉。有关更多信息，请参阅 [Amazon SNS 订阅筛选策略](sns-subscription-filter-policies.md)。 ## 使用 Amazon 监控消息重播指标 CloudWatch 您可以使用以下指标在 Amaz CloudWatch on 上监控重播消息。为了收到工作负载异常的通知并帮助避免影响，您可以根据这些指标配置 Amazon CloudWatch 警报。有关更多详细信息，请参阅[Amazon SNS 中的日志记录和监控](sns-logging-monitoring.md)。 | 指标 | 说明 | | --- | --- | | **NumberOfReplayedNotificationsDelivered** | 以 1 分钟的分辨率向订阅用户提供主题归档中重播的消息总数。 | | **NumberOfReplayedNotificationsFailed** | 以 1 分钟的分辨率向订阅用户提供主题归档中未能传送的已重播消息的总数。 | # 适用于 FIFO 主题的 Amazon SNS 代码示例 您可以使用以下代码示例，将[汽车零部件价格管理示例使用案例](fifo-example-use-case.md)与 Amazon SNS FIFO 主题集成，并可选择搭配 Amazon SQS FIFO 队列或标准队列使用。 ## 使用 AWS 软件开发工具包 使用 AWS 软件开发工具包，您可以通过将其`FifoTopic`属性设置为来创建 Amazon SNS FIFO 主题。**true**您可以通过将 Amazon SQS FIFO 队列的 `FifoQueue` 属性设置为 **true** 来创建该队列。此外，您必须将 **.fifo** 后缀添加到每个 FIFO 资源的名称。创建 FIFO 主题或队列后，无法将其转换为标准主题或队列。 以下代码示例创建这些 FIFO 和标准队列资源： + 分发价格更新的 Amazon SNS FIFO 主题 + 为批发和零售应用程序提供这些更新的 Amazon SQS FIFO 队列 + 用于存储记录的分析应用程序的 Amazon SQS 标准队列，可以查询这些记录以获取商业智能 (BI) + 将三个队列连接到主题的 Amazon SNS FIFO 订阅 本示例将设置订阅中的[筛选条件策略](sns-subscription-filter-policies.md)。如果通过向主题发布消息来测试示例，请确保您发布的是带 `business` 属性的消息。为属性值指定 `retail` 或 `wholesale`。否则，消息将被筛选掉，且不会传递到订阅的队列中。有关更多信息，请参阅 [FIFO 主题的 Amazon SNS 消息筛选](fifo-message-filtering.md)。 ------ #### [ Java ] **适用于 Java 的 SDK 2.x** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 此示例 + 创建一个 Amazon SNS FIFO 主题、两个 Amazon SQS FIFO 队列和一个标准队列。 + 将队列订阅到主题，发布一条消息到主题。 该[测试](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sns/src/test/java/com/example/sns/PriceUpdateExampleTest.java)验证每个队列是否收到消息。[完整的示例](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sns/src/main/java/com/example/sns/PriceUpdateExample.java)还显示了添加访问策略，并在最后删除了资源。 ``` public class PriceUpdateExample { public final static SnsClient snsClient = SnsClient.create(); public final static SqsClient sqsClient = SqsClient.create(); public static void main(String[] args) { final String usage = "\n" + "Usage: " + " \n\n" + "Where:\n" + " fifoTopicName - The name of the FIFO topic that you want to create. \n\n" + " wholesaleQueueARN - The name of a SQS FIFO queue that will be created for the wholesale consumer. \n\n" + " retailQueueARN - The name of a SQS FIFO queue that will created for the retail consumer. \n\n" + " analyticsQueueARN - The name of a SQS standard queue that will be created for the analytics consumer. \n\n"; if (args.length != 4) { System.out.println(usage); System.exit(1); } final String fifoTopicName = args[0]; final String wholeSaleQueueName = args[1]; final String retailQueueName = args[2]; final String analyticsQueueName = args[3]; // For convenience, the QueueData class holds metadata about a queue: ARN, URL, // name and type. List queues = List.of( new QueueData(wholeSaleQueueName, QueueType.FIFO), new QueueData(retailQueueName, QueueType.FIFO), new QueueData(analyticsQueueName, QueueType.Standard)); // Create queues. createQueues(queues); // Create a topic. String topicARN = createFIFOTopic(fifoTopicName); // Subscribe each queue to the topic. subscribeQueues(queues, topicARN); // Allow the newly created topic to send messages to the queues. addAccessPolicyToQueuesFINAL(queues, topicARN); // Publish a sample price update message with payload. publishPriceUpdate(topicARN, "{\"product\": 214, \"price\": 79.99}", "Consumables"); // Clean up resources. deleteSubscriptions(queues); deleteQueues(queues); deleteTopic(topicARN); } public static String createFIFOTopic(String topicName) { try { // Create a FIFO topic by using the SNS service client. Map topicAttributes = Map.of( "FifoTopic", "true", "ContentBasedDeduplication", "false", "FifoThroughputScope", "MessageGroup"); CreateTopicRequest topicRequest = CreateTopicRequest.builder() .name(topicName) .attributes(topicAttributes) .build(); CreateTopicResponse response = snsClient.createTopic(topicRequest); String topicArn = response.topicArn(); System.out.println("The topic ARN is" + topicArn); return topicArn; } catch (SnsException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } return ""; } public static void subscribeQueues(List queues, String topicARN) { queues.forEach(queue -> { SubscribeRequest subscribeRequest = SubscribeRequest.builder() .topicArn(topicARN) .endpoint(queue.queueARN) .protocol("sqs") .build(); // Subscribe to the endpoint by using the SNS service client. // Only Amazon SQS queues can receive notifications from an Amazon SNS FIFO // topic. SubscribeResponse subscribeResponse = snsClient.subscribe(subscribeRequest); System.out.println("The queue [" + queue.queueARN + "] subscribed to the topic [" + topicARN + "]"); queue.subscriptionARN = subscribeResponse.subscriptionArn(); }); } public static void publishPriceUpdate(String topicArn, String payload, String groupId) { try { // Create and publish a message that updates the wholesale price. String subject = "Price Update"; String dedupId = UUID.randomUUID().toString(); String attributeName = "business"; String attributeValue = "wholesale"; MessageAttributeValue msgAttValue = MessageAttributeValue.builder() .dataType("String") .stringValue(attributeValue) .build(); Map attributes = new HashMap<>(); attributes.put(attributeName, msgAttValue); PublishRequest pubRequest = PublishRequest.builder() .topicArn(topicArn) .subject(subject) .message(payload) .messageGroupId(groupId) .messageDeduplicationId(dedupId) .messageAttributes(attributes) .build(); final PublishResponse response = snsClient.publish(pubRequest); System.out.println(response.messageId()); System.out.println(response.sequenceNumber()); System.out.println("Message was published to " + topicArn); } catch (SnsException e) { System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); } } ``` + 有关 API 详细信息，请参阅《AWS SDK for Java 2.x API Reference》**中的以下主题。 + [CreateTopic](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/CreateTopic) + [发布](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/Publish) + [Subscribe](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/Subscribe) ------ #### [ Python ] **适用于 Python 的 SDK（Boto3）** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 创建 Amazon SNS FIFO 主题，将 Amazon SQS FIFO 队列和标准队列订阅到主题，并发布一条消息到主题。 ``` def usage_demo(): """Shows how to subscribe queues to a FIFO topic.""" print("-" * 88) print("Welcome to the `Subscribe queues to a FIFO topic` demo!") print("-" * 88) sns = boto3.resource("sns") sqs = boto3.resource("sqs") fifo_topic_wrapper = FifoTopicWrapper(sns) sns_wrapper = SnsWrapper(sns) prefix = "sqs-subscribe-demo-" queues = set() subscriptions = set() wholesale_queue = sqs.create_queue( QueueName=prefix + "wholesale.fifo", Attributes={ "MaximumMessageSize": str(4096), "ReceiveMessageWaitTimeSeconds": str(10), "VisibilityTimeout": str(300), "FifoQueue": str(True), "ContentBasedDeduplication": str(True), }, ) queues.add(wholesale_queue) print(f"Created FIFO queue with URL: {wholesale_queue.url}.") retail_queue = sqs.create_queue( QueueName=prefix + "retail.fifo", Attributes={ "MaximumMessageSize": str(4096), "ReceiveMessageWaitTimeSeconds": str(10), "VisibilityTimeout": str(300), "FifoQueue": str(True), "ContentBasedDeduplication": str(True), }, ) queues.add(retail_queue) print(f"Created FIFO queue with URL: {retail_queue.url}.") analytics_queue = sqs.create_queue(QueueName=prefix + "analytics", Attributes={}) queues.add(analytics_queue) print(f"Created standard queue with URL: {analytics_queue.url}.") topic = fifo_topic_wrapper.create_fifo_topic("price-updates-topic.fifo") print(f"Created FIFO topic: {topic.attributes['TopicArn']}.") for q in queues: fifo_topic_wrapper.add_access_policy(q, topic.attributes["TopicArn"]) print(f"Added access policies for topic: {topic.attributes['TopicArn']}.") for q in queues: sub = fifo_topic_wrapper.subscribe_queue_to_topic( topic, q.attributes["QueueArn"] ) subscriptions.add(sub) print(f"Subscribed queues to topic: {topic.attributes['TopicArn']}.") input("Press Enter to publish a message to the topic.") message_id = fifo_topic_wrapper.publish_price_update( topic, '{"product": 214, "price": 79.99}', "Consumables" ) print(f"Published price update with message ID: {message_id}.") # Clean up the subscriptions, queues, and topic. input("Press Enter to clean up resources.") for s in subscriptions: sns_wrapper.delete_subscription(s) sns_wrapper.delete_topic(topic) for q in queues: fifo_topic_wrapper.delete_queue(q) print(f"Deleted subscriptions, queues, and topic.") print("Thanks for watching!") print("-" * 88) class FifoTopicWrapper: """Encapsulates Amazon SNS FIFO topic and subscription functions.""" def __init__(self, sns_resource): """ :param sns_resource: A Boto3 Amazon SNS resource. """ self.sns_resource = sns_resource def create_fifo_topic(self, topic_name): """ Create a FIFO topic. Topic names must be made up of only uppercase and lowercase ASCII letters, numbers, underscores, and hyphens, and must be between 1 and 256 characters long. For a FIFO topic, the name must end with the .fifo suffix. :param topic_name: The name for the topic. :return: The new topic. """ try: topic = self.sns_resource.create_topic( Name=topic_name, Attributes={ "FifoTopic": str(True), "ContentBasedDeduplication": str(False), "FifoThroughputScope": "MessageGroup", }, ) logger.info("Created FIFO topic with name=%s.", topic_name) return topic except ClientError as error: logger.exception("Couldn't create topic with name=%s!", topic_name) raise error @staticmethod def add_access_policy(queue, topic_arn): """ Add the necessary access policy to a queue, so it can receive messages from a topic. :param queue: The queue resource. :param topic_arn: The ARN of the topic. :return: None. """ try: queue.set_attributes( Attributes={ "Policy": json.dumps( { "Version":"2012-10-17", "Statement": [ { "Sid": "test-sid", "Effect": "Allow", "Principal": {"AWS": "*"}, "Action": "SQS:SendMessage", "Resource": queue.attributes["QueueArn"], "Condition": { "ArnLike": {"aws:SourceArn": topic_arn} }, } ], } ) } ) logger.info("Added trust policy to the queue.") except ClientError as error: logger.exception("Couldn't add trust policy to the queue!") raise error @staticmethod def subscribe_queue_to_topic(topic, queue_arn): """ Subscribe a queue to a topic. :param topic: The topic resource. :param queue_arn: The ARN of the queue. :return: The subscription resource. """ try: subscription = topic.subscribe( Protocol="sqs", Endpoint=queue_arn, ) logger.info("The queue is subscribed to the topic.") return subscription except ClientError as error: logger.exception("Couldn't subscribe queue to topic!") raise error @staticmethod def publish_price_update(topic, payload, group_id): """ Compose and publish a message that updates the wholesale price. :param topic: The topic to publish to. :param payload: The message to publish. :param group_id: The group ID for the message. :return: The ID of the message. """ try: att_dict = {"business": {"DataType": "String", "StringValue": "wholesale"}} dedup_id = uuid.uuid4() response = topic.publish( Subject="Price Update", Message=payload, MessageAttributes=att_dict, MessageGroupId=group_id, MessageDeduplicationId=str(dedup_id), ) message_id = response["MessageId"] logger.info("Published message to topic %s.", topic.arn) except ClientError as error: logger.exception("Couldn't publish message to topic %s.", topic.arn) raise error return message_id @staticmethod def delete_queue(queue): """ Removes an SQS queue. When run against an AWS account, it can take up to 60 seconds before the queue is actually deleted. :param queue: The queue to delete. :return: None """ try: queue.delete() logger.info("Deleted queue with URL=%s.", queue.url) except ClientError as error: logger.exception("Couldn't delete queue with URL=%s!", queue.url) raise error ``` + 有关 API 详细信息，请参阅《AWS SDK for Python (Boto3) API Reference》**中的以下主题。 + [CreateTopic](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/CreateTopic) + [发布](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Publish) + [Subscribe](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Subscribe) ------ #### [ SAP ABAP ] **适用于 SAP ABAP 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/sns#code-examples)中查找完整示例，了解如何进行设置和运行。 创建 FIFO 主题并为此订阅 Amazon SQS FIFO 队列，然后向 Amazon SNS 主题发布消息。 ``` " Creates a FIFO topic. " DATA lt_tpc_attributes TYPE /aws1/cl_snstopicattrsmap_w=>tt_topicattributesmap. DATA ls_tpc_attributes TYPE /aws1/cl_snstopicattrsmap_w=>ts_topicattributesmap_maprow. ls_tpc_attributes-key = 'FifoTopic'. ls_tpc_attributes-value = NEW /aws1/cl_snstopicattrsmap_w( iv_value = 'true' ). INSERT ls_tpc_attributes INTO TABLE lt_tpc_attributes. TRY. DATA(lo_create_result) = lo_sns->createtopic( iv_name = iv_topic_name it_attributes = lt_tpc_attributes ). DATA(lv_topic_arn) = lo_create_result->get_topicarn( ). ov_topic_arn = lv_topic_arn. " ov_topic_arn is returned for testing purposes. " MESSAGE 'FIFO topic created' TYPE 'I'. CATCH /aws1/cx_snstopiclimitexcdex. MESSAGE 'Unable to create more topics. You have reached the maximum number of topics allowed.' TYPE 'E'. ENDTRY. " Subscribes an endpoint to an Amazon Simple Notification Service (Amazon SNS) topic. " " Only Amazon Simple Queue Service (Amazon SQS) FIFO queues can be subscribed to an SNS FIFO topic. " TRY. DATA(lo_subscribe_result) = lo_sns->subscribe( iv_topicarn = lv_topic_arn iv_protocol = 'sqs' iv_endpoint = iv_queue_arn ). DATA(lv_subscription_arn) = lo_subscribe_result->get_subscriptionarn( ). ov_subscription_arn = lv_subscription_arn. " ov_subscription_arn is returned for testing purposes. " MESSAGE 'SQS queue was subscribed to SNS topic.' TYPE 'I'. CATCH /aws1/cx_snsnotfoundexception. MESSAGE 'Topic does not exist.' TYPE 'E'. CATCH /aws1/cx_snssubscriptionlmte00. MESSAGE 'Unable to create subscriptions. You have reached the maximum number of subscriptions allowed.' TYPE 'E'. ENDTRY. " Publish message to SNS topic. " TRY. DATA lt_msg_attributes TYPE /aws1/cl_snsmessageattrvalue=>tt_messageattributemap. DATA ls_msg_attributes TYPE /aws1/cl_snsmessageattrvalue=>ts_messageattributemap_maprow. ls_msg_attributes-key = 'Importance'. ls_msg_attributes-value = NEW /aws1/cl_snsmessageattrvalue( iv_datatype = 'String' iv_stringvalue = 'High' ). INSERT ls_msg_attributes INTO TABLE lt_msg_attributes. DATA(lo_result) = lo_sns->publish( iv_topicarn = lv_topic_arn iv_message = 'The price of your mobile plan has been increased from $19 to $23' iv_subject = 'Changes to mobile plan' iv_messagegroupid = 'Update-2' iv_messagededuplicationid = 'Update-2.1' it_messageattributes = lt_msg_attributes ). ov_message_id = lo_result->get_messageid( ). " ov_message_id is returned for testing purposes. " MESSAGE 'Message was published to SNS topic.' TYPE 'I'. CATCH /aws1/cx_snsnotfoundexception. MESSAGE 'Topic does not exist.' TYPE 'E'. ENDTRY. ``` + 有关 API 详细信息，请参阅*适用于 SAP ABAP 的AWS SDK 的 API 参考*中的以下主题。 + [CreateTopic](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html) + [发布](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html) + [Subscribe](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html) ------ ### 接收来自 FIFO 订阅的消息 现在，您可以在三个订阅的应用程序中接收价格更新。如中所示[Amazon SNS FIFO 主题示例使用案例](fifo-example-use-case.md)，每个使用者应用程序的入口点都是 Amazon SQS 队列，其相应 AWS Lambda 函数可以自动轮询该队列。当 Amazon SQS 队列是 Lambda 函数的事件源时，Lambda 会根据需要扩展其轮询器队列，以高效地使用消息。 有关更多信息，请参阅*AWS Lambda 开发者*指南中的[AWS Lambda 与 Amazon SQS 配合使用](https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html)。有关编写自己的队列轮询器的信息，请参阅《亚马逊简单队列*服务开发者指南》[ReceiveMessage](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html)和《亚马逊简单队列服务 API 参考》中的 Amazon* [SQS 标准和 FIFO *队列*建议](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-best-practices.html#sqs-standard-fifo-queue-best-practices)。 ## 使用 AWS CloudFormation CloudFormation 允许您使用模板文件将 AWS 资源集合一起创建和配置为一个单元。本部分提供的模板示例，用于创建以下内容： + 分发价格更新的 Amazon SNS FIFO 主题 + 为批发和零售应用程序提供这些更新的 Amazon SQS FIFO 队列 + 用于存储记录的分析应用程序的 Amazon SQS 标准队列，可以查询这些记录以获取商业智能 (BI) + 将三个队列连接到主题的 Amazon SNS FIFO 订阅 + 指定订阅者应用程序的[筛选策略](sns-subscription-filter-policies.md)只接收他们需要的价格更新 **注意** 如果通过向主题发布消息来测试此代码示例，请确保您发布的是带 `business` 属性的消息。为属性值指定 `retail` 或 `wholesale`。否则，消息将被筛选掉，且不会传递到订阅的队列中。 ``` { "AWSTemplateFormatVersion": "2010-09-09", "Resources": { "PriceUpdatesTopic": { "Type": "AWS::SNS::Topic", "Properties": { "TopicName": "PriceUpdatesTopic.fifo", "FifoTopic": true, "ContentBasedDeduplication": false, "ArchivePolicy": { "MessageRetentionPeriod": "30" } } }, "WholesaleQueue": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "WholesaleQueue.fifo", "FifoQueue": true, "ContentBasedDeduplication": false } }, "RetailQueue": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "RetailQueue.fifo", "FifoQueue": true, "ContentBasedDeduplication": false } }, "AnalyticsQueue": { "Type": "AWS::SQS::Queue", "Properties": { "QueueName": "AnalyticsQueue" } }, "WholesaleSubscription": { "Type": "AWS::SNS::Subscription", "Properties": { "TopicArn": { "Ref": "PriceUpdatesTopic" }, "Endpoint": { "Fn::GetAtt": [ "WholesaleQueue", "Arn" ] }, "Protocol": "sqs", "RawMessageDelivery": "false", "FilterPolicyScope": "MessageBody", "FilterPolicy": { "business": [ "wholesale" ] } } }, "RetailSubscription": { "Type": "AWS::SNS::Subscription", "Properties": { "TopicArn": { "Ref": "PriceUpdatesTopic" }, "Endpoint": { "Fn::GetAtt": [ "RetailQueue", "Arn" ] }, "Protocol": "sqs", "RawMessageDelivery": "false", "FilterPolicyScope": "MessageBody", "FilterPolicy": { "business": [ "retail" ] } } }, "AnalyticsSubscription": { "Type": "AWS::SNS::Subscription", "Properties": { "TopicArn": { "Ref": "PriceUpdatesTopic" }, "Endpoint": { "Fn::GetAtt": [ "AnalyticsQueue", "Arn" ] }, "Protocol": "sqs", "RawMessageDelivery": "false" } }, "SalesQueuesPolicy": { "Type": "AWS::SQS::QueuePolicy", "Properties": { "PolicyDocument": { "Statement": [ { "Effect": "Allow", "Principal": { "Service": "sns.amazonaws.com" }, "Action": [ "sqs:SendMessage" ], "Resource": "*", "Condition": { "ArnEquals": { "aws:SourceArn": { "Ref": "PriceUpdatesTopic" } } } } ] }, "Queues": [ { "Ref": "WholesaleQueue" }, { "Ref": "RetailQueue" }, { "Ref": "AnalyticsQueue" } ] } } } } ``` 有关使用 CloudFormation 模板部署 AWS 资源的更多信息，请参阅《*CloudFormation 用户指南》*中的 “[入门](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/GettingStarted.Walkthrough.html)”。