RocketMQ's Core Features

RocketMQ offers several key features:

1. Support for transactional messages
2. Support for delayed messages
3. Native support for clustering and load balancing
4. Failed message redelivery with configurable retry count and interval

RocketMQ's Components

RocketMQ consists of four main components:

• Producer Cluster: Generates business messages and sends them to brokers, retrieving broker addresses from NameServer
• NameServer Cluster: Coordinates cluster operations (acts as a registry), storing and distributing broker metadata without processing messages
• Broker Cluster: Core component handling message reception, storage, and distribution
• Consumer Cluster: Handles asynchronous message consumption, typically from backend systems

Configuration and Startup

JVM Configuration (runserver.sh)

# Default 4GB heap; reduce for development
-Xmx4g  # Maximum heap size
-Xms4g  # Minimum heap size
-Xmn2g  # Young generation size

Startup Commands

# Start NameServer
nohup sh mqnamesrv &

# Start Broker and connect to NameServer
nohup sh mqbroker -n localhost:9876 &

# Set NameServer address for testing
export NAMESRV_ADDR=127.0.0.1:9876

# Run built-in quickstart producer
bash tools.sh org.apache.rocketmq.example.quickstart.Producer

Message Sending Modes and Structure

Sending Modes

1. Synchronous Messages

The producer waits for a response from the broker before continuing execution. This guarantees reliable delivery and immediate feedback.

2. Asynchronous Messages

The producer returns immediately after sending and invokes a callback (success/failure) in a new thread. This improves throughput.

3. One-way Messages

The producer sends the message and returns without waiting for any response. This is fastest but has no reliability guarantees.

Message Structure

Category	Properties
Basic	topic (primary classification), body (content, max 4MB), flag (business-defined)
Extended	tag (secondary classification for filtering), keys (searchable indexes), waitStoreMsgOK (wait for storage before returning)

Simple Implementation with Spring Boot

Producer Code

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
public class RocketMQMessageProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendSynchronousMessage(String topic, String content) {
        rocketMQTemplate.syncSend(topic, content, 100000);
    }

    public void sendAsynchronousMessage(String topic, String content) {
        rocketMQTemplate.asyncSend(topic, content, new SendCallback() {
            @Override
            public void onSuccess(SendResult result) {
                System.out.println("Message sent successfully: " + result);
            }
            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        }, 100000);
    }

    public void sendDelayedMessage(String topic, Object payload, int delayLevel) {
        Message<Object> message = MessageBuilder.withPayload(payload).build();
        rocketMQTemplate.syncSend(topic, message, 1000, delayLevel);
        System.out.printf("Delayed message sent: %s", payload);
    }
}

Producer Configuration

server:
  port: 8181
  servlet:
    context-path: /rocketmq-producer
spring:
  application:
    name: rocketmq-producer
rocketmq:
  producer:
    group: demo-producer-group
  name-server: 49.52.10.41:9876

Producer Test

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RocketMQMessageProducerTest {
    @Autowired
    private RocketMQMessageProducer producer;

    @Test
    public void testSynchronousMessage() {
        producer.sendSynchronousMessage("testTopic", "Synchronous message #1");
        System.out.println("Test completed.");
    }

    @Test
    public void testAsynchronousMessage() {
        producer.sendAsynchronousMessage("testTopic", "Asynchronous message #1");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Consumer Code

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

@Component
@RocketMQMessageListener(topic = "testTopic", consumerGroup = "demo-consumer-group")
public class RocketMQMessageConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

Consumer Configuration

server:
  port: 8182
  servlet:
    context-path: /rocketmq-consumer
spring:
  application:
    name: rocketmq-consumer
rocketmq:
  consumer:
    group: demo-consumer-group
  name-server: 49.52.10.41:9876

Message Delivery Flow and Consumption Modes

Delivery Process

1. Producers retrieve topic-to-queue mapping from NameServer
2. Messages are sent to selected queues after validation
3. Consumers continuously listen to queues and process messages

Cluster vs Broadcast Mode

• Cluster Mode (Point-to-Point): A message is consumed by only one consumer in a group
• Broadcast Mode (Publish-Subscribe): A message is consumed by all consumers in a group

Push vs Pull Mode

• Push: Broker active pushes messages to consumers for real-time processing
• Pull: Consumers periodically pull messages from brokers for controlled consumption

Delayed Messages

Application Scenario: Order Auto-Close

When a user creates an unpaid order, send a delayed message that triggers order closure after a specified delay if payment fails.

Implementation

RocketMQ uses 18 predefined delay levels (1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h). Messages are temporarily stored in SCHEDULE_TOPIC_XXXX queues and re-delivered at the specified time.

Delay Message Flow

1. For delayed messages, the topic is temporarily changed to SCHEDULE_TOPIC_XXXX and queue ID = delay level - 1
2. Scheduled tasks pull messages from these queues
3. Original topic and queue ID are restored before delivery
4. Message are sent to the target topic for consumption

Consumption Retries

When to Retry

• Messages not received due to network errors
• Messages received but failed processing with exceptions

How Brokers Determine Success/Failure

Brokers wait for consumption status from consumers; no success response triggers a retry.

Handling Exceptions

• Default policy: Retries start at level 3 (10s) and escalate, with up to 16 retries
• After 16 failures, messages are moved to the dead letter queue (%DLQ%+consumerGroup)

Custom Retry Strategy

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

@Component
@RocketMQMessageListener(topic = "testTopic", consumerGroup = "demo-consumer-group")
public class CustomRetryMessageConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        int retryCount = message.getReconsumeTimes();
        if (retryCount >= 2) {
            System.out.println("Storing failed message for manual processing: " + new String(message.getBody()));
            return;
        }
        throw new RuntimeException(String.format("Processing failed for the %dth time", retryCount));
    }
}