Skip to content
JavaAgentic

Type at least two characters. Try “RAG”, “pgvector” or “tool calling”.

Spring AMQP & RabbitMQ Integration

Spring AMQP in production: RabbitTemplate and message converters, @RabbitListener containers, manual acknowledgement, retry with backoff, and a dead-letter topology that works.

Intermediate6 min readUpdated
On this page

Spring AMQP gives RabbitMQ the same shape as the rest of Spring: a template for sending, an annotation for receiving, and a container managing consumption. Most of the work is in the topology and the failure paths.

Key Takeaways

  • Declare the topology as beans — RabbitAdmin creates it at startup, so environments cannot drift.
  • Use logical type names in the message converter, not fully-qualified class names.
  • Set default-requeue-rejected: false, or a failing message loops forever.
  • Build the dead-letter topology first; it is not something to add after an incident.
  • Retry with backoff in the container, and dead-letter what backoff cannot fix.

Declaring topology in code

RabbitConfig.java
@Configuration
public class RabbitConfig {
 
    public static final String EXCHANGE = "orders";
    public static final String DLX = "orders.dlx";
    public static final String QUEUE = "q.order.created";
    public static final String DLQ = "q.order.created.dlq";
 
    @Bean TopicExchange orders() { return ExchangeBuilder.topicExchange(EXCHANGE).durable(true).build(); }
    @Bean TopicExchange ordersDlx() { return ExchangeBuilder.topicExchange(DLX).durable(true).build(); }
 
    @Bean
    Queue orderCreated() {
        return QueueBuilder.durable(QUEUE)
                .quorum()
                .deadLetterExchange(DLX)
                .deadLetterRoutingKey("order.created.dead")
                .deliveryLimit(5)
                .build();
    }
 
    @Bean
    Queue orderCreatedDlq() {
        // No TTL and no delivery limit: a DLQ is a place to inspect, not to
        // expire. Losing a dead letter loses the evidence.
        return QueueBuilder.durable(DLQ).quorum().build();
    }
 
    @Bean Binding bindMain(Queue orderCreated, TopicExchange orders) {
        return BindingBuilder.bind(orderCreated).to(orders).with("order.*.created");
    }
 
    @Bean Binding bindDlq(Queue orderCreatedDlq, TopicExchange ordersDlx) {
        return BindingBuilder.bind(orderCreatedDlq).to(ordersDlx).with("order.created.dead");
    }
 
    @Bean
    MessageConverter jsonConverter() {
        var converter = new Jackson2JsonMessageConverter(JsonMapper.builder()
                .addModule(new JavaTimeModule())
                .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
                .build());
 
        // Map a stable logical name to the class. Without this the FQCN travels
        // in the header, so renaming a package breaks every in-flight message
        // and every consumer that has not been redeployed.
        var mapper = new DefaultJackson2JavaTypeMapper();
        mapper.setIdClassMapping(Map.of(
                "OrderCreated", OrderCreated.class,
                "OrderCancelled", OrderCancelled.class));
        mapper.setTrustedPackages("com.acme.events");
        converter.setJavaTypeMapper(mapper);
        return converter;
    }
}

RabbitAdmin declares every exchange, queue and binding bean on startup, so a fresh environment gets the exact topology the code expects. Declaring topology by hand in the management UI works until the day you need to rebuild a cluster.

One caveat on that: declaration is not migration. RabbitAdmin creates what is missing, but an existing queue whose arguments differ from the bean — a changed deliveryLimit, a new dead-letter exchange — causes a PRECONDITION_FAILED rather than an update, because queue arguments are immutable once declared. Changing them means declaring a new queue and moving consumers across, which is worth knowing before you discover it on a Friday deploy.

.quorum() is the other choice worth being deliberate about. Quorum queues replicate through Raft and survive a broker failure with the messages intact, which is what you want for anything that represents work. They cost more memory and disk than classic queues and do not support every feature — notably message TTL per message and priorities — so the reasonable default is quorum for durable business events and classic for the high-volume ephemeral traffic where losing a message is acceptable.

Sending

OrderEventPublisher.java
@Component
public class OrderEventPublisher {
 
    private final RabbitTemplate template;
 
    public void publish(OrderCreated event) {
        template.convertAndSend(RabbitConfig.EXCHANGE,
                "order.%s.created".formatted(event.region()),
                event,
                message -> {
                    var props = message.getMessageProperties();
                    props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                    props.setMessageId(event.eventId());
                    props.setCorrelationId(MDC.get("correlationId"));
                    props.setContentEncoding("UTF-8");
                    // The consumer deduplicates on this.
                    props.setHeader("eventId", event.eventId());
                    return message;
                },
                new CorrelationData(event.eventId()));
    }
}

The CorrelationData argument is what lets the confirm callback tell you which message the broker accepted or rejected. Without it, a negative confirm tells you something failed but not what.

Confirms need enabling before any of that applies — spring.rabbitmq.publisher-confirm-type: correlated and publisher-returns: true. Without them convertAndSend returns as soon as the message reaches the socket, so a broker that is out of disk, or a routing key matching no binding, both look like success from the publisher's side. Returns cover the second case specifically: a message published to an exchange with no matching binding is silently discarded by default, and a return callback is the only thing that tells you it happened.

Receiving

OrderCreatedListener.java
@Component
public class OrderCreatedListener {
 
    private final ProcessedEventRepository processed;
    private final FulfilmentService fulfilment;
 
    @RabbitListener(queues = RabbitConfig.QUEUE, concurrency = "3-10")
    @Transactional
    public void onOrderCreated(@Payload OrderCreated event,
                               @Header("eventId") String eventId,
                               Message raw) {
        MDC.put("correlationId", eventId);
        try {
            try {
                processed.save(new ProcessedEvent(eventId, Instant.now()));
            } catch (DataIntegrityViolationException duplicate) {
                return;   // already handled; AUTO ack still acknowledges
            }
 
            fulfilment.begin(event);
 
        } catch (InvalidOrderException ex) {
            // Permanent: no retry will fix a malformed order. AmqpRejectAndDontRequeueException
            // bypasses the retry policy and dead-letters immediately.
            throw new AmqpRejectAndDontRequeueException("invalid order " + event.orderId(), ex);
 
        } catch (DataAccessResourceFailureException ex) {
            // Transient: let the retry interceptor handle it.
            throw ex;
 
        } finally {
            MDC.clear();
        }
    }
}

Distinguishing the two exception paths is the whole of good AMQP error handling. A permanent failure retried four times wastes ten seconds and still dead-letters; a transient one dead-lettered immediately loses a message that would have succeeded a second later.

Retry and dead-lettering

Bounded in-process retry, then dead-letter. The delivery limit on the quorum queue is the backstop if this configuration is wrong.
application.yml
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto
        prefetch: 10
        # Critical: with true, a rejected message returns to the head of the
        # queue and is redelivered immediately, forever.
        default-requeue-rejected: false
        retry:
          enabled: true
          initial-interval: 1s
          multiplier: 2
          max-interval: 20s
          max-attempts: 4

These retries are in-process — the container holds the message and re-invokes the listener, blocking that consumer thread for the backoff. With prefetch 10 and concurrency 3, a few messages in backoff can occupy the whole consumer pool.

For long backoffs, move the delay into the broker instead using a TTL queue that dead-letters back to the main exchange: publish the failure to q.order.retry with a message TTL of 30 seconds and a dead-letter exchange pointing at orders. When the TTL expires, RabbitMQ routes it back for another attempt, and no consumer thread was blocked meanwhile.

Replaying dead letters

Write this before you need it:

DlqReplayService.java
@Service
public class DlqReplayService {
 
    private final RabbitTemplate template;
 
    /** Move up to `limit` messages from the DLQ back to the main exchange. */
    public int replay(int limit) {
        int moved = 0;
        for (int i = 0; i < limit; i++) {
            Message message = template.receive(RabbitConfig.DLQ, 1_000);
            if (message == null) break;
 
            var death = (List<Map<String, ?>>) message.getMessageProperties()
                    .getHeaders().get("x-death");
            String originalKey = (String) ((List<?>) death.get(0).get("routing-keys")).get(0);
 
            template.send(RabbitConfig.EXCHANGE, originalKey, message);
            moved++;
        }
        return moved;
    }
}

The x-death header carries the original queue, routing key, reason and count — everything needed to route the message back where it came from. Replaying is safe precisely because consumers are idempotent; if they are not, replay is a second incident.

Testing

OrderListenerTest.java
@SpringBootTest
@Testcontainers
class OrderCreatedListenerTest {
 
    @Container
    @ServiceConnection
    static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.13-management");
 
    @Autowired RabbitTemplate template;
    @Autowired FulfilmentRepository repository;
 
    @Test
    void deadLettersAnInvalidOrder() {
        template.convertAndSend("orders", "order.eu.created", new OrderCreated("BAD", List.of()));
 
        await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
            Message dead = template.receive("q.order.created.dlq");
            assertThat(dead).isNotNull();
        });
    }
}

Testing the dead-letter path matters more than testing the happy path, because the dead-letter topology is declarative configuration that is easy to get subtly wrong and produces no error when it is — messages just vanish.

What to take away

Declare the whole topology as beans so environments cannot drift. Use logical type names in the converter. Set default-requeue-rejected: false and throw AmqpRejectAndDontRequeueException for permanent failures. Build the DLQ and a replay path up front, and test that messages actually reach it.

Frequently Asked Questions

Why is my consumer stuck redelivering the same message?
A rejected message with requeue=true goes back to the head of the queue and is immediately redelivered, forever. Set default-requeue-rejected to false so failures are dead-lettered, and use a delivery limit on quorum queues as a second guard.
Do I need manual acknowledgement?
Spring AMQP defaults to AUTO, which acknowledges after your listener returns normally and rejects when it throws — that is genuinely correct for most listeners. Switch to MANUAL only when you need to acknowledge at a different point, such as after an async callback completes.
Why does my JSON message fail to deserialise?
Jackson2JsonMessageConverter reads a __TypeId__ header naming the class. If the producer and consumer disagree on package names it fails. Either configure a type mapping between logical names and classes, or set a default type on the consumer and ignore the header entirely.

Related tutorials