queue - 在 SQS 队列中使用多个消费者

Question

我知道可以使用多个线程来使用 SQS 队列。我想保证每条消息都会被消费一次。我知道可以更改消息的可见性超时，例如，等于我的处理时间。如果我的进程花费的时间比可见性超时(例如慢速连接)多，则其他线程可以使用相同的消息。

保证消息将被处理一次的最佳方法是什么？

Answer 1

What is the best approach to guarantee that a message will be processed once?

您要求保证 - 你不会得到一个 .您可以将消息被多次处理的概率降低到 极少量 ，但你不会得到保证。

我将解释原因，以及减少重复的策略。

重复从何而来

当您将消息放入 SQS 时，SQS 实际上可能会多次收到该消息

例如:发送消息时的轻微网络故障导致自动重试的瞬时错误 - 从消息发送者的角度来看，它失败了一次，成功发送了一次，但 SQS 收到了这两条消息。

SQS can internally generate duplicates

与第一个示例类似 - 有很多计算机在幕后处理消息，并且 SQS 需要确保没有任何东西丢失 - 消息存储在多个服务器上，这可能会导致重复。

大多数情况下，通过利用 SQS message visibility timeout ，从这些来源复制的机会已经很小 - 就像百分之一小。

如果处理重复真的没那么糟糕( strive to make your message consumption idempotent !)，我会认为这已经足够了 - 进一步减少重复的机会很复杂而且可能很昂贵......

您的应用程序可以做什么来进一步减少重复？

好的，在这里我们进入兔子洞......在高层次上，您需要为您的消息分配唯一的 id，并在开始处理之前检查正在进行或已完成的 id 的原子缓存:

确保您的消息具有在插入时提供的唯一标识符

没有这个，您将无法区分重复项。

处理消息“行尾”的重复。

如果您的消息接收者需要将消息发送到箱外进行进一步处理，那么它可能是另一个重复来源(出于与上述类似的原因)

您需要在某个地方以原子方式存储和检查这些唯一 ID(并在超时后刷新它们)。有两个重要的状态:“InProgress”和“Completed”

InProgress 条目应根据处理失败时需要恢复的速度设置超时。

已完成的条目应根据您希望重复数据删除窗口的时间设置超时

最简单的可能是 Guava cache ，但仅适用于单个处理应用程序。如果你有大量的消息或者分布式消费，可以考虑为这个工作建立一个数据库(有一个后台进程来清除过期条目)

在处理消息之前，尝试将 messageId 存储在“InProgress”中。如果它已经存在，请停止-您只是处理了重复项。

检查消息是否“已完成”(如果存在则停止)

您的线程现在对该 messageId 拥有独占锁定 - 处理您的消息

将 messageId 标记为“已完成” - 只要此 messageId 保留在此处，您就不会处理该 messageId 的任何重复项.

不过，您可能负担不起无限存储空间。

从“InProgress”中删除 messageId(或者让它从这里过期)

一些笔记

请记住，在没有所有这些的情况下重复的机会已经很低了。根据消息重复数据删除的时间和金钱对您的值(value)，您可以随意跳过或修改任何步骤

例如，您可以省略“InProgress”，但这会增加两个线程同时处理重复消息的可能性很小(第二个线程在第一个“完成”之前开始)

只要您可以将 messageIds 保持在“已完成”状态，您的重复数据删除窗口就足够了。由于您可能负担不起无限存储，因此至少要使其持续时间为 SQS 消息可见性超时的 2 倍；之后重复的机会减少了(除了已经非常低的机会，但仍然不能保证)。

即便如此，仍有重复的机会 - 所有的预防措施和 SQS 消息可见性超时有助于将这种机会减少到很小，但机会仍然存在:

您的应用程序可能会在处理消息后立即崩溃/挂起/执行很长时间的 GC，但在 messageId 为“已完成”之前(可能您正在使用数据库进行此存储并且与它的连接已关闭)

在这种情况下，“Processing”最终将过期，另一个线程可以处理此消息(在 SQS 可见性超时也过期之后或因为 SQS 中有重复项)。