Прочитайте сообщение только счетное количество раз в Spring AMQP

У меня есть приложение, которое получает сообщение от RabbitMQ. После того, как сообщение будет использовано и возникнет какая-либо проблема. Я установил setDefaultRequeueRejected на true, и он снова ставится в очередь.

Теперь у меня есть сценарий: если сообщение вернется в очередь, оно снова будет использовано и снова поставлено в очередь. Это имеет форму своего рода цикла.

Теперь мне нужно использовать повторно поставленное в очередь сообщение только дважды (поддерживая тип счетчика), и после этого я отправлю его в очередь ErrorQueue/DeadLetter. Как поддерживать такой счетчик?

С уважением,

Чандан


rabbitmq spring-rabbit spring-amqp
person Chandan    schedule 17.09.2016    source источник

Ответы (1)


arrow_upward
0
arrow_downward

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

Если отправитель устанавливает заголовок messageId, вы можете использовать повторную попытку с отслеживанием состояния. перехватчик с, и платформа будет отслеживать, сколько раз было доставлено одно и то же сообщение. При использовании RejectAndDontRequeueRecoverer сообщение будет отправлено в DLQ (если настроено) после исчерпания повторных попыток.

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

person Gary Russell    schedule 17.09.2016
comment
Спасибо Гэри. Мне интересно узнать больше о MessageKeyGenerator . Не могли бы вы предоставить мне пример реализации/пример. Это мне очень поможет :) - person Chandan; 17.09.2016
comment
Это простой интерфейс: Object getKey(Message message) вы получаете ссылку на сообщение и возвращаете значение, которое однозначно идентифицирует сообщение (например, из заголовка или тела сообщения). Если генератора нет, используем свойство messageId — которое должно быть задано отправителем, и уникальное. Сообщение является необработанным, до какого-либо преобразования. - person Gary Russell; 17.09.2016
comment
Я все еще не получил ясности. Не могли бы вы перенаправить меня на пример реализации? - person Chandan; 17.09.2016
comment
Образцов нет — это очень простая концепция — в сообщении должно быть что-то, что однозначно его идентифицирует. Если вы не можете этого сделать, вам не повезло. Самое простое решение — изменить отправителя, чтобы установить свойство messageId с уникальным значением, например, свойство UUID.randomUUID.toString(). If the sender is a Spring AMQP application make sure the createMessageIds имеет значение true в преобразователе сообщений RabbitTemplate, и в каждом сообщении будет уникальный идентификатор сообщения. - person Gary Russell; 17.09.2016
comment
Да, я получил это. Но каждый раз, когда потребитель потребляет сообщение, как я могу поддерживать счетчик, как будто я получаю это сообщение во второй раз, чтобы я его сбросил? - person Chandan; 17.09.2016
comment
См. документацию, которую я вам тоже указал - перехватчик повторных попыток позаботится обо всем этом. После исчерпания повторных попыток за дело берется восстановитель. По умолчанию он просто регистрируется; RejectAndDontRequeueRecoverer приведет к тому, что сообщение будет отклонено и не помещено в очередь повторно (может быть отправлено в очередь недоставленных сообщений, если это настроено), или RepublishMessageRecoverer запишет ошибочное сообщение в другую очередь. - person Gary Russell; 17.09.2016