Integrating RabbitMQ with Spring Boot: Configuration, Message Sending, and Reliability

1. Introduction

Message middleware such as RabbitMQ is commonly used for asynchronous processing, system decoupling, and traffic shaping. This article uses RabbitMQ as an example to discuss integration with Spring Boot and related reliability concerns.

2. Integration with Spring Boot

2.1 Add Dependency

Include the Spring Boot AMQP starter in the pom.xml:

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

2.2 MQ Service Configuration

Configure the RabbitMQ connection in application.yml:

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: boot-example

2.3 Inject Message Template

@Autowired
private RabbitTemplate rabbitTemplate;

2.4 Send Message

public void sendMessage() {
    rabbitTemplate.convertAndSend("test", "test", "mq produce send a message");
}

2.5 Consume Message

@Component
@Slf4j
public class MqConsumer {
    @RabbitListener(id = "consumerMessage1", queues = "test")
    public void consumeMessage1(Message message, Channel channel, String content) {
        log.info("receive message1 ：{}", content);
    }
}

3. Sending Complex Messages

When interacting with other systems using JSON objects, a message converter is required.

3.1 Producer Message Converter

@Bean
public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate();
    configurer.configure(rabbitTemplate, connectionFactory);
    rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
    return rabbitTemplate;
}

3.2 Consumer Message Converter

@Bean
public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(new Jackson2JsonMessageConverter());
    return factory;
}

3.3 Specify Listener Container Factory

@RabbitListener(queues = "test3", containerFactory = "rabbitListenerContainerFactory")
public void consumeComplexMessage(Order order) {
    log.info("receive complex message：{}", order);
}

4. Message Sending Reliability

4.1 Why Reliability Is Needed

RabbitMQ may lose persistent messages if the broker crashes before writing them to disk. Enabling publisher confirms ensures the broker acknowledges only after the message is safely stored.

4.2 How to Guarantee Reliability

4.2.1 Configuration

rabbitmq:
  publisher-confirm-type: correlated
  publisher-returns: true

4.2.2 Callback Implementation

@Configuration
@Slf4j
public class MqConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("receive ack confirm：{} from broker server", ack);
    }
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("receive return message：{} from broker server，reply code：{}，reply text：{}，exchange：{}，routing key：{}",
                message.toString(), replyCode, replyText, exchange, routingKey);
    }
    @Bean
    public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        configurer.configure(rabbitTemplate, connectionFactory);
        rabbitTemplate.setReturnCallback(this);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
}

4.2.3 Confirm vs ReturnedMessage

For unroutable messages the broker sends a basic.return before the basic.ack. Therefore both callbacks are needed to detect routing failures; if routing is guaranteed, the confirm callback alone suffices.

5. Consumer Reliability

5.1 Why It Matters

If a message is removed from the queue before the consumer acknowledges it, failures such as connection loss can cause message loss. Using manual acknowledgements ensures the broker only deletes a message after a positive ack.

5.2 How to Ensure Reliability

boolean autoAck = false;
channel.basicConsume(queueName, autoAck, "a-consumer-tag",
    new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                                 AMQP.BasicProperties properties, byte[] body) throws IOException {
            long deliveryTag = envelope.getDeliveryTag();
            // process message ...
            channel.basicAck(deliveryTag, false); // or basicReject for negative ack
        }
    });

Spring Boot abstracts this with the acknowledge-mode property; setting it to MANUAL enables the same behaviour.

6. Consumer Scalability

When multiple consumers share a queue, RabbitMQ distributes messages in a round‑robin fashion. Adding more consumer instances increases overall consumption capacity, providing horizontal scalability.

7. Traffic Shaping with Consumers

By configuring QoS (prefetch count) via basic.qos, a consumer can limit the number of unacknowledged messages it holds, preventing overload and achieving traffic throttling.

8. References

Consumer Acknowledgements and Publisher Confirms