SpringBoot RabbitMQ/RocketMQ Integration: Reliable Delivery, Idempotent Consumption & Ordering Guarantees

1. A Real Production Incident

During a Double 11 flash sale, an e-commerce platform experienced a critical failure: 800,000 payment success messages accumulated in the message queue, and some messages were lost entirely — payments succeeded but order statuses never updated. Root cause analysis revealed three severe problems:

Unacknowledged messages : Consumers used autoAck=true, causing messages to be acknowledged immediately upon delivery; if processing failed, the message was lost.

Insufficient consumption capacity : Each message required 200 ms to process; a single-threaded consumer handled only 5 messages per second, meaning 800,000 messages would take 44 hours.

Duplicate consumption : Already-processed messages were redelivered, causing duplicate shipments for the same order.

Incident retrospective :

Message loss : autoAck=true prevented requeue on processing failure

Insufficient consumption capacity : Single-threaded consumer could not handle peak traffic

Missing idempotency : Redelivered messages were processed repeatedly

No monitoring : Backlog of 800,000 messages discovered too late

Lesson : MQ is easy to use but hard to use correctly. Reliable delivery, idempotent consumption, ordering guarantees, and backlog monitoring — every link is a potential pitfall.

2. Four Core MQ Challenges

2.1 Message Loss

Messages can be lost at three stages:

Producer → Broker → Consumer
   ↑          ↑         ↑
may lose   may lose  may lose

Producer stage : Network jitter, broker not acknowledging → Solution: Confirm mechanism to confirm message reached broker

Broker stage : Not persisted, disk failure → Solution: Persistence — write messages to disk

Consumer stage : autoAck=true, processing failure → Solution: Manual ACK — acknowledge only after successful processing

2.2 Message Duplication

MQ guarantees At Least Once , not Exactly Once . Duplication scenarios:

Producer retries causing duplicate sends

Consumer processes but ACK fails, broker redelivers

Network jitter causes ACK loss

Conclusion : Duplication is unavoidable; only idempotent consumption can solve it.

2.3 Message Ordering

Producer sends sequentially: Message A → Message B → Message C
If multiple queues or consumers are used:
Queue 1: A → C
Queue 2: B
Consumers may process A → B → C or A → C → B
Order broken!

Three conditions for ordering guarantee :

Producer sends in order

Same business messages route to the same queue

Same queue has only one consumer (or consumer processes serially)

2.4 Message Backlog

Cause: Slow consumer processing → Solution: Add consumers, optimize logic

Cause: Consumer logic infinite loop → Solution: Exception handling + dead letter queue

Cause: Traffic spikes → Solution: Rate limiting, degradation, async processing

2.5 RabbitMQ vs RocketMQ

Language : RabbitMQ (Erlang) vs RocketMQ (Java)

Throughput : RabbitMQ (10K level) vs RocketMQ (100K level)

Latency : RabbitMQ (Microsecond) vs RocketMQ (Millisecond)

Ordered messages : RabbitMQ requires single queue + single consumer; RocketMQ supports partition ordering

Delayed messages : RabbitMQ via dead letter queue; RocketMQ native 18 levels

Transaction messages : Both supported

Best fit : RabbitMQ for small/medium scale, complex routing; RocketMQ for large scale, high throughput, e-commerce

Selection advice : Small/medium → RabbitMQ (lightweight, good ecosystem); Large scale → RocketMQ (high performance, full features). This article focuses on RabbitMQ with RocketMQ differences noted separately.

3. Core Technical Principles

3.1 RabbitMQ Core Concepts

Producer : Message producer

Consumer : Message consumer

Exchange : Routes messages to queues

Queue : Stores messages

Binding : Rules connecting Exchange and Queue

Virtual Host : Isolates different businesses

Four Exchange Types :

Direct : Exact match on routingKey — Point-to-point, task distribution

Topic : Wildcard match — Flexible routing, multi-scenario

Fanout : Broadcast to all queues — Notifications, broadcasting

Headers : Match on headers — Complex conditional routing

3.2 Complete Reliable Delivery Chain

① Producer → Broker
   - Enable confirm mode
   - Broker confirms receipt

② Broker Persistence
   - Exchange persistence
   - Queue persistence
   - Message deliveryMode=2 (persistent)

③ Broker → Consumer
   - Manual ACK mode
   - On success call basicAck
   - On failure call basicNack, message requeued

④ Consumption Failure Retry
   - Local retry (3 times)
   - Retry exhausted → dead letter queue

3.3 Idempotent Consumption Principle

Core idea : Deduplicate using business unique ID .

Approach 1: Database unique index
  - Insert unique ID before processing; duplicate insert fails, skip processing

Approach 2: Redis SET NX
  - SET message:id:{msgId} 1 NX EX 86400
  - Success → process; failure → skip

Approach 3: State machine
  - Check business state before processing; already processed → skip

This article uses: Redis SET NX + business state check for double safety .

3.4 Ordered Consumption Implementation

RabbitMQ ordered consumption :

Condition 1: Single queue
Condition 2: Single consumer
Condition 3: Producer sends in order

RocketMQ ordered consumption :

Use MessageQueueSelector to route same business messages to same queue
Consumer uses MessageListenerOrderly for ordered consumption

4. Implementation Solutions & Scenarios

4.1 Article's Solutions

Reliable delivery : Confirm + Persistence + Manual ACK

Consumption failure retry : Local retry (3×) + Dead letter queue

Idempotent consumption : Redis SET NX + Business state check

Ordered consumption : Single queue + Single consumer (partitioned by business)

Backlog monitoring : Scheduled queue length check + DingTalk alert

4.2 Applicable Scenarios

Order async processing : Send SMS, notifications after order placement

Payment result notification : Async update order status after payment success

Inventory deduction : Async inventory deduction after order creation

Data synchronization : Cross-system async data sync

5. Complete Implementation

5.1 Project Structure

heyou-rabbitmq/
├── pom.xml
└── src/main/java/com/heyou/mq/
    ├── MqApplication.java
    ├── config/
    │   ├── RabbitMQConfig.java    # MQ configuration
    │   └── RedisConfig.java
    ├── producer/
    │   └── OrderMessageProducer.java  # Producer
    ├── consumer/
    │   ├── OrderMessageConsumer.java  # Consumer
    │   └── DeadLetterConsumer.java    # Dead letter consumer
    ├── service/
    │   └── OrderService.java
    └── controller/
        └── OrderController.java

5.2 Maven Dependencies

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</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-data-redis</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
  </dependency>
  <dependency>
    <groupId>com.mysql</groupId>
    <artifactId>mysql-connector-j</artifactId>
    <scope>runtime</scope>
  </dependency>
  <dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
  </dependency>
</dependencies>

5.3 RabbitMQ Configuration

package com.example.mq.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitMQConfig {

    // ========== Business Queue ==========
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_ROUTING_KEY = "order.create";

    // ========== Dead Letter Queue ==========
    public static final String DLX_EXCHANGE = "dlx.exchange";
    public static final String DLX_QUEUE = "dlx.queue";
    public static final String DLX_ROUTING_KEY = "dlx.routing";

    /**
     * Business queue (bound to dead letter exchange)
     */
    @Bean
    public Queue orderQueue() {
        Map<String, Object> args = new HashMap<>();
        // Message TTL: 30 minutes
        args.put("x-message-ttl", 1800000);
        // Dead letter exchange
        args.put("x-dead-letter-exchange", DLX_EXCHANGE);
        // Dead letter routing key
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
        return new Queue(ORDER_QUEUE, true, false, false, args);
    }

    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE, true, false);
    }

    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
                .to(orderExchange())
                .with(ORDER_ROUTING_KEY);
    }

    // ========== Dead Letter Queue Config ==========

    @Bean
    public Queue dlxQueue() {
        return new Queue(DLX_QUEUE, true);
    }

    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(DLX_EXCHANGE, true, false);
    }

    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue())
                .to(dlxExchange())
                .with(DLX_ROUTING_KEY);
    }

    /**
     * Configure RabbitTemplate (enable confirm)
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // Enable confirm
        template.setMandatory(true);
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("Message sent successfully: " + correlationData);
            } else {
                System.err.println("Message send failed: " + cause);
            }
        });
        return template;
    }
}

5.4 Producer

package com.example.mq.producer;

import com.example.mq.config.RabbitMQConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderMessageProducer {

    private final RabbitTemplate rabbitTemplate;

    /**
     * Send order message
     * Uses message unique ID + Confirm mechanism for reliable delivery
     */
    public void sendOrderMessage(Long orderId, String message) {
        // Generate unique message ID (for idempotent consumption)
        String messageId = UUID.randomUUID().toString().replace("-", "");

        // Message content
        String content = orderId + ":" + message;

        // Correlation data (for Confirm callback)
        CorrelationData correlationData = new CorrelationData(messageId);

        // Send message
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.ORDER_EXCHANGE,
                RabbitMQConfig.ORDER_ROUTING_KEY,
                content,
                message -> {
                    // Set message persistence
                    message.getMessageProperties().setDeliveryMode(
                            org.springframework.amqp.core.MessageDeliveryMode.PERSISTENT);
                    // Set message ID
                    message.getMessageProperties().setMessageId(messageId);
                    return message;
                },
                correlationData
        );

        log.info("Order message sent: orderId={}, messageId={}", orderId, messageId);
    }
}

5.5 Consumer (Core)

package com.example.mq.consumer;

import com.example.mq.config.RabbitMQConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@RequiredArgsConstructor
public class OrderMessageConsumer {

    private final StringRedisTemplate redisTemplate;
    private final JdbcTemplate jdbcTemplate;

    private static final String IDEMPOTENT_PREFIX = "mq:idempotent:";

    /**
     * Consume order message
     * Reliability guarantees:
     * 1. Manual ACK - acknowledge only on success
     * 2. Idempotent consumption - Redis SET NX prevents duplicates
     * 3. Business state check - skip if already processed
     * 4. Failure retry - local retry 3×, then dead letter queue
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void onOrderMessage(Message message, Channel channel) throws IOException {
        String messageId = message.getMessageProperties().getMessageId();
        String content = new String(message.getBody());
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        log.info("Received order message: messageId={}, content={}", messageId, content);

        try {
            // 1. Idempotency check (Redis SET NX)
            String idempotentKey = IDEMPOTENT_PREFIX + messageId;
            Boolean firstProcess = redisTemplate.opsForValue()
                    .setIfAbsent(idempotentKey, "1", 24, TimeUnit.HOURS);

            if (Boolean.FALSE.equals(firstProcess)) {
                log.warn("Message already processed, skipping: messageId={}", messageId);
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 2. Parse message
            String[] parts = content.split(":");
            Long orderId = Long.parseLong(parts[0]);

            // 3. Business state check (prevent paid order from being cancelled)
            Integer status = jdbcTemplate.queryForObject(
                    "SELECT status FROM orders WHERE id = ?",
                    Integer.class, orderId);

            if (status == null) {
                log.warn("Order does not exist, ignoring message: orderId={}", orderId);
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 4. Execute business logic
            processOrder(orderId, status, parts[1]);

            // 5. Manual ACK (processing succeeded)
            channel.basicAck(deliveryTag, false);
            log.info("Order message processed successfully: messageId={}", messageId);

        } catch (Exception e) {
            log.error("Order message processing failed: messageId={}", messageId, e);

            // 6. Determine if retry needed
            // Get retry count from message header
            Integer retryCount = (Integer) message.getMessageProperties()
                    .getHeaders().get("x-retry-count");
            if (retryCount == null) retryCount = 0;

            if (retryCount < 3) {
                // Retry: reject and requeue (with retry marker)
                log.info("Message retry attempt {}: messageId={}", retryCount + 1, messageId);
                channel.basicNack(deliveryTag, false, true);
            } else {
                // Retries exhausted, reject to dead letter queue
                log.error("Message retried over 3 times, entering dead letter queue: messageId={}", messageId);
                channel.basicNack(deliveryTag, false, false);
            }
        }
    }

    /**
     * Business processing logic
     */
    private void processOrder(Long orderId, Integer status, String action) {
        if ("cancel".equals(action)) {
            // Cancel order
            if (status == 0) {
                jdbcTemplate.update(
                        "UPDATE orders SET status = 2, cancel_time = NOW() " +
                        "WHERE id = ? AND status = 0",
                        orderId);
                log.info("Order cancelled: orderId={}", orderId);
            } else {
                log.info("Order status does not allow cancellation: orderId={}, status={}", orderId, status);
            }
        }
    }
}

5.6 Dead Letter Consumer

package com.example.mq.consumer;

import com.example.mq.config.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;

@Slf4j
@Component
public class DeadLetterConsumer {

    /**
     * Consume dead letter queue messages
     * Typically used for manual intervention or special handling
     */
    @RabbitListener(queues = RabbitMQConfig.DLX_QUEUE)
    public void onDeadLetter(Message message, Channel channel) throws IOException {
        String messageId = message.getMessageProperties().getMessageId();
        String content = new String(message.getBody());

        log.error("Received dead letter message: messageId={}, content={}, time={}",
                messageId, content, LocalDateTime.now());

        // Handle dead letter: record to DB, send alert, manual processing
        try {
            // 1. Record to dead letter table
            // 2. Send DingTalk alert
            // 3. Manual intervention

            // ACK dead letter message
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("Dead letter processing failed", e);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }
}

5.7 Configuration

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

    # ========== Producer Config ==========
    publisher-confirm-type: correlated   # Enable confirm
    publisher-returns: true              # Enable return

    # ========== Consumer Config ==========
    listener:
      simple:
        acknowledge-mode: manual         # Manual ACK
        prefetch: 10                     # Prefetch 10 messages
        concurrency: 5                   # Concurrent consumers
        max-concurrency: 10              # Max concurrent
        retry:
          enabled: false                 # Disable Spring Retry, use manual retry

  redis:
    host: localhost
    port: 6379

  datasource:
    url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false
    username: root
    password: 123456

5.8 Database Table

CREATE TABLE `orders` (
    `id` BIGINT NOT NULL AUTO_INCREMENT,
    `user_id` BIGINT NOT NULL,
    `product_id` VARCHAR(64) NOT NULL,
    `status` TINYINT DEFAULT 0 COMMENT '0-pending 1-paid 2-cancelled',
    `create_time` DATETIME DEFAULT CURRENT_TIMESTAMP,
    `cancel_time` DATETIME DEFAULT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

5.9 Controller

package com.example.mq.controller;

import com.example.mq.producer.OrderMessageProducer;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/order")
@RequiredArgsConstructor
public class OrderController {

    private final OrderMessageProducer producer;

    @PostMapping("/send")
    public String sendOrderMessage(@RequestParam Long orderId,
                                   @RequestParam String action) {
        producer.sendOrderMessage(orderId, action);
        return "Message sent";
    }
}

6. Advanced Optimizations

6.1 Consumer Concurrency Improvement

Increase consumption capacity via concurrency parameter:

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 10          # 10 consumers
        max-concurrency: 20      # Max 20 consumers
        prefetch: 20             # Prefetch 20 messages

Note : Concurrent consumption causes message ordering loss . If business requires ordered consumption, must use single consumer .

6.2 Ordered Consumption

RabbitMQ ordered consumption :

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 1           # Single consumer
        prefetch: 1              # Prefetch 1, process then fetch next

RocketMQ ordered consumption :

// Producer: route by business ID to same queue
producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Long orderId = (Long) arg;
        int index = (int) (orderId % mqs.size());
        return mqs.get(index);
    }
}, orderId);

// Consumer: ordered consumption
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            processMessage(msg);
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

6.3 Message Backlog Monitoring

@Component
@Slf4j
public class QueueMonitor {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(fixedDelay = 60000)
    public void monitor() {
        // Get queue message count (via RabbitMQ Management API)
        // Or query via AmqpAdmin
        Properties props = rabbitTemplate.execute(channel -> {
            return channel.getConnection().getServerProperties();
        });

        log.info("Queue monitoring: {}", props);

        // If backlog exceeds threshold, send alert
        // RabbitMQ Management HTTP API gives more accurate queue info
    }
}

6.4 RocketMQ Integration Highlights

RocketMQ core differences :

Ordered messages : Native partition ordering support

Delayed messages : Native 18 delay levels (1s, 5s, 10s...2h)

Transaction messages : Native half-message mechanism

Message retry : Auto retry 16×, then dead letter topic

Consumption mode : Cluster / Broadcast consumption

RocketMQ delayed message example :

Message message = new Message("topic", "tag", "Order timeout cancel".getBytes());
// Delay 30 minutes (level 14)
message.setDelayTimeLevel(14);
producer.send(message);

RocketMQ delay levels :

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

6.5 Message Tracing

Enable message tracing for troubleshooting:

# RabbitMQ
spring:
  rabbitmq:
    publisher-returns: true
    template:
      mandatory: true

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setReturnsCallback(returned -> {
        log.error("Message routing failed: exchange={}, routingKey={}, replyText={}",
                returned.getExchange(), returned.getRoutingKey(), returned.getReplyText());
    });
    return template;
}

7. Pitfall Guide

Message loss : Symptom — Message disappears after processing failure → Solution: Manual ACK + Persistence + Dead letter queue

Message duplication : Symptom — Same order processed multiple times → Solution: Redis SET NX + Business state check

Consumption order chaos : Symptom — Same order messages out of order → Solution: Single queue + Single consumer (or partition ordering)

Slow consumer : Symptom — Message backlog → Solution: Add concurrent consumers, optimize logic

ACK forgotten : Symptom — Messages never acknowledged → Solution: Use try-finally to guarantee ACK

Infinite retry loop : Symptom — Consumption failure retries endlessly → Solution: Limit retries, move to dead letter queue

Oversized message body : Symptom — Broker rejects → Solution: Keep body <1MB, large files via OSS

Connection leak : Symptom — Consumers cannot connect → Solution: Use connection pool, avoid per-request creation

8. Production Recommendations

Small/medium scale (<100K/day) : RabbitMQ + Manual ACK + Dead letter queue

Large scale (>1M/day) : RocketMQ + Partition ordered consumption

Need delayed messages : RocketMQ (native support)

Need ordered messages : RocketMQ partition ordering

Need transaction messages : RocketMQ transaction messages

Parameter recommendations :

prefetch : 10-20 — Prefetch count

concurrency : 5-10 — Concurrent consumers

max-concurrency : 20 — Max concurrent

Max retries : 3 — Local retry + dead letter

Dead letter retention : 7 days — Manual handling window

Queue monitor interval : 1 minute — Detect backlog early

9. Summary & Reflection

Three core takeaways :

Reliable delivery = Confirm + Persistence + Manual ACK , all three layers essential.

Duplication is inevitable; guarantee idempotency via Redis SET NX + Business state check .

Ordered consumption requires Single queue + Single consumer (or RocketMQ partition ordering).

Quick reference :

No message loss : Confirm + Persistence + Manual ACK

Idempotent consumption : Redis SET NX + Business state check

Ordered consumption : Single consumer or RocketMQ partition ordering

Delayed messages : RocketMQ delay levels or Dead letter queue

Message backlog : Add consumers + Monitoring alerts