amazon-web-services - 如何防止重复的 SQS 消息？

Question

在 Amazon SQS 中防止重复消息的最佳方法是什么？我有一个等待抓取的域的 SQS。在向 SQS 添加新域之前，我可以检查保存的数据以查看它最近是否已被抓取，以防止重复。

问题在于尚未爬网的域。例如，如果队列中有 1000 个未爬网的域。这些链接中的任何一个都可以一次又一次地添加。这使我的 SQS 膨胀到数十万条消息，其中大部分是重复的。

我如何防止这种情况？有没有办法从队列中删除所有重复项？或者有没有办法在添加消息之前在队列中搜索消息？我觉得这是任何拥有 SQS 的人都必须经历过的问题。

我可以看到的一种选择是，是否在将域添加到 SQS 之前存储一些数据。但是，如果我必须将数据存储两次，那么首先就破坏了使用 SQS 的意义。

Answer 1

正如其他答案所提到的，您无法阻止来自 SQS 的重复消息。

大多数情况下，您的消息将被传递给您的一位消费者一次，但您将 在某个阶段遇到重复。

我认为这个问题没有简单的答案，因为它需要提出一个可以处理重复的适当架构，这意味着它本质上是幂等的。

如果分布式架构中的所有工作人员都是幂等的，那将很容易，因为您无需担心重复。但在现实中，这种环境并不存在，一路上的某个地方将无法处理它。

我目前正在做一个项目，需要我解决这个问题，并想出一种方法来处理它。我认为在这里分享我的想法可能会让其他人受益。这可能是获得一些关于我的想法的反馈的好地方。

实体店

开发服务是一个很好的主意，以便它们收集理论上可以重放的事实，以在所有受影响的下游系统中重现相同的状态。

例如，假设您正在为股票交易平台构建消息代理。 (我之前确实做过这样的项目，很可怕，但也是一次很好的学习经历。)

现在假设交易进来了，并且有 3 个系统对其感兴趣:

需要保持更新的老式大型机

整理所有交易并在 FTP 服务器上与合作伙伴共享的系统

记录交易并将股份重新分配给新所有者的服务

我知道这有点令人费解，但这个想法是传入的一个消息(事实)具有各种分布式下游效应。

现在让我们假设我们维护一个事实存储，记录进入我们经纪人的所有交易。并且所有 3 个下游服务所有者都调用我们，告诉我们他们丢失了过去 3 天的所有数据。 FTP下载延迟3天，主机延迟3天，所有交易延迟3天。

因为我们有事实存储，所以理论上我们可以从某个时间到某个时间重放所有这些消息。在我们的示例中，从 3 天前到现在。下游服务可能会被 catch 。

这个例子可能看起来有点夸张，但我试图传达一些非常特别的东西:事实是需要跟踪的重要事情，因为我们将在我们的架构中使用它来对抗重复。

事实存储如何帮助我们处理重复消息

如果您在持久层上实现事实存储，该持久层为您提供 CAP theorem 的 CA 部分、一致性和可用性，您可以执行以下操作:

一旦从队列中接收到消息，您就可以在事实存储中检查您之前是否已经看到过此消息，如果已经看到，则它目前是否被锁定以及处于挂起状态。就我而言，我将使用 MongoDB 来实现我的事实存储，因为我对它非常满意，但其他各种数据库技术应该能够处理这个问题。

如果事实还不存在，它会被插入到事实存储中，并带有挂起状态和锁定到期时间。这应该使用原子操作来完成，因为您不希望这种情况发生两次!在这里您可以确保您的服务 idempotence .

快乐案例 - 大部分时间都会发生

当事实存储返回到您的服务，告诉它事实不存在并且已创建锁时，该服务会尝试执行它的工作。一旦完成，它 删除 SQS 消息 ，并将事实标记为已完成。

重复消息

所以这就是当一条消息通过并且它不是重复时会发生的情况。但是让我们看看什么时候有重复的消息进来。服务把它捡起来，并要求事实存储用锁记录它。事实存储告诉它它已经存在，并且它已被锁定。该服务忽略该消息并跳过它!消息处理完成后，其他工作人员将从队列中删除此消息，我们将不会再看到它。

灾难案例 - 很少发生

那么当一个服务第一次在 store 中记录这个事实，然后获得了一段时间的锁，但是失败了会发生什么？那么 SQS 会再次向您展示一条消息，如果它被捡起，但在从队列中提供服务后的一段时间内没有被删除。这就是为什么我们对事实存储进行编码，以便服务在有限的时间内保持锁定。因为如果它失败了，我们希望 SQS 将消息呈现给服务，或者稍后的另一个实例，允许该服务假设该事实应该再次合并到状态(执行)中。