RabbitMQ with Spring AMQP

Declare exchanges and queues, publish JSON with RabbitTemplate, consume with @RabbitListener, and stop a poison message from looping forever.

Syntax@Bean Queue / TopicExchange / Binding rabbitTemplate.convertAndSend(exchange, routingKey, payload); @RabbitListener(queues = "queue.name") void handle(Payload p) { }

Add spring-boot-starter-amqp and Spring Boot creates a connection factory, a RabbitTemplate for sending and a listener container factory for @RabbitListener methods. Connection settings live under spring.rabbitmq.*; locally the defaults point at localhost:5672 with the guest user.

Exchanges, queues and bindings

Producers never write to a queue directly. They publish to an exchange with a routing key, and bindings decide which queues receive a copy. A topic exchange matches keys with wildcards: a queue bound with order.* receives order.placed and order.cancelled. Declare Queue, TopicExchange and Binding beans (from org.springframework.amqp.core) and Spring creates them on the broker when the first connection opens.

Sending JSON

The default converter only handles strings, byte arrays and Serializable objects. Declare a JacksonJsonMessageConverter bean, the Jackson 3 based converter in org.springframework.amqp.support.converter, and Spring Boot applies it to both the template and the listener containers. Then convertAndSend(exchange, routingKey, record) sends JSON and a listener method can take the record as its parameter.

rabbit.convertAndSend("orders", "order.placed", event);

@RabbitListener(queues = "billing.order-placed")
void on(OrderPlaced event) { ... }

Acknowledgements and failures

Spring acknowledges the message when the listener method returns and rejects it when it throws. By default a rejected message is requeued, so a message that always fails, a poison message, is redelivered in a tight loop that burns CPU and fills the logs. Two fixes belong in every application:

  • Give the queue a dead-letter exchange so rejected messages go to a parking queue you can inspect.
  • Enable listener retry with spring.rabbitmq.listener.simple.retry.enabled=true. After the retries are used up the message is rejected without requeueing, which sends it to the dead-letter exchange. Throwing AmqpRejectAndDontRequeueException rejects a message without requeueing. The retry policy Spring Boot builds retries every exception, this one included, so to skip retries for errors that will never succeed, such as invalid data, narrow the policy with a RabbitListenerRetrySettingsCustomizer bean.

Production notes

Use durable quorum queues for anything you cannot lose; classic non-durable queues vanish when the broker restarts. Set prefetch to limit how many unacknowledged messages one consumer holds, so a slow instance does not hoard work. Keep listener methods idempotent, because a consumer that dies before acknowledging hands the message to someone else.

Example

Example · java
import java.math.BigDecimal;
import java.util.UUID;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.JacksonJsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

record OrderPlaced(UUID eventId, long orderId, BigDecimal total) {
}

@Configuration
public class OrderMessagingConfig {

    static final String EXCHANGE = "orders";
    static final String BILLING_QUEUE = "billing.order-placed";

    @Bean
    TopicExchange ordersExchange() {
        return new TopicExchange(EXCHANGE);
    }

    @Bean
    Queue billingQueue() {
        return QueueBuilder.durable(BILLING_QUEUE)
                .quorum()
                .deadLetterExchange("orders.dlx")
                .build();
    }

    @Bean
    Binding billingBinding(Queue billingQueue, TopicExchange ordersExchange) {
        return BindingBuilder.bind(billingQueue).to(ordersExchange).with("order.placed");
    }

    @Bean
    MessageConverter jsonMessageConverter() {
        return new JacksonJsonMessageConverter();
    }
}

@Service
class OrderEventPublisher {

    private final RabbitTemplate rabbit;

    OrderEventPublisher(RabbitTemplate rabbit) {
        this.rabbit = rabbit;
    }

    void orderPlaced(long orderId, BigDecimal total) {
        rabbit.convertAndSend(OrderMessagingConfig.EXCHANGE, "order.placed",
                new OrderPlaced(UUID.randomUUID(), orderId, total));
    }
}

@Component
class BillingListener {

    @RabbitListener(queues = OrderMessagingConfig.BILLING_QUEUE)
    void onOrderPlaced(OrderPlaced event) {
        // create the invoice; throwing here triggers retry, then dead-lettering
    }
}

When to use it

  • Distribute PDF invoice generation across four worker instances that compete for messages on one queue.
  • Route order.placed to billing and shipping queues, and order.cancelled only to shipping, with one topic exchange.
  • Park malformed partner messages in a dead-letter queue and alert support instead of crashing the consumer loop.

More examples

Connection, retry and prefetch settings

Retry happens inside the listener container, on the same consumer thread, with a growing delay between attempts. In Spring Boot 4 the count is <code>max-retries</code>, the retries after the first attempt; the older <code>max-attempts</code> property was replaced in Spring Boot 4.0 and is no longer supported. When retries run out the message is rejected without requeueing, so it reaches the queue's dead-letter exchange.

Example · properties
spring.rabbitmq.host=${RABBITMQ_HOST:localhost}
spring.rabbitmq.port=5672
spring.rabbitmq.username=${RABBITMQ_USER:guest}
spring.rabbitmq.password=${RABBITMQ_PASSWORD:guest}
spring.rabbitmq.virtual-host=/

spring.rabbitmq.listener.simple.prefetch=20
spring.rabbitmq.listener.simple.concurrency=2
spring.rabbitmq.listener.simple.max-concurrency=8
spring.rabbitmq.listener.simple.retry.enabled=true
spring.rabbitmq.listener.simple.retry.max-retries=3
spring.rabbitmq.listener.simple.retry.initial-interval=1s
spring.rabbitmq.listener.simple.retry.multiplier=2

Rejecting a message that can never succeed

Retrying a message with a negative amount only delays the inevitable. The listener throws <code>AmqpRejectAndDontRequeueException</code>, and the customizer tells Spring Boot's retry policy not to retry when that exception appears anywhere in the cause chain (the container wraps listener exceptions, so checking only the top level would miss it). The message is then rejected without requeueing and moves straight to the dead-letter queue, while transient errors such as timeouts still get their retries.

Example · java
import java.math.BigDecimal;
import java.util.UUID;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.boot.amqp.autoconfigure.RabbitListenerRetrySettingsCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

record PaymentRequested(UUID eventId, long orderId, BigDecimal amount, String currency) {
}

@Configuration
class ListenerRetryConfig {

    @Bean
    RabbitListenerRetrySettingsCustomizer skipRetryForRejectedMessages() {
        return settings -> settings.setExceptionPredicate(ex -> !causedByReject(ex));
    }

    private static boolean causedByReject(Throwable ex) {
        for (Throwable t = ex; t != null; t = t.getCause()) {
            if (t instanceof AmqpRejectAndDontRequeueException) {
                return true;
            }
        }
        return false;
    }
}

@Component
public class PaymentListener {

    @RabbitListener(queues = "payments.requested")
    void handle(PaymentRequested request) {
        if (request.amount() == null || request.amount().signum() <= 0) {
            throw new AmqpRejectAndDontRequeueException("Invalid amount for order " + request.orderId());
        }
        charge(request);
    }

    private void charge(PaymentRequested request) {
        // call the payment provider; a timeout here is worth retrying
    }
}

A dead-letter exchange and a parking queue

Every queue declared with <code>deadLetterExchange("orders.dlx")</code> sends rejected messages here. A fanout exchange ignores the routing key, so one parking queue collects dead letters from every source queue. No listener consumes <code>orders.parking</code> on purpose: a consumer would acknowledge and remove the messages you need to inspect. RabbitMQ adds an <code>x-death</code> header naming the original queue and the reason, which you can read in the management UI. In production, alert on the parking queue's depth and replay messages by hand once the bug is fixed.

Example · java
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DeadLetterConfig {

    @Bean
    FanoutExchange deadLetterExchange() {
        return new FanoutExchange("orders.dlx");
    }

    @Bean
    Queue parkingQueue() {
        return QueueBuilder.durable("orders.parking").quorum().build();
    }

    @Bean
    Binding parkingBinding(Queue parkingQueue, FanoutExchange deadLetterExchange) {
        return BindingBuilder.bind(parkingQueue).to(deadLetterExchange);
    }
}

Related

sb messaging for web developerssb kafka with spring for apache kafkasb the transactional outbox patternsb websockets with stompsb docker compose in development

Discussion

  • Be the first to comment on this lesson.