防止重复消息答案

【问题标题】：Prevent duplicate messages防止重复消息
【发布时间】：2022-02-06 22:52:15
【问题描述】：

技术栈 - Python、Rabbitmq、Elasticsearch、Heroku

我有一个应用程序，它根据某个时间表在应用程序中添加内容，当添加内容时，需要根据某些条件向某些用户触发电子邮件（可能是大约 100 万用户）

我决定的设计是第一个应用程序（生产者）将发布内容，然后在 Rabbitmq（消息代理）中放置一条消息，其中包含每个应该收到电子邮件的用户的用户 ID 和电子邮件地址，所以基本上添加大约 100 万条消息到 Rabbitmq，然后电子邮件将通过电子邮件应用程序（消费者）发送

我的问题是关于生产者应用程序的。它将从 elasticsearch 获取用户 ID 列表并开始使用 for 循环将它们添加到队列中，但如果由于某种原因此应用程序出现故障（例如在新的情况下）部署），那么只有少数用户将被添加到队列中，当应用程序返回时，它将再次开始排队，我们最终可能会收到重复的电子邮件给相同的用户。有没有一种标准方法可以避免这种情况。似乎是一个非常常见的问题。

我正在考虑在我的生产者应用程序数据库中维护来自消费者应用程序的消息确认（针对给定内容的每个用户）。但感觉这可能会在发送几封电子邮件后导致我的 PostgreSQL 数据库中的记录爆炸。每个内容可以通过电子邮件触发约 100 万用户。

【问题讨论】：

标签： architecture rabbitmq microservices message-queue

【解决方案1】：

你说得对，这是一个非常普遍的问题。不幸的是，一般来说，当在异步执行的进程之间发送消息时，您不能保证消息只发送一次：您必须在最多一次和至少一次之间进行选择。

您可以通过让生产者应用程序从不重试发送（即使生产者失败）来保证最多一次（这很容易防止重复）。然而，这可能不是你想要的。

您可以通过让消费者向生产者确认它已收到并根据发送的消息采取行动来保证至少一次；生产者维护给定消息尚未被确认的状态（例如在数据库中），并在一段时间后重试未确认的消息。请注意，如果消费者在执行操作后确认，则操作（即正在发送的电子邮件）可能至少部分重复（考虑如果消费者在开始做某事和确认已完成之间崩溃会发生什么），如果消费者在执行操作之前确认，您已将其变为最多一次交付（考虑消费者在确认后但在完全执行操作之前崩溃的情况）。

通过让消费者维护包含其已确认消息的状态，消费者可以对消息进行重复数据删除：如果它收到一条已经确认的消息，它会再次确认它并且不做任何其他事情。它仍然至少是一次，但是重复率大大降低（以一定的代价：现在每条消息都可能需要 3 次 DB 写入（生产者写入未确认的消息，消费者和生产者每个写入确认），尽管这些不需要是相同的数据库。

【讨论】：

【解决方案2】：

您肯定希望实现完全一致性，并且您可能需要使用支持同步获取/选择的数据库（持久存储）扩展您的堆栈。

作为“SWC-DB”的开发人员，我欢迎您试用最新的SWC-DB 版本解决了#1 和#2 的问题，使一致性成为可能。有一个priority-queue的例子queue-example.cc（其他语言可以用具体的thrift-client实现）您可能需要注意“queue-example.cc”的扫描规范中的.set_opt__deleting()：

  SWC::DB::Specs::Interval intval;
  intval.set_opt__deleting();

虽然，我可以看到“重复电子邮件”的其他问题，当电子邮件已经发送并且新任务到达时也会发生这种问题。正如@levi-ramsey 提到的那样，用于跟踪消费者的处理状态。在这种情况下，SELECT 可以并且应该使用UPDATE=(ns+3d,"SET_A_VALUE_AS_PROCESSED") 规范，并且在这种情况下（数据库中保留了许多记录），可以使用 TTL 设置列的架构（比如说 3 天，所以尚未处理的也不会被删除）。鉴于，间隔扫描规范要求使用除 (-ne)“SET_A_VALUE_AS_PROCESSED”以外的值的规范用于选择匹配记录。有SWC-DB Condition-Value-Expression syntax 解释了 SQL 的用法。

【讨论】：