【发布时间】：2011-04-15 02:05:45
【问题描述】：
我在 RabbitMQ 上有一个类似于作业队列的东西，在请求取消作业时，我想撤回尚未开始处理的任务（它们的消息尚未被确认），这对应于从它们被路由到的队列中收回这些消息。
我没有在 AMQP 或 RabbitMQ API 中找到这个功能；也许我搜索得不够好？还是我必须使用解决方法（这并不难，但仍然如此）？
【问题讨论】：
标签： message-queue rabbitmq amqp cancellation
【发布时间】：2011-04-15 02:05:45
【问题描述】：
至少有两种方法可以实现您的目标：
如果设置了
requeue=true，basic.reject将重新排队消息（否则它将拒绝消息）。
（自 RabbitMQ 2.0.0 起受支持；请参阅 http://www.rabbitmq.com/blog/2010/08/03/well-ill-let-you-go-basicreject-in-rabbitmq/）。basic.recover将要求代理在通道上重新传递未确认的消息。
【讨论】：
-
不，这解决了另一个问题。我不需要在消费者端拒绝消息，我想从生产者端取消它的传递，这样消息就不会像它根本不存在一样到达消费者。在我的问题中，消费者无法决定是否应该拒绝消息。
-
如果消息成功发布并且您需要从已知队列中删除它们，然后订阅它并在生产者中使用它们。
RabbitMQ 不允许您在消息入队后修改或删除消息。为此，您需要某种数据库来保存每个作业的状态，并使用 RabbitMQ 通知相关方该状态的变化。
对于低容量，您可以将它与每个作业的队列组合在一起。创建队列，将工作描述发布到队列，向工作人员宣布队列的名称。如果作业需要在处理之前取消，删除作业的队列；当工人来获取职位描述时，他们会注意到队列已经消失了。
更轻量级且通常更好的是使用 redis 或其他键/值存储来保存作业状态（删除或缺失记录意味着已取消或不存在的作业）并使用 rabbitmq 通知新/删除/更改的记录在键/值存储中。
【讨论】：
您需要订阅所有已路由消息的队列，并使用 ack 消费它们。
例如，如果您使用“test”作为路由键发布到主题交换，并且有 3 个订阅“test”的持久队列，您将需要使用这三个队列。最好添加另一个您的消费者进程也会监听的队列，并告诉他们忽略这些消息。
由于您使用的是 RabbitMQ，因此另一种选择是编写一个自定义交换插件，该插件将接受一些带外指令以清除所有队列。例如，您可能让该交换读取一个特殊的消息头，告诉它清除该消息的目的地的所有队列。这确实需要编写 Erlang 代码，但是实现了 4 种不同的交换类型，因此您只需复制最相似的一种并为新的行为编写代码。如果您只为此使用自定义标头，则消息的正文可以是消费者的正常消息。
总结一下：
1) 发布者需要自己消费消息 2）发布者可以在特殊队列中发送特殊消息，告诉消费者忽略该消息 3) 发布者可以向自定义交换器发送一条特殊消息，该交换器将在将此特殊消息发送给消费者之前清除队列中的所有现有消息。
【讨论】：
我会通过让工作人员检查某种权威数据源来确定工作是否应该继续来解决这种情况。例如，工作人员会在数据库中检查作业的状态，以查看作业是否已被取消。
对于处理作业的速度可能比更新和读取权威存储的速度更快的情况，以速度换取其他特性的保证较少的数据存储可能会很有用。
这方面的一个例子是使用 Redis 作为取消处理消息的存储，而不是像 MySQL 这样的关系数据库。 Redis 非常快，但对其所持有的数据提供的保证较少，而 MySQL 则慢得多，但对其所持有的数据提供更多的保证。
最后，与另一个来源检查是否处理消息的概念是相同的，但您实现的方式取决于您的特定场景。
【讨论】：