Integrating Spring Boot with RocketMQ for Message Production and Consumption

1. Why Choose RocketMQ?

RocketMQ is an open‑source distributed messaging middleware from Alibaba, proven in high‑concurrency scenarios such as Double‑11. Its core advantages are high throughput, high reliability and low latency, which help solve asynchronous decoupling, traffic shaping and data‑consistency problems. Compared with RabbitMQ it is more suitable for high‑concurrency business.

1.1 Core Functions

Asynchronous decoupling : services communicate via messages, avoiding cascade failures.

Traffic shaping : requests are buffered in the queue before being consumed.

Data consistency : transactional messages ensure cross‑service consistency.

Asynchronous notification : e.g., order changes trigger SMS or email without blocking the main flow.

1.2 Typical Scenarios

e‑commerce order flow (order → inventory → payment → logistics → SMS).

High‑concurrency flash‑sale or live‑streaming traffic buffering.

Log/monitoring collection to avoid blocking business threads.

Distributed transactions such as order + inventory.

1.3 Core Components

Producer : sends messages.

Consumer : receives and processes messages.

Topic : logical container for messages.

Message Queue : sub‑queues under a topic for load balancing.

NameServer : stateless registry that maps producers/consumers to topics.

Broker : stores and forwards messages.

2. Environment Preparation

Docker one‑click installation is recommended; alternatively install locally.

2.1 Docker One‑Click

# Pull image
docker pull rocketmqinc/rocketmq:4.9.4
# Run container
docker run -d --name rocketmq -p 9876:9876 -p 10911:10911 -e "ROCKETMQ_NAMESRV_ADDR=localhost:9876" -e "ROCKETMQ_BROKER_ADDR=localhost:10911" rocketmqinc/rocketmq:4.9.4

Verify with docker ps – ports 9876 (NameServer) and 10911 (Broker) should be running.

2.2 Local Installation

Download RocketMQ 4.9.4 binary.

Extract to a path without spaces.

Start NameServer: start mqnamesrv.cmd.

Start Broker:

start mqbroker.cmd -n localhost:9876 autoCreateTopicEnable=true

.

2.3 Maven Dependency

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>${rocketmq.version}</version>
</dependency>

3. Spring Boot Basic Configuration

Add the following to application.yml (replace only if needed):

spring:
  application:
    name: springboot-rocketmq-intro
rocketmq:
  name-server: localhost:9876
  producer:
    group: rocketmq-producer-group
    send-message-timeout: 3000
    retry-times-when-send-failed: 1
  consumer:
    group: rocketmq-consumer-group
    message-model: CLUSTERING
    consume-thread-max: 10
    consume-timeout: 15000

Key notes: name-server must match the running NameServer; producer/consumer groups are used for load‑balancing; message-model CLUSTERING is default.

4. Message Production and Consumption

4.1 Producer (Ordinary Message)

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
public class RocketMQProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    private static final String TOPIC_NAME = "springboot_rocketmq_topic";

    @GetMapping("/send/rocketmq/{msg}")
    public String sendMessage(@PathVariable String msg) {
        String message = "【RocketMQ 普通消息】" + msg;
        rocketMQTemplate.convertAndSend(TOPIC_NAME, message);
        return "RocketMQ 消息发送成功！发送内容：" + message;
    }
}

4.2 Consumer

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "springboot_rocketmq_topic",
        consumerGroup = "${rocketmq.consumer.group}")
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String msg) {
        System.out.println("RocketMQ 消费者接收消息：" + msg);
    }
}

4.3 Test Steps

Start RocketMQ (NameServer + Broker).

Run the Spring Boot project.

Visit http://localhost:8080/send/rocketmq/HelloRocketMQ in a browser.

Observe the success message in the browser.

Check the IDE console – the consumer should print the received message.

4.4 Key Points

Producer and consumer must use the same topic.

Consumer class needs @Component so Spring can manage it. consumerGroup in @RocketMQMessageListener must match the value in application.yml.

Start RocketMQ before the Spring Boot application.

Message types must match (String in this example).

5. Extensions – Two Common Message Types

5.1 Synchronous Message

@GetMapping("/send/rocketmq/sync/{msg}")
public String sendSyncMessage(@PathVariable String msg) {
    String message = "【RocketMQ 同步消息】" + msg;
    SendResult sendResult = rocketMQTemplate.syncSend(TOPIC_NAME, message);
    if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
        return "同步消息发送成功！发送内容：" + message + "，消息ID：" + sendResult.getMsgId();
    } else {
        return "同步消息发送失败！";
    }
}

5.2 Asynchronous Message

@GetMapping("/send/rocketmq/async/{msg}")
public String sendAsyncMessage(@PathVariable String msg) {
    String message = "【RocketMQ 异步消息】" + msg;
    rocketMQTemplate.asyncSend(TOPIC_NAME, message, new RocketMQSendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println("异步消息发送成功，消息ID：" + sendResult.getMsgId());
        }
        @Override
        public void onException(Throwable e) {
            System.out.println("异步消息发送失败，原因：" + e.getMessage());
        }
    });
    return "异步消息发送中，请查看控制台回调结果！发送内容：" + message;
}

6. Common Issues and Solutions

Could not connect to NameServer – ensure RocketMQ is running, verify the name-server address in application.yml, and open ports 9876/10911 on the server.

Message sent but not received – check that producer and consumer use the same topic, that the consumer group matches the configuration, and that the consumer class is annotated with @Component.

Broker out of memory – reduce JVM memory options in runbroker.cmd (e.g., -Xms512m -Xmx512m -Xmn256m) and restart the broker.

Topic does not exist – enable automatic topic creation with autoCreateTopicEnable=true when starting the broker or create the topic manually.

The tutorial concludes by reminding readers that learning is a long‑term process and encourages hands‑on practice.