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

Question

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

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

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

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

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

我们知道，如果出现以下情况，消费者无法处理作业:

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

它从 *job_queue* 获得了一份工作，然后死了

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

Answer 1

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

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

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

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