Как отозвать сообщение в RabbitMQ?

Question

У меня есть что-то вроде очереди заданий через RabbitMQ, и по запросу об отмене задания я бы хотел отозвать задания, которые еще не начали обрабатываться (их сообщения не были подтверждены), чтосоответствует возврату этих сообщений из очередей, в которые они были перенаправлены.

Я не нашел этой функции в AMQP или в API RabbitMQ;возможно я не достаточно хорошо искал?Или мне придется использовать обходной путь (это не сложно, но все же)?

Answer 1

Я бы решил этот сценарий, попросив работника проверить какой-нибудь авторитетный источник данных, чтобы определить, следует ли выполнять задание или нет.Например, работник проверяет состояние задания в базе данных, чтобы узнать, отменено ли уже задание.

Для сценариев, в которых скорость обработки заданий может быть выше, чем скорость, с которой может быть обновлено хранилище.и читать, менее гарантированное хранилище данных, которое меняет скорость на другие характеристики, может быть полезным.

Примером этого может быть использование Redis в качестве хранилища для отмены обработки сообщения вместо реляционной БД, такой как MySQL.Redis очень быстр, но дает меньше гарантий относительно данных, которые он содержит, тогда как MySQL намного медленнее, но предлагает больше гарантий относительно данных, которые он хранит.

В конце концов, концепция проверки с другим источником на предметили не обрабатывать сообщение - то же самое, но способ реализации зависит от вашего конкретного сценария.

Answer 2

RabbitMQ не позволяет вам изменять или удалять сообщения после того, как они были помещены в очередь. Для этого вы хотите, чтобы какая-то база данных содержала состояние каждой работы и использовала RabbitMQ для уведомления заинтересованных сторон об изменениях в этом состоянии.

Для слабых томов вы можете класть его вместе с очередью на задание. Создайте очередь, опубликуйте описание задания в очереди, объявите имя очереди работникам. Если задание необходимо отменить до его обработки, удалите очередь задания; когда рабочие придут за описанием работы, они заметят, что очередь исчезла.

Облегченный и, как правило, лучше использовать redis или другое хранилище ключей / значений для хранения состояния задания (с удаленной или отсутствующей записью, означающей отмененную или несуществующую работу) и использовать rabbitmq для уведомления о новых / удаленных / измененных записях. в хранилище ключей / значений.

Answer 3

Как минимум два способа достижения цели:

• basic.reject запросит сообщение, если установлено requeue=true (в противном случае сообщение будет отклонено).
  (поддерживается сRabbitMQ 2.0.0; см. http://www.rabbitmq.com/blog/2010/08/03/well-ill-let-you-go-basicreject-in-rabbitmq/).

• basic.recover и попросит брокера повторно доставить неиспользованные сообщения на канал.

Answer 4

Вам необходимо подписаться на все очереди, в которые были направлены сообщения, и использовать их с помощью ack.

Например, если вы публикуете в обмене темами с "test" в качестве ключа маршрутизации, и там3 постоянные очереди, которые подписываются на «тест», вам нужно будет использовать эти три очереди.Возможно, было бы лучше добавить еще одну очередь, которую также будут слушать ваши потребительские процессы, и попросить их игнорировать эти сообщения.

Альтернативой, поскольку вы используете RabbitMQ, является написание пользовательского плагина обмена, который будет приниматьнекоторые внеплановые инструкции для очистки всех очередей.Например, этот обмен может прочитать специальный заголовок сообщения, который говорит ему очистить все очереди, которым предназначено это сообщение.Это требует написания кода Erlang, но реализовано 4 различных типа обмена, поэтому вам нужно будет только скопировать наиболее похожий и написать код для новых действий.Если вы используете для этого только пользовательские заголовки, то тело сообщения может быть обычным сообщением для потребителей.

Подводя итог:

1) издатель должен сам принимать сообщения.2) издатель может отправить специальное сообщение в специальной очереди, чтобы сказать потребителям игнорировать сообщение. 3) издатель может отправить специальное сообщение на пользовательский обмен, который удалит все существующие сообщения из очередей перед отправкой этого специального сообщения потребителям.