将死信队列中的所有消息移回订阅的主队列

Posted 2023-03-10

【中文标题】将死信队列中的所有消息移回订阅的主队列

【英文标题】：Move all messages from deadletter queue back into main queue of subscription

【发布时间】：2018-09-04 21:20:12

【问题描述】：

我的服务使用来自 Azure 服务总线订阅的消息。我的服务的依赖关系关闭了一段时间，这导致大量消息最终进入死信队列 (DLQ)。现在服务已备份，我想重新处理来自 DLQ 的所有消息。如何将 DLQ 中的所有消息移动/重新提交回主队列。

限制：

这是数千条消息，因此手动处理它们是不可行的。 该主题有大约十个订阅。我不想向主题重新提交消息，因为这样所有订阅都会收到消息，导致双重处理。 我不想直接针对 DLQ 运行服务，因为某些消息被破坏并导致永久性错误，即它们会再次出现在 DLQ 中，这将导致无限循环。此外，损坏的消息被放回队列的前面，有效地使损坏的消息之后的健康消息挨饿。

【参考方案1】：

Thomas，您可能已经找到了答案，因为这是很久以前的事了。将 DLQ（或您拥有的任何现有队列）视为另一个集合变量，就像在 PC 应用程序中一样，但驻留在云中。就像您的工具包中的 PC 应用程序或内存中的集合变量一样，您可以通过多种方式使用它。当然，这两种类型的集合变量之间存在限制和差异，但是通过了解这些限制和差异，您可以设计解决方案，就好像 DLQ 只是另一个集合变量一样。

对于某些排队实现，其中一种解决方案是让同一应用程序的另一个实例指向 DLQ，但具有相当长的可见性超时（例如 6 或 12 甚至 24 小时，具体取决于您的 SLA），因为你不想经常重复它们。但是，这不适用于 Azure 服务总线，因为它将可见性超时限制为最多 5 分钟。

如果 DLQ 包含损坏的不可恢复的作业，您应该修复应用程序以在发生未知异常时根据错误消息将其删除。部署修复程序后，您的应用程序将删除此类损坏的不可恢复作业，并且一开始就不会发送到 DLQ。而那些已经在DLQ中的将被固定的应用程序删除。

【参考方案2】：

我意识到这是在原始帖子之后的一段时间，但如果其他人偶然发现这个问题，Service Bus Explorer 中有一个相当方便的解决方案（我发现它对于 ASB 开发非常方便）。

连接到您的服务总线并找到所需的命名空间后，找到所需的 topic 和 subscription 以及其中的死信。从那里Right Click 和Receive Deadletter Queue Messages 然后点击确定。

从那里，突出显示您要发送回主队列的内容，然后点击Resubmit Selected Messages in Batch Mode。

【参考方案3】：

重放 DLQ 消息的唯一选择是从 DLQ 接收它们，创建具有相同内容的新消息并再次将其发送到主题。它们将在订阅队列的末尾结束。

您不能直接向订阅发送消息。有一个技巧可以将元数据属性添加到消息中，然后调整除一个订阅之外的所有内容以过滤掉此类消息。由您决定它是否会对您的方案有所帮助。

至于工具，我们总是使用自定义代码来做这件事，因为我们总是需要做一些额外的工作，比如记录每条重播的消息以供进一步分析。

【讨论】：

根据设计不能将消息移回主队列是否有充分的理由？我希望这个用例相当普遍。

@ThS 好问题，但我从未见过任何官方动机。我想这只是“设计使然”。

我的假设是您不想无限期地重试消息。如果您尝试调用不可用的服务，您希望等待该服务再次可用，然后重试。因此，如果您需要更多重试或实施断路器，您可以增加max delivery count？

实际上，当服务重新运行时，我想手动按下一个按钮，将所有消息从 DLQ 移回主队列。手动执行此操作可防止无限重试。增加max delivery count 是行不通的，因为如果服务长时间宕机，消息仍然会在DLQ中结束。

【参考方案4】：

快速回答是，您不能直接将消息移回订阅的主队列。这是 Microsoft 如何实现其主题和订阅的设计。

选项 #1 可以选择使用 Azure 服务总线主题过滤器 https://docs.microsoft.com/en-us/azure/service-bus-messaging/topic-filters 并以仅允许在目标订阅上接收它们的方式定义/标记您的消息。

选项 #2 另一种选择是更改您当前的实现。您将设置“传递队列”（常规服务总线队列）并配置每个相应的订阅以自动将其消息转发到这些传递队列。然后，您的消息处理逻辑将侦听这些“交付队列”与订阅。任何失败都会在这些关联的“交付队列”上产生 DLQ 消息，然后可以在主题/订阅之外进行处理。