Как установить несколько попыток повтора в RabbitMQ?

Я использую RabbitMQ, и у меня есть очередь, в которой хранятся сообщения электронной почты. Мой потребительский сервис отключает сообщения и пытается их отправить. Если по какой-либо причине мой потребитель не может отправить сообщение, я бы хотел повторно отправить сообщение для отправки. Я понимаю, что могу сделать basicNack и установить флаг Requeue как истинный, однако я не хочу запрашивать сообщение бесконечно (скажем, если наша система электронной почты опускается, я не хочу постоянно запрашивать неотправленные сообщения). Я хотел бы определить конечное число раз, чтобы я мог повторно отправить сообщение для отправки. Я не могу установить поле в объекте сообщения электронной почты, однако, когда я удаляю его и отправляю нюк. Обновленное поле отсутствует в сообщении в очереди. Есть ли другой способ, с помощью которого я могу подойти к этому? Спасибо заранее.

Ответ 1

Нет такой функции, как попытки повтора в RabbitMQ (а также в протоколе AMQP).

Возможное решение для реализации поведения попыток попыток попыток:

Сообщение Redeliver, если оно ранее не было добавлено (проверьте параметр redelivered на basic.deliver) - ваша библиотека должна иметь некоторый интерфейс для этого) и отпустите его, а затем поймайте в обмен мертвыми буквами, затем каким-то образом обработайте.
Каждый раз, когда сообщение не может быть обработано, опубликуйте его снова, но укажите поле заголовка или приращения/уменьшения, скажем x-redelivered-count (вы можете выбрать любое имя, которое вам нравится). Чтобы получить контроль над redeliveries, в этом случае вам нужно проверить поле, которое вы установили, достигнет ли он определенного предела (верхний или нижний - 0 - мой выбор, a-la ttl в заголовке ip из tcp/ip).
Сохранить уникальный ключ сообщения (например, uuid, но вы должны установить его вручную при публикации сообщения) в Redis, memcache или другом хранилище, даже в mysql наряду с подсчетом повторных отправлений, а затем по каждому приращению/декрету добавления значение, пока оно не достигнет предела.
(для реальных geeks) напишите плагин, который будет реализовывать такое поведение, как вы хотите.

Про # 3 заключается в том, что сообщение redelivered остается в очереди. Это важно, если у вас длинная очередь или если порядок сообщений важен для вас (обратите внимание, что redeliveries нарушают строгий порядок сообщений, см. Официальные документы для деталей или этот вопрос на SO).

P.S.:

В этом разделе есть аналогичный ответ, но в php. Посмотрите, возможно, это поможет вам немного (начните читать его со слов "Существует несколько методов решения проблемы повторного цикла".

Ответ 2

Хотя это старый вопрос, я думаю, что теперь вы можете легко сделать это с помощью комбинации обмена мертвыми буквами и массива заголовков x-death, добавленных после сообщения с мертвой буквой:

Процесс с разбором букв добавляет массив к заголовку каждого сообщения с заглавными буквами с именем x-death. Этот массив содержит запись о каждом событии с недействительными буквами, идентифицированную парой {очередь, причина}. Каждая такая запись представляет собой таблицу, состоящую из нескольких полей:

queue: имя очереди, в которой было сообщение до того, как оно было буквально

причина: причина мертвой надписи, см. ниже

время: дата и время, когда сообщение было полностью помечено как 64-битная временная метка AMQP 0-9-1

exchange - обмен, к которому было опубликовано сообщение (обратите внимание, что это будет обмен недоставленными буквами, если сообщение несколько раз было буквально)

ключи маршрутизации: ключи маршрутизации (включая ключи CC, но исключая ключи BCC), с которыми было опубликовано сообщение

count: сколько раз это сообщение было буквально в этой очереди

оригинал-истечение срока действия (если сообщение было бездействующим из-за TTL для каждого сообщения): исходное свойство истечения срока действия сообщения. Свойство expiration удаляется из сообщения о недоставленных письмах, чтобы предотвратить его повторное истечение в любых очередях, в которые оно направляется.

PS: вы также можете использовать поле тега доставки

Прочитайте эту замечательную статью для получения дополнительной информации

проверьте этот розыгрыш:

Ответ 3

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

Ниже вы можете найти прототип обмена мертвыми буквами, реализованный с помощью node-amqp и Rabbitmq https://github.com/kharandziuk/dead-letter-exchange-prototype.