Rabbit listener error handler

RabbitMQ Spring Boot #05-error handling 1/2 — basic concepts

Hi! In this story I’m going to cover error handling strategies for RabbitMq consumers, and I’ll show You some examples using Spring Boot. In my opinion this is a really important aspect , and I strongly recommend to gain some knowledge about RabbitMQ error handling behaviour before using this message broker on production 🙂

Default behaviour on NACK- requeue the message

By default, if consumer will return a negative acknowledgement of a message, the message is going to be requeued back to the queue. Reasons of negative acknowledgement might be different, e.g.

• consumer application might shut down in a middle of processing
• consumer application might have a temporal problem — it will be working again in a few minutes, but right now it won’t be able to process any new messages
• business requirements haven’t been met, and consumer can’t accept the received message

Is this a good error handling strategy? As usually, it depends:) If the problem is temporal on consumer side (let’s say that consumer application will restart itself and it’ll be working again in a few seconds/minutes), then it‘s not the worst case scenario. However, if consumer’s application issue needs more time to be solved, or the message is never going to be accepted, then this strategy is terrible, because it will cause infinite loop on a consumer side.

Infinite loop demo

Message is going to be requeued on NACK by default, so to present this behaviour there is no need to write any extra error handling configuration.

Flow is very simple:

1. Producer produces the message and sends it to x.error-handling-demo exchange using infinite-loop routing key
2. Exchange routes the message to queue q.error-handling-demo.infinite-loop based on messages’ key
3. Consumer consumes the message and throws an exception
4. Message will be requeued back to the queue
5. Back to point 3 — here we have infinite loop problem

• exchange x.error-handling-demo is being declared
• queue q.error-handling-demo.infinite-loop is being declared
• queue q.error-handling-demo.infinite-loop is bound to an exchange x.error-handling-demo using infinite-loop routing key

• producer sends a message to x.error-handling-demo exchange using infinite-loop routing key
• consumer receives a message from q.error-handling-demo.infinite-loop and throws an exception — message won’t be acknowledged, parameter requeue hasn’t been modified, so it’s value is going to be true by default

As a result, You are going to see a lot of similar logs:

1. consumer received the message — message id is b308a8d8–4eb8–4d65-a627–91c39474d50b and message time is 2021–10–31T14:06:47.351079
2. message was rejected — stack trace will be printed
3. the same message was consumed again — message id and time matches values from the bullet point above

After running this demo, remember to clear messages from queue q.error-handling-demo.infinite-loop. You can do it in management interface, by going to Queue tab, selecting queue q.error-handling-demo.infinite-loop, and clicking Purge Messages button

TIP: If You want to reject a message but You don’t want to requeue it, then You can use a built-in AmqpRejectAndDontRequeueException. It’s going to be used later

DLX (Dead Letter Exchange) and DLQ (Dead Letter Queue) concepts

Messages from a queue can be dead-lettered , which means, that they will be republished to a different exchange in case of:

• message negative acknowledgment or rejection, but with requeue parameter set to false
• message expiration due to configured TTL parameter
• message drop due to queue length limit

Dead letter exchanges are normal exchanges — if You want to learn more about exchanges in general, You can read my story 🙂

How to configure DLX for queue? You can do it either by defining server’s policy or by passing x-dead-letter-exchange argument during queue declaration.

There is also an option to override message’s original routing key by using x-dead-letter-routing-key argument. If not set, then original routing key will be used.

Messages send to DLX will contain a set of x-death headers, which You can access to find an information about:

• the reason why message was send to DLX
• original queue
• how many times this message was dead-lettered
• … and others — here You can read more about these headers

Dead letter queue is the regular queue, bound to dead letter exchange. If You would like to read more about queues in RabbitMq, I wrote a story about it.

When to use it

DLX and DLQ are being used to collect dropped messages. What to do later is up to You. You can:

• monitor DLQs size and present it in Grafana dashboard or other tool
• send an email to the support team
• …or do something different 🙂 In general it is important to have some kind of monitoring on these queues, so that You are going to know if there is something wrong happening to Your system

1. exchange x.error-handling-demo.dlx-dlq-example is being declared
2. queue q.error-handling-demo.dlx-dlq-example is being declared with x-dead-letter-exchange and x-dead-letter-routing-key arguments
3. queue q.error-handling-demo.dlx-dlq-example is bound to an exchange x.error-handling-demo.dlx-dlq-example using dlx-before routing key
4. exchange x.error-handling-demo.dlx-dlq-example.dlx is being declared
5. queue q.error-handling-demo.dlx-dlq-example.dlq is being declared
6. queue q.error-handling-demo.dlx-dlq-example.dlq is bound to an exchange x.error-handling-demo.dlx-dlq-example.dlx using dlx-after routing key

1. Producer sends a message to x.error-handling-demo.dlx-dlq-example , using dlx-before routing key
2. Consumer listens on q.error-handling-demo.dlx-dlq-example queue, and throws AmqpRejectAndDontRequeueException. This means, that message won’t be acknowledged, but it won’t be rejected back to the queue

• consumer received the message — message id is 2e05f7a6-be9d-40ba-8a35–89d497f57316 and message time is 2021–10–31T15:47:17.364577
• message was rejected — stack trace will be printed
• the same message was consumed again , but this time from dead letter queue — message id and time matches values from the bullet point above
• and… that’s over! There was no loop, and message was consumed twice — firstly from an original queue, and secondly from dead-lettered queue

DLX and DLQ with retry

Sometimes it’s worth to retry the operation. Why ?

• consumer application might had a temporal problem — e.g. it run out of memory and had to be restarted
• maybe consumer application uses other external service which is currently unavailable, but it will work soon — let’s consider e-mail sending service, which uses external e-mail server provided. Producer sends a command to send an e-mai. Company’s e-mail server is currently restarting — it will take about 2 minutes. Without the retry mechanism, e-mails won’t be send, messages will end up in DLQ, and they might require manual actions

When to retry

In general, it makes sense to retry a message if:

• an issue on consumer side is temporary — like in above example with restarting e-mail server. Message itself is correct, but some required external services are currently unavailable. Who knows, maybe it will be available in a second?
• it still makes sense to retry the message — e.g. it rather doesn’t makes sense to book a tickets for a show, which has already finished

For the simplest scenario, above properties are enough to turn on retry mechanism:

• spring.rabbitmq.listener.simple.retry.enabled — enables/disabled properties
• spring.rabbitmq.listener.simple.retry.initial-interval — duration between the first and second attempt to deliver a message
• spring.rabbitmq.listener.simple.retry.max-attempts — maximum number of attempts to deliver a message
• spring.rabbitmq.listener.simple.retry.max-interval — maximum duration between attempts
• spring.rabbitmq.listener.simple.retry.multiplier — multiplier to apply to the previous retry interval

Based on above configuration, RabbitMQ listener will try to retry the nack-ed message.

• there will be 3 attempts to deliver the message
• initial interval is going to be 5 seconds
• every other retry attempt will wait for previous interval time multiplied by multiplier, e.g. for 5,10, 20, 30,60 seconds etc.

It might be a good idea to use Exponential backoff algorithm for retry properties configuration.

Set up is the same as in the previous demo — the only difference is the configuration of above properties

1. consumer received the message — message id is c086a5a1–251b-490f-931a-0e122ca33d35 and message time is 2021–10–31T19:27:20.371739
2. message was rejected — first attempt
3. after 5 seconds (configured in spring.rabbitmq.listener.simple.retry.initial-interval property) consumer receives the same message again
4. message was rejected — second attempt
5. after 10 seconds (previous interval multiplied by spring.rabbitmq.listener.simple.retry.multiplier property)
6. message rejected — third attempt, spring.rabbitmq.listener.simple.retry.max-attempts threshold reached
7. the same message was consumed again , but this time from dead letter queue
8. and… that’s over!

Parking Lot Queue

Sometimes we might want to retry the messages from…DLQ ! In such case we are talking about another retry mechanism:

• firstly we may retry an attempt to consume a message from the queue — like in a previous example, we try to consume message 3 times
• after that, message will be send to DLX and DLQ
• and here is where the second retry mechanism might take place — after some time You might want to send messages from the DLQ back to the original queue

Источник

Обработка ошибок с помощью Spring AMQP

Узнайте о различных способах обработки ошибок с помощью Spring AMQP с помощью RabbitMQ.

Автор: baeldung
Дата записи

1. введение

Асинхронный обмен сообщениями-это тип слабо связанной распределенной коммуникации, который становится все более популярным для реализации архитектуры, управляемой событиями . К счастью, Spring Framework предоставляет проект Spring AMQP, позволяющий нам создавать решения для обмена сообщениями на основе AMQP.

С другой стороны, работа с ошибками в таких средах может быть нетривиальной задачей . Поэтому в этом уроке мы рассмотрим различные стратегии обработки ошибок.

2. Настройка среды

Для этого урока мы будем использовать RabbitMQ , который реализует стандарт AMQP. Кроме того, Spring AMQP предоставляет модуль spring-rabbit , который делает интеграцию очень простой.

Давайте запустим RabbitMQ как автономный сервер. Мы запустим его в контейнере Docker, выполнив следующую команду:

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

3. Сценарий Отказа

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

Мы можем указать на некоторые типы исключений:

• Сеть- или Связанные с вводом-выводом/| – общие сбои сетевых соединений и операций ввода-вывода Протокол-
• или связанные с инфраструктурой -ошибки, которые обычно представляют собой неправильную конфигурацию инфраструктуры обмена сообщениями Связанные с брокером
• -сбои, предупреждающие о неправильной конфигурации между клиентами и брокером AMQP. Например, достижение определенных пределов или порога, аутентификация или недопустимая конфигурация политик Application-
• и message-related – исключения, которые обычно указывают на нарушение некоторых бизнес-или прикладных правил

Конечно, этот список отказов не является исчерпывающим, но содержит наиболее распространенные типы ошибок.

Следует отметить, что Spring AMQP обрабатывает связанные с подключением и низкоуровневые проблемы из коробки, например, применяя политики retry или requeue . Кроме того, большинство отказов и неисправностей преобразуются в AmqpException или один из его подклассов.

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

4. Настройка проекта

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

Далее, давайте создадим простого производителя:

И, наконец, потребитель, который бросает исключение:

По умолчанию все неудачные сообщения будут немедленно запрашиваться в начале целевой очереди снова и снова.

Давайте запустим наш пример приложения, выполнив следующую команду Maven:

Теперь мы должны увидеть аналогичный результирующий результат:

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

Чтобы изменить это поведение у нас есть два варианта:

• Установите параметр defaultrequeuerejected в значение false на стороне слушателя – spring.rabbitmq.listener.simple.default-requeue-rejected=false
• Throw an AmqpRejectAndDontRequeueException – t his может быть полезен для сообщений, которые не будут иметь смысла в будущем, поэтому их можно отбросить.

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

5. Очередь Мертвых писем

Очередь мертвых писем (DLQ) – это очередь, содержащая недоставленные или неудачные сообщения . DLQ позволяет нам обрабатывать неисправные или плохие сообщения, отслеживать шаблоны сбоев и восстанавливаться после исключений в системе.

Что еще более важно, это помогает предотвратить бесконечные циклы в очередях, которые постоянно обрабатывают плохие сообщения и снижают производительность системы.

Всего существует два основных понятия: Обмен мертвыми письмами (DLX) и сама очередь мертвых писем (DLQ). На самом деле, DLX-это обычный обмен, который мы можем определить как один из распространенных типов : direct , topic или fanout .

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

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

5.1. Базовая конфигурация

Чтобы настроить DLQ, нам нужно указать дополнительные аргументы при определении нашей очереди:

В приведенном выше примере мы использовали два дополнительных аргумента: x-dead-letter-exchange и x-dead-letter-routing-key . Пустое строковое значение для параметра x-dead-letter-exchange указывает брокеру использовать обмен по умолчанию .

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

5.2. Неудачная Маршрутизация Сообщений

Поэтому, когда сообщение не доставляется, оно направляется на обмен мертвыми письмами. Но, как мы уже отмечали, DLX-это нормальный обмен. Поэтому, если ключ маршрутизации неудачного сообщения не совпадает с обменом, он не будет доставлен в DLQ.

Таким образом, если мы опустим аргумент x-dead-letter-routing-key в нашем примере, неудачное сообщение застрянет в бесконечном цикле повторных попыток.

Кроме того, исходная метаинформация сообщения доступна в x-death heaven:

То приведенная выше информация доступна в консоли управления RabbitMQ обычно работает локально на порту 15672.

Помимо этой конфигурации, если мы используем Spring Cloud Stream , мы можем даже упростить процесс настройки, используя свойства конфигурации republishToDlq и autoBindDlq .

5.3. Обмен Мертвыми письмами

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

На этот раз мы определили пользовательский обмен типа fan out , поэтому сообщения будут отправляться во все ограниченные очереди . Кроме того, мы установили значение аргумента x-dead-letter-exchange для имени нашего DLX. В то же время мы удалили аргумент x-dead-letter-routing-key .

Теперь, если мы запустим наш пример, неудачное сообщение должно быть доставлено в DLQ, но без изменения начального ключа маршрутизации:

5.4. Обработка Сообщений Очереди Мертвых Писем

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

Давайте определим слушателя для очереди мертвых писем:

Если мы запустим наш пример кода сейчас, то увидим вывод журнала:

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

Например, мы можем просто запросить сообщение в исходное место назначения:

Но такая логика исключений не отличается от политики повторных попыток по умолчанию:

Общая стратегия может потребовать повторной обработки сообщения в течение n раз, а затем отклонить его. Давайте реализуем эту стратегию, используя заголовки сообщений:

Сначала мы получаем значение заголовка x-retries-count , затем сравниваем это значение с максимально допустимым значением. Впоследствии, если счетчик достигнет предельного числа попыток, сообщение будет отброшено:

Мы должны добавить, что мы также можем использовать заголовок x-message-ttl , чтобы установить время, после которого сообщение должно быть отброшено. Это может быть полезно для предотвращения бесконечного роста очередей.

5.5. Очередь на Парковку

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

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

Давайте теперь воплотим эту идею:

Во – вторых, давайте реорганизуем логику прослушивателя, чтобы отправить сообщение в очередь парковки:

В конце концов, нам также нужно обрабатывать сообщения, поступающие в очередь парковки:

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

Давайте проверим эту логику, запустив наше приложение:

Как видно из выходных данных, после нескольких неудачных попыток сообщение было отправлено в очередь парковки.

6. Пользовательская Обработка Ошибок

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

6.1. Глобальный обработчик ошибок

До сих пор мы использовали по умолчанию SimpleRabbitListenerContainerFactory и эта фабрика по умолчанию использует ConditionalRejectingErrorHandler . Этот обработчик ловит различные исключения и преобразует их в одно из исключений в иерархии AmqpException .

Важно отметить, что если нам нужно обрабатывать ошибки подключения, то нам нужно реализовать интерфейс ApplicationListener .

Проще говоря, Условное отклонение ErrorHandler решает, отклонять ли конкретное сообщение или нет. Когда сообщение, вызвавшее исключение, отклоняется, оно не запрашивается.

Давайте определим пользовательский Обработчик ошибок , который будет просто запрашивать только Бизнес-исключение s:

Кроме того, поскольку мы выбрасываем исключение внутри нашего метода listener, оно оборачивается в ListenerExecutionFailedException . Итак, нам нужно вызвать метод getCause , чтобы получить исходное исключение.

6.2. Стратегия Фатальных Исключений

Под капотом этот обработчик использует стратегию Fatal Exception , чтобы проверить, следует ли считать исключение фатальным. Если это так, то неудачное сообщение будет отклонено.

По умолчанию эти исключения фатальны:

• MessageConversionException
• MessageConversionException
• MethodArgumentNotValidException
• Метод Аргумент TypeMismatchException
• NoSuchMethodException
• ClassCastException

Вместо реализации интерфейса Ierrorhandler мы можем просто предоставить нашу стратегию Фатальных исключений :

Наконец, нам нужно передать нашу пользовательскую стратегию конструктору Conditional Rejecting ErrorHandler :

7. Заключение

В этом уроке мы обсудили различные способы обработки ошибок при использовании Spring AMQP и, в частности, RabbitMQ.

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

Как всегда, полный исходный код статьи доступен на GitHub .

Источник

In this article we will implement a RabbitMQ Error Handling.
Whenever any data in the message is
transmitted that the receiver does not accept, or when a message is sent to a queue that does not
exist. The message is retried and sent up to a set number of times. Even if the communication is not
received by the recipient but is sent from the sender’s end.
Now In such instances, the message queue is marked as undeliverable or deadLetter queue.

RabbitMQ provides a method for handling message failures in a really efficient way known as the Retry
and Error Handling feature.

What is Dead Message in RabbitMQ.

If certain messages become undeliverable or unhandled even though when message received by the
broker. This can
occur whenever the amount of time that the message spent in a queue is longer than the time it takes
to live TTL, when the queue hits its capacity or the consumer admits a message negatively. Such a
message is known as dead
message.

However there is a better way to handle this situation. Setting up a RabbitMQ dead letter
exchange and a dead letter queue enables for the storage and processing of orphaned messages.
Now there is no need to lose messages completely.

In this case, if the employee experience provided in the RequestParam does not meet the criteria,
then retry and error handling feature is invoked, and the message queue is declared as deadLetter
queue.

Let’s start developing Spring Boot for Producer and Consumer applications.

Producer

Project Structure

Maven Dependencies

We will use the Spring AMQP dependency to develop AMQP-based messaging solutions.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.5.1</version>
		<relativePath /> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.techgeeknext</groupId>
	<artifactId>SpringBootRabbitMQProducer</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>SpringBootRabbitMQProducer</name>
	<description>Spring Boot RabbitMQ + Error Handling</description>
	<properties>
		<java.version>1.8</java.version>
	</properties>
	<dependencies>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-logging</artifactId>
		</dependency>

		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<version>1.18.20</version>
			<scope>provided</scope>
		</dependency>

	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

Employee Model Class

Using lombok, which will generate getters and setters automatically.

package com.techgeeknext.model;


import com.fasterxml.jackson.annotation.JsonIdentityInfo;
import com.fasterxml.jackson.annotation.ObjectIdGenerators;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

@Setter
@ToString
@Getter
@JsonIdentityInfo(generator = ObjectIdGenerators.IntSequenceGenerator.class, property = "@id"
		, scope = Employee.class)
public class Employee {

    private String name;
    private String domain;
    private int experience;

}

RabbitMQ Configuration Class

We require x-dead-letter-exchange and x-dead-letter-routing-key in order
to configure a dead letter.
It instructs the broker to use the standard exchange.
The deadLetterExchange is used to bind dlq, and the techgeeknextExchange
is used to bind
techgeeknext.

The Jackson2JsonMessageConverter translates an object to JSON and then from JSON to a Java object.

package com.techgeeknext.config;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQProducerConfig {

    @Bean
    DirectExchange deadLetterExchange() {
        return new DirectExchange("deadLetterExchange");
    }

    @Bean
    DirectExchange exchange() {
        return new DirectExchange("techgeeknextExchange");
    }

    @Bean
    Queue dlq() {
        return QueueBuilder.durable("deadLetter.queue").build();
    }

    @Bean
    Queue queue() {
        return QueueBuilder.durable("techgeeknext.queue")
                .withArgument("x-dead-letter-exchange", "deadLetterExchange")
                .withArgument("x-dead-letter-routing-key", "deadLetter").build();
    }

    @Bean
    Binding DLQbinding() {
        return BindingBuilder.bind(dlq())
                .to(deadLetterExchange()).with("deadLetter");
    }

    @Bean
    Binding binding() {
        return BindingBuilder.bind(queue())
                .to(exchange()).with("techgeeknext");
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    public AmqpTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }
}

Controller

Produceer method is used to send the message to the consumer through AMQP message.

package com.techgeeknext.controller;

import com.techgeeknext.model.Employee;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController

public class RabbitMQProducerController {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @GetMapping(value = "/sendMessage")
    public String producer(@RequestParam("name") String name, @RequestParam("domain") String domain,
						   @RequestParam("exp") int exp) {
        Employee emp = new Employee();
        emp.setName(name);
        emp.setDomain(domain);
        emp.setExperience(exp);

        //The convertAndSend method converts the java object to an amqp message,
        // and then sends this message via the routing key to the exchange.
        amqpTemplate.convertAndSend("techgeeknextExchange", "techgeeknext", emp);
        return "Message sent to RabbitMQ server successfully!!";
    }
}

Consumer

Project Structure

Maven Dependencies

We will use the Spring AMQP dependency to develop AMQP-based messaging solutions.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.5.1</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.techgeeknext</groupId>
	<artifactId>SpringBootRabbitMQConsumer</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>SpringBootRabbitMQConsumer</name>
	<description>Spring Boot RabbitMQ + Error Handling Example</description>
	<properties>
		<java.version>1.8</java.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-logging</artifactId>
		</dependency>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<scope>provided</scope>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

Properties

Here in application.yml file, we have activated the RabbitMQ retry mechanism for Spring Boot.

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true

##After an interval of 3s the message should be retried.
          initial-interval: 3s

##Maximum 6 times the message should be retried.
##It is then sent to the dead letter Queue.
          max-attempts: 6

##The maximum interval should not be more than 10s between two retries.
          max-interval: 10s

##The gap is multiplied by 2 between second repetition.
## But the maximum interval can never be exceeded
          multiplier: 2
##So the retry interval values will be 3s, 6s, 10s, 10s, 10s. As 10 sec is the max interval specified.

server:
  port: 8081

Custom Checked Exception

Create custom checked exception called CustomInvalidException.

package com.techgeeknext.exception;

public class CustomInvalidException extends Exception {

    private static final long serialVersionUID = -3154618962130084535L;

}

RabbitMQ Consumer

Create the class EmployeeConsumerService that uses RabbitListener to consume message from RabbitMQ.
For
incoming messages the RabbitMQ Listener listens for RabbitMQ Queueue. The Queue/Topic Name (name of
the queue/topic where the message should be consumed) is used for the basic configuration.

We’ll also check the employee experience field here and throw CustomInvalidException in the
case of
a negative or invalid experience.

package com.techgeeknext.service;

import com.techgeeknext.exception.CustomInvalidException;
import com.techgeeknext.model.Employee;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


@Component
public class EmployeeConsumerService {

	private static final Logger logger = LoggerFactory.getLogger(EmployeeConsumerService.class);

	@RabbitListener(queues = "techgeeknext.queue")
	public void consumeMessage(Employee employee) throws CustomInvalidException {
		logger.info("Recieved Message From RabbitMQ techgeeknextExchange: " + employee);
		if (employee.getExperience() < 0 || employee.getExperience() > 30 ) {
			throw new CustomInvalidException();
		}
	}
}

Test

1. Now, test this application at local. Start local RabbitMQ.





2. Start the Producer Spring Boot Application.

  .   _____  __ _ _
 /\ / ___'_ __ _ _(_)_ __  __ _    
( ( )___ | '_ | '_| | '_ / _` |    
 \/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |___, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::(v2.5.1)

16:00:18.798  INFO 17000 --- [ main] c.t.SpringBootRabbitMQProducer : Starting SpringBootRabbitMQProducer using Java 12.0.2 on LAPTOP-S6R44CQL with PID 17000 (D:SpringBootRabbitMQProducertargetclasses started in D:SpringBootRabbitMQProducer)
16:00:18.801  INFO 17000 --- [ main] c.t.SpringBootRabbitMQProducer : No active profile set, falling back to default profiles: default
16:00:19.922  INFO 17000 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8080 (http)
16:00:19.932  INFO 17000 --- [ main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
16:00:19.933  INFO 17000 --- [ main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.46]
16:00:20.008  INFO 17000 --- [ main] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring embedded WebApplicationContext
16:00:20.008  INFO 17000 --- [ main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 1174 ms
16:00:20.525  INFO 17000 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer  :
Tomcat started on port(s): 8080 (http) with context path ''
16:00:20.534  INFO 17000 --- [ main] c.t.SpringBootRabbitMQProducer :
Started SpringBootRabbitMQProducer in 2.073 seconds (JVM running for 2.362)
16:00:44.371  INFO 17000 --- [nio-8080-exec-2] o.a.c.c.C.[Tomcat].[localhost].[/] :
Initializing Spring DispatcherServlet 'dispatcherServlet'
16:00:44.372  INFO 17000 --- [nio-8080-exec-2] o.s.web.servlet.DispatcherServlet  :
Initializing Servlet 'dispatcherServlet'
16:00:44.374  INFO 17000 --- [nio-8080-exec-2] o.s.web.servlet.DispatcherServlet  :
Completed initialization in 2 ms
16:01:04.917  INFO 17000 --- [nio-8080-exec-3] o.s.a.r.c.CachingConnectionFactory :
Attempting to connect to: [localhost:5672]
16:01:05.029  INFO 17000 --- [nio-8080-exec-3] o.s.a.r.c.CachingConnectionFactory :
Created new connection: rabbitConnectionFactory#14229fa7:0/SimpleConnection@58b43bdf
[delegate=amqp://guest@127.0.0.1:5672/, localPort= 61153]

3. Produce/send the message with invalid Employee experience as 35 to consumer by using endpoint as
   http://localhost:8080/sendMessage?name=techgeeknext&domain=XYZ&exp=35



4. Now start the Consumer Spring Boot Application.
5. Consumer : The message is sent by rabbitMQ queue to the consumer’s application, which is designated
   techgeeknext.queue. Since the employee experience is invalid,
   it’ll throw CustomInvalidException. This message is tried again 6 times and then placed in the queue of the dead letter.

    .   ____    _ __ _ _
    /\ / ___'_ __ _ _(_)_ __  __ _    
   ( ( )___ | '_ | '_| | '_ / _` |    
    \/  ___)| |_)| | | | | || (_| |  ) ) ) )
     '  |____| .__|_| |_|_| |___, | / / / /
    =========|_|==============|___/=/_/_/_/
    :: Spring Boot ::     (v2.5.1)
   
   16:00:31.805  INFO 8532 --- [main] c.t.SpringBootRabbitMQConsumer: Starting SpringBootRabbitMQConsumer using Java 12.0.2 on LAPTOP-S6R44CQL with PID 8532 (D:SpringBootRabbitMQConsumertargetclasses started in D:SpringBootRabbitMQConsumer)
   16:00:31.808  INFO 8532 --- [main] c.t.SpringBootRabbitMQConsumer: No active profile set, falling back to default profiles: default
   16:00:32.848  INFO 8532 --- [main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8081 (http)
   16:00:32.856  INFO 8532 --- [main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
   16:00:32.856  INFO 8532 --- [main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.46]
   16:00:32.939  INFO 8532 --- [main] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring embedded WebApplicationContext
   16:00:32.939  INFO 8532 --- [main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 1088 ms
   16:00:33.417  INFO 8532 --- [main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8081 (http) with context path ''
   16:00:33.418  INFO 8532 --- [main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [localhost:5672]
   16:00:33.458  INFO 8532 --- [main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#37b56ac7:0/SimpleConnection@6ebc9573 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 64346]
   16:00:33.510  INFO 8532 --- [main] c.t.SpringBootRabbitMQConsumer: Started SpringBootRabbitMQConsumer in 2.069 seconds (JVM running for 2.353)
   16:03:10.713  INFO 8532 --- [ntContainer#0-1] c.t.service.EmployeeConsumerService:
   Recieved Message From RabbitMQ techgeeknextExchange:
   Employee(name=techgeeknext, domain=XYZ, experience=35)
   16:03:13.719  INFO 8532 --- [ntContainer#0-1] c.t.service.EmployeeConsumerService:
   Recieved Message From RabbitMQ techgeeknextExchange:
   Employee(name=techgeeknext, domain=XYZ, experience=35)
   16:03:19.735  INFO 8532 --- [ntContainer#0-1] c.t.service.EmployeeConsumerService:
   Recieved Message From RabbitMQ techgeeknextExchange:
   Employee(name=techgeeknext, domain=XYZ, experience=35)
   16:03:29.746  INFO 8532 --- [ntContainer#0-1] c.t.service.EmployeeConsumerService:
   Recieved Message From RabbitMQ techgeeknextExchange:
   Employee(name=techgeeknext, domain=XYZ, experience=35)
   16:03:39.758  INFO 8532 --- [ntContainer#0-1] c.t.service.EmployeeConsumerService:
   Recieved Message From RabbitMQ techgeeknextExchange:
   Employee(name=techgeeknext, domain=XYZ, experience=35)
   16:03:49.760  INFO 8532 --- [ntContainer#0-1] c.t.service.EmployeeConsumerService:
   Recieved Message From RabbitMQ techgeeknextExchange:
   Employee(name=techgeeknext, domain=XYZ, experience=35)
   16:03:49.784  WARN 8532 --- [ntContainer#0-1] o.s.a.r.r.RejectAndDontRequeueRecoverer  :
   Retries exhausted for message (Body:'{"@id":1,"name":"techgeeknext","domain":"XYZ","experience":35}' MessageProperties [headers={__TypeId__=com.techgeeknext.model.Employee}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=techgeeknextExchange, receivedRoutingKey=techgeeknext, deliveryTag=2, consumerTag=amq.ctag-u4Xd3zuiVqOKGf8FzwA7qg, consumerQueue=techgeeknext.queue])
   
   org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void
   com.techgeeknext.service.EmployeeConsumerService.recievedMessage(com.techgeeknext.model.Employee)
   throws com.techgeeknext.exception.CustomInvalidException' threw exception
   	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:247) ~[spring-rabbit-2.3.8.jar:2.3.8]
   	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:191) ~[spring-rabbit-2.3.8.jar:2.3.8]
   	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:137) ~[spring-rabbit-2.3.8.jar:2.3.8]
   	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1654) ~[spring-rabbit-2.3.8.jar:2.3.8]
   	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1573) ~[spring-rabbit-2.3.8.jar:2.3.8]
   	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
   	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
   	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
   	at java.base/java.lang.reflect.Method.invoke(Method.java:567) ~[na:na]
   	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) ~[spring-aop-5.3.8.jar:5.3.8]
   	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198) ~[spring-aop-5.3.8.jar:5.3.8]
   	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.8.jar:5.3.8]
   	at org.springframework.retry.interceptor.RetryOperationsInterceptor$1.doWithRetry(RetryOperationsInterceptor.java:93) ~[spring-retry-1.3.1.jar:na]
   	at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329) ~[spring-retry-1.3.1.jar:na]
   	at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225) ~[spring-retry-1.3.1.jar:na]
   	at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:116) ~[spring-retry-1.3.1.jar:na]
   	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.8.jar:5.3.8]
   	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:215) ~[spring-aop-5.3.8.jar:5.3.8]
   	at org.springframework.amqp.rabbit.listener.$Proxy65.invokeListener(Unknown Source) ~[na:2.3.8]
   	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1561) ~[spring-rabbit-2.3.8.jar:2.3.8]
   	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1552) 

6. Rabbitmq Exchange
   

   

7. Rabbitmq Queues
   
   

Download Source Code

The full source code for this article can be found on below.
Download it here —

1. Producer Spring Boot + RabbitMQ + Error Handling Example
2. Consumer Spring Boot + RabbitMQ + Error Handling Example

Introduction

In this example I am going to show you how to retry the failed messages in RabbitMQ message broker. So, here I am going to retry messages for few times and if error still exists after retrying the maximum attempts then the message will be put into the dead letter queue.

Related Posts:

• RabbitMQ Spring Boot Publish Subscribe Example
• RabbitMQ Spring Boot Work Queues Example
• RabbitMQ Spring Boot Producer Consumer Example
• RabbitMQ Spring Boot Routing Example
• RabbitMQ Spring Boot Message Pattern Example
• RabbitMQ Spring Boot RPC Example

What is Dead Letter Queue?

A dead letter queue is an undelivered message queue, so a dead letter queue is a service implementation used to store messages that meet one or more of the following criteria:

1. Message that is sent to a queue that does not exist.
2. Queue length limit exceeded.
3. Message length limit exceeded.
4. Message is rejected by another queue exchange.
5. Message reaches a threshold read counter number, because it is not consumed. Sometimes this is called a “back out queue”.
6. The message expires due to per-message TTL (time to live).
7. Message is not processed successfully.

The dead letter queue storing these messages allows developers to analyze messages to look for the problems why the messages are failing to be delivered.

Prerequisites

Java 1.8+, Maven 3.8.2, Spring Boot 2.6.2, Spring AMQP 2.6.2, RabbitMQ Server 3.9.4 – 3.9.13

Project Setup

The following pom.xml file can be used for the maven based project creation in your favorite IDE or tool.

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.roytuts</groupId>
	<artifactId>spring-rabbitmq-error-retry-dlq</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<maven.compiler.source>11</maven.compiler.source>
		<maven.compiler.target>11</maven.compiler.target>
	</properties>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.6.2</version>
	</parent>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-devtools</artifactId>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
</project>

Application Properties

The application.properties file is kept with the following content under classpath folder src/main/resources.

exchange.dl=dlExchange
exchange.roytuts=roytutsExchange
queue.dl=dl.queue
queue.roytuts=roytuts.queue
routing.key.dl=dl
routing.key.roytuts=roytuts

spring.rabbitmq.listener.simple.retry.enabled=true
spring.rabbitmq.listener.simple.retry.initial-interval=2s
spring.rabbitmq.listener.simple.retry.max-attempts=3
spring.rabbitmq.listener.simple.retry.max-interval=6s
spring.rabbitmq.listener.simple.retry.multiplier=2

I have enabled retry mechanism using the property key spring.rabbitmq.listener.simple.retry.enabled=true. The spring.rabbitmq.listener.simple.retry.initial-interval=2s tells to retry the message after initial interval of 2 seconds.

spring.rabbitmq.listener.simple.retry.max-attempts=3 tells that the maximum of 3 retries will be happened and after that the message will be put into dead letter queue.

spring.rabbitmq.listener.simple.retry.max-interval=6s says that the maximum interval between two retries is 6 seconds.

The interval in subsequent retry gets multiplied by 2 using the key/value pair spring.rabbitmq.listener.simple.retry.multiplier=2. Therefore, the retry interval will be 2s, 4s, 6s, etc.

Custom Exception

I am creating a custom exception class to throw an exception for invalid string.

package com.roytuts.spring.rabbitmq.error.retry.dlq.exception;

public class InvalidNameException extends Exception {

	private static final long serialVersionUID = 1L;

}

Producer or Sender

The producer or sender class produces or sends a message.

package com.roytuts.spring.rabbitmq.error.retry.dlq.producer;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class Producer {

	@Value("${exchange.roytuts}")
	private String roytutsExchange;

	@Value("${routing.key.roytuts}")
	private String routingKeyRoytuts;

	@Autowired
	private RabbitTemplate rabbitTemplate;

	public void sendName(final String name) {
		rabbitTemplate.convertAndSend(roytutsExchange, routingKeyRoytuts, name);
		System.out.println("Sent: " + name);
	}

}

Consumer or Receiver

The consumer or receiver consumes or receives a message.

package com.roytuts.spring.rabbitmq.error.retry.dlq.consumer;

import java.util.regex.Pattern;

import javax.naming.InvalidNameException;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "${queue.roytuts}")
public class Consumer {

	@RabbitHandler
	public void receiveMsg(final String name) throws InvalidNameException {
		if (!Pattern.matches("[a-zA-Z]+", name)) {
			throw new InvalidNameException("Name should contain only alphabets");
		}

		System.out.println("Received: " + name);
	}

}

Config Class

The configuration class will create several beans required for the applications.

package com.roytuts.spring.rabbitmq.error.retry.dlq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class Config {

	@Value("${exchange.dl}")
	private String dlExchange;

	@Value("${exchange.roytuts}")
	private String roytutsExchange;

	@Value("${queue.dl}")
	private String dlQueue;

	@Value("${queue.roytuts}")
	private String roytutsQueue;

	@Value("${routing.key.dl}")
	private String routingKeyDl;

	@Value("${routing.key.roytuts}")
	private String routingKeyRoytuts;

	@Bean
	public DirectExchange dlExchange() {
		return new DirectExchange(dlExchange);
	}

	@Bean
	public DirectExchange roytutsExchange() {
		return new DirectExchange(roytutsExchange);
	}

	@Bean
	public Queue dlq() {
		return QueueBuilder.durable(dlQueue).build();
	}

	@Bean
	public Queue queue() {
		return QueueBuilder.durable(roytutsQueue).withArgument("x-dead-letter-exchange", dlExchange)
				.withArgument("x-dead-letter-routing-key", routingKeyDl).build();
	}

	@Bean
	public Binding binding() {
		return BindingBuilder.bind(queue()).to(roytutsExchange()).with(routingKeyRoytuts);
	}

	@Bean
	public Binding dlqBinding() {
		return BindingBuilder.bind(dlq()).to(dlExchange()).with(routingKeyDl);
	}

}

Spring Boot Main Class

A class with a main method and @SpringBootApplication annotation is enough to deploy the application into embedded Tomcat server.

package com.roytuts.spring.rabbitmq.error.retry.dlq;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import com.roytuts.spring.rabbitmq.error.retry.dlq.producer.Producer;

@SpringBootApplication
public class SpringRabbitMqDlqApp implements CommandLineRunner {

	@Autowired
	private Producer producer;

	public static void main(String[] args) {
		SpringApplication.run(SpringRabbitMqDlqApp.class, args);
	}

	@Override
	public void run(String... args) throws Exception {
		producer.sendName("Soumitra");
		producer.sendName("Roy");
		producer.sendName("Roy Tutorials");
		producer.sendName("Soumitra 2");
		producer.sendName("Roy Tutorials2");
	}

}

Testing RabbitMQ Dead Letter Queue

Once you run the above main class you will see the following output:

20.378  INFO 22836 --- [  restartedMain] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
20.430  INFO 22836 --- [  restartedMain] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#5fe18d6d:0/SimpleConnection@18cce147 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56240]
21.131  INFO 22836 --- [  restartedMain] c.r.s.r.e.r.dlq.SpringRabbitMqDlqApp     : Started SpringRabbitMqDlqApp in 2.001 seconds (JVM running for 2.433)
Sent: Soumitra
Sent: Roy
Sent: Roy Tutorials
Sent: Soumitra 2
Sent: Roy Tutorials2
Received: Soumitra
Received: Roy
27.167  WARN 22836 --- [ntContainer#0-1] o.s.a.r.r.RejectAndDontRequeueRecoverer  : Retries exhausted for message (Body:'Roy Tutorials' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=roytutsExchange, receivedRoutingKey=roytuts, deliveryTag=3, consumerTag=amq.ctag-d6XWSD3q37Fmbo6o9tqthw, consumerQueue=roytuts.queue])

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(java.lang.String) throws javax.naming.InvalidNameException' threw exception
	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:270) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: javax.naming.InvalidNameException: Name should contain only alphabets
	at com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(Consumer.java:18) ~[classes/:na]
	...

27.168  WARN 22836 --- [ntContainer#0-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
	at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(java.lang.String) throws javax.naming.InvalidNameException' threw exception
	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:270) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: javax.naming.InvalidNameException: Name should contain only alphabets
	at com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(Consumer.java:18) ~[classes/:na]
	...

33.194  WARN 22836 --- [ntContainer#0-1] o.s.a.r.r.RejectAndDontRequeueRecoverer  : Retries exhausted for message (Body:'Soumitra 2' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=roytutsExchange, receivedRoutingKey=roytuts, deliveryTag=4, consumerTag=amq.ctag-d6XWSD3q37Fmbo6o9tqthw, consumerQueue=roytuts.queue])

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(java.lang.String) throws javax.naming.InvalidNameException' threw exception
	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:270) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: javax.naming.InvalidNameException: Name should contain only alphabets
	at com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(Consumer.java:18) ~[classes/:na]
	...

33.195  WARN 22836 --- [ntContainer#0-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
	at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(java.lang.String) throws javax.naming.InvalidNameException' threw exception
	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:270) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: javax.naming.InvalidNameException: Name should contain only alphabets
	at com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(Consumer.java:18) ~[classes/:na]
	...

39.209  WARN 22836 --- [ntContainer#0-1] o.s.a.r.r.RejectAndDontRequeueRecoverer  : Retries exhausted for message (Body:'Roy Tutorials2' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=roytutsExchange, receivedRoutingKey=roytuts, deliveryTag=5, consumerTag=amq.ctag-d6XWSD3q37Fmbo6o9tqthw, consumerQueue=roytuts.queue])

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(java.lang.String) throws javax.naming.InvalidNameException' threw exception
	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:270) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: javax.naming.InvalidNameException: Name should contain only alphabets
	at com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(Consumer.java:18) ~[classes/:na]
	...

39.213  WARN 22836 --- [ntContainer#0-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Retry Policy Exhausted
	at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:76) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method 'public void com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(java.lang.String) throws javax.naming.InvalidNameException' threw exception
	at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:270) ~[spring-rabbit-2.4.2.jar:2.4.2]
	...
Caused by: javax.naming.InvalidNameException: Name should contain only alphabets
	at com.roytuts.spring.rabbitmq.error.retry.dlq.consumer.Consumer.receiveMsg(Consumer.java:18) ~[classes/:na]
	...

In the above output you will see that only Soumitra and Roy were received by the receiver because these two strings do not contain any space or other than alphabets. The rest of the strings were discarded and hence exceptions thrown and moved to the dead letter queue as shown in the following image:

Source Code

Download