是否有用于提交/回滚消息处理的 amqp 架构的设计模式？答案

【问题标题】：is there a design pattern for an amqp architecture for commit/rollback message handling?是否有用于提交/回滚消息处理的 amqp 架构的设计模式？
【发布时间】：2011-10-14 10:50:20
【问题描述】：

我有一个简单的生产者/消费者 amqp 设置如下：

producer -> e1:jobs_queue -> consumer -> e2:results_queue -> result_handler

生产者发送一些作业。消费者一次拉下一个作业，并处理它们，将结果推送到另一个队列中。然后将这些结果由将结果发布到数据库的 result_handler 提取出来。

有时消费者会失败 - 它可能会被操作系统杀死或引发异常。如果在处理消息时发生这种情况，则该消息会丢失，不会产生相应的结果，我很难过。如果失败的作业重新排队，我会再次高兴的。

我正在寻找的是一种设计模式，用于确保 consumer 处理作业以完成并将相应的结果放入 *results_queue*，或者如果失败则将作业放回进入 *jobs_queue*。由于 consumer 是失败的原因，consumer 不应负责管理与其自身监管相关的任何消息。

我们知道消费者在以下情况下未能处理工作：

它从 *job_queue* 获取了一个作业，并且在超时后没有产生任何结果
它从 *job_queue* 获取了一份工作，然后死了

对于我的应用程序，我们可以通过简单地等待作业处理超时来捕获第二种情况。在生产中，将有许多工人要监督，所有工人都从一个共同的工作列表中提取工作并将结果放入一个单一的结果交换/队列中。

【问题讨论】：

你是自动确认消息吗？假设您可以关闭自动确认，然后在作业成功处理后确认，从而让 Rabbit 负责重新排队消息。
有几种模式用于可靠消息传递、持久消息队列、确认传递、两阶段确认（接收和保存/转发）和其他几种模式。如果不详细了解您的平台和要求，这不是一个容易回答的问题。

标签： design-patterns rabbitmq amqp

【解决方案1】：

实现您想要的最简单的方法是手动处理收到的消息的确认。在node-amqp 中，只需将选项{ ack: true } 添加到queue.subscribe 调用中即可。然后，您可以通过调用队列上的某些函数来确认消息。如果是node-amqp，则为queue.shift()。

您还可以使用prefetchCount 设置允许消费者使用的尚未确认消息的数量。

如果消费者断开连接，任何未确认的消息现在都将被重新传递（给任何连接的消费者）。

通过还将队列设置为durable 和autoDelete: false，您还可以确保队列（及其上的消息）不会在您的MQ 服务器重新启动或最后一个消费者断开连接时被删除。

【讨论】：