Spring boot框架下的RocketMQ消息中间件

1. RocketMQ 基础概念

1.1 核心概念

以下是 RocketMQ 核心概念在 Spring Boot 的 Java 后端代码中的实际使用方式：

Producer（生产者）

定义：Producer 是负责发送消息到 RocketMQ 的组件。它可以将消息发送到指定的 Topic。

实际代码： 使用 RocketMQTemplate 提供的 API 发送消息。

@Service
public class DemoProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;public void sendMessage(String topic, String message) {rocketMQTemplate.convertAndSend(topic, message);System.out.println("Message sent to topic: " + topic);}
}

Consumer（消费者）

定义：Consumer 是负责接收和处理从 RocketMQ 中接收到的消息的组件。

实际代码： 使用 @RocketMQMessageListener 注解标注的类来实现消费者功能。

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

Topic（主题）

定义：Topic 是消息分类的基本单元，Producer 将消息发送到 Topic，Consumer 从 Topic 中接收消息。

实际代码： Topic 是通过代码中指定的字符串定义的，例如 demo-topic。

• Producer 中指定 Topic：

  rocketMQTemplate.convertAndSend("demo-topic", "Hello RocketMQ!");
  

• Consumer 中监听 Topic：

  @RocketMQMessageListener(topic = "demo-topic", consumerGroup = "demo-consumer-group")
  

Message（消息）

定义：消息是 Producer 发送到 Topic 的载体，包含消息体和属性。

实际代码示例：

• 发送简单文本消息：

  rocketMQTemplate.convertAndSend("demo-topic", "Hello RocketMQ!");
  

• 发送带属性的消息：

  Message message = MessageBuilder.withPayload("Hello with properties").setHeader("key", "value").build();
  rocketMQTemplate.syncSend("demo-topic", message);
  

---

1.2 消息模型

点对点（Queue）模型

定义：

• 每条消息只能被一个消费者（Consumer）消费。

• 消息会被存储在一个队列（Queue）中，消费者从队列中拉取消息。

• 适合任务分发、负载均衡等场景。

特点：

• 消息独占性：一条消息只能被一个消费者消费。

• 负载均衡：如果有多个消费者属于同一个消费者组（Consumer Group），消息会均匀分配给组内的消费者。

• 消息顺序性：在 RocketMQ 中，队列（Queue）是 Topic 的底层实现，每个 Queue 内的消息是有序的。

实现：

• 在 RocketMQ 中，Queue 是 Topic 的底层实现，开发者无需直接配置 Queue，只需关注 Topic 和 Consumer Group。

• 多个消费者属于同一个 Consumer Group 时，消息会按照负载均衡策略分配给组内的消费者。

---

 发布订阅（Topic）模型

定义：

• 消息会被广播给所有订阅了该 Topic 的消费者。

• 每个消费者都会收到相同的消息。

• 适合广播通知、事件分发等场景。

特点：

• 消息广播：一条消息会被发送给所有订阅了该 Topic 的消费者。

• 消费者独立性：每个消费者独立消费消息，彼此之间没有影响。

• 无负载均衡：所有消费者都会收到相同的消息，而不是分配不同的消息。

实现：

• 在 RocketMQ 中，可以通过设置 MessageModel.BROADCASTING 来实现广播消费。

• 每个消费者都会收到 Topic 中的所有消息。

示例：在 Spring Boot 中实现广播消费

@RocketMQMessageListener(topic = "broadcast-topic",consumerGroup = "broadcast-group",messageModel = MessageModel.BROADCASTING
)
public class BroadcastConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Broadcast received: " + message);}
}

---

两者的核心区别

特性	点对点（Queue）模型	发布订阅（Topic）模型
消息消费方式	一条消息只能被一个消费者消费	一条消息会被所有消费者消费
负载均衡	支持，消息会均匀分配给组内的消费者	不支持，所有消费者收到相同消息
适用场景	任务分发、订单处理等需要负载均衡的场景	广播通知、事件分发等需要广播的场景
消费者组的作用	组内消费者共享消息，实现负载均衡	组内消费者各自独立消费所有消息
消息顺序性	单个 Queue 内的消息是有序的	无顺序性，所有消费者收到相同消息

---

2. 环境准备

2.1 安装与运行 RocketMQ

1. 下载 RocketMQ

• 访问 RocketMQ 的官方 GitHub 仓库：Apache RocketMQ GitHub

• 下载最新版本的 RocketMQ 二进制包（例如：rocketmq-all-x.x.x-bin-release.zip）。

• 解压下载的文件

2. 启动 NameServer

• NameServer 是 RocketMQ 的注册中心，负责管理 Broker 的路由信息。

• 启动 NameServer：

  sh bin/mqnamesrv &

• 检查 NameServer 是否启动成功：查看日志文件 logs/rocketmqlogs/namesrv.log，确认是否有以下内容：The Name Server boot success...

3. 启动 Broker

• Broker 是 RocketMQ 的消息存储和转发服务器。

• 启动 Broker，并指定 NameServer 地址：

  sh bin/mqbroker -n localhost:9876 autoCreateTopicEnable=true &

  • -n localhost:9876：指定 NameServer 的地址。

  • autoCreateTopicEnable=true：允许自动创建 Topic。

• 检查 Broker 是否启动成功：

  • 查看日志文件 logs/rocketmqlogs/broker.log，确认是否有以下内容：The broker boot success...

4. 验证 RocketMQ 运行状态

• 使用 RocketMQ 提供的工具命令检查服务状态：

  sh bin/mqadmin clusterList -n localhost:9876

• 如果看到 Broker 和 NameServer 的信息，说明 RocketMQ 已成功启动。

---

2.2 RocketMQ Dashboard

RocketMQ Dashboard 是一个可视化管理工具，用于监控和管理 RocketMQ 的 Topic、Producer、Consumer 等信息。

1. 下载 RocketMQ Dashboard

• 访问 RocketMQ Dashboard 的官方 GitHub 仓库：RocketMQ Dashboard GitHub

• 下载最新版本的 Dashboard 二进制包（例如：rocketmq-dashboard-x.x.x.jar）。

2. 部署 RocketMQ Dashboard

• 使用以下命令启动 Dashboard：

  java -jar rocketmq-dashboard-x.x.x.jar

• 默认情况下，Dashboard 会监听 8080 端口。

3. 访问 RocketMQ Dashboard

• 打开浏览器，访问 http://localhost:8080。

• 在 Dashboard 页面中，配置 NameServer 地址（例如：localhost:9876），然后点击连接。

4. 监控和管理 RocketMQ

• Topic 管理：查看所有 Topic 的详细信息，包括消息堆积情况、生产者、消费者等。

• Producer 管理：查看生产者的运行状态和消息发送情况。

• Consumer 管理：查看消费者的运行状态和消息消费情况。

• 消息查询：支持按 Message ID、Topic、Key 等条件查询消息。

• 集群状态：查看 NameServer 和 Broker 的运行状态。

3. Spring Boot 集成 RocketMQ 的依赖配置 

3.1 添加依赖

在 pom.xml 文件中添加 Spring Boot 提供的 RocketMQ 集成 Starter 依赖：

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.1</version> <!-- 确保版本为你需要的最新稳定版 -->
</dependency>

3.2 配置文件

在 application.properties 或 application.yml 中配置 RocketMQ 的相关属性。

# 指定 RocketMQ NameServer 地址
rocketmq.name-server=127.0.0.1:9876# 配置生产者组名称
rocketmq.producer.group=demo_producer_group# 配置消费者组名称
rocketmq.consumer.group=demo_consumer_group

配置项说明

配置项	说明
rocketmq.name-server	RocketMQ NameServer 的地址，格式为 IP:Port，例如 127.0.0.1:9876。 支持集群地址，多个地址用分号分隔，例如：127.0.0.1:9876;127.0.0.2:9876。
rocketmq.producer.group	生产者组的名称，用于标识一组生产者。
rocketmq.consumer.group	消费者组的名称，用于标识一组消费者。
rocketmq.access-key	如果使用阿里云 RocketMQ，需要配置 Access Key。
rocketmq.secret-key	如果使用阿里云 RocketMQ，需要配置 Secret Key。
rocketmq.send-message-timeout	生产者发送消息的超时时间，默认值为 3000（单位：毫秒）。
rocketmq.consumer.consume-thread-nums	消费者消费消息的线程数，默认值为 20。
rocketmq.consumer.maxReconsumeTimes	自定义重试次数，消息消费失败会自动重试，默认最大重试次数为 16 次（包括第一次消费）

4. Spring Boot 集成 RocketMQ 的消息生产与消费

RocketMQTemplate 是否需要定义为 Bean？

• 不需要手动定义：在 Spring Boot 项目中，RocketMQTemplate 已经由 rocketmq-spring-boot-starter 自动配置为 Bean，因此可以直接通过 @Autowired 注入使用。

• 自动配置原理：

  • rocketmq-spring-boot-starter 会在启动时自动创建 RocketMQTemplate 的实例。

  • 只要在 application.properties 或 application.yml 中正确配置了 RocketMQ 的相关参数（如 rocketmq.name-server），Spring Boot 就会自动完成初始化。

发送非 String 类型的消息

• 支持任意类型：RocketMQTemplate 支持发送任意类型的消息，包括 POJO（普通 Java 对象）。

• 实现方式：

  • RocketMQ 会自动将消息序列化为 JSON 格式（默认使用 Jackson）。

  • 消费者接收时，RocketMQ 会自动将 JSON 反序列化为对应的 Java 对象。

示例代码

发送 POJO 消息：

public class User {private String name;private int age;// 省略 getter 和 setter
}// 生产者
public void sendUserMessage(String topic, User user) {rocketMQTemplate.convertAndSend(topic, user);System.out.println("Sent user message: " + user);
}

消费 POJO 消息：

@Service
@RocketMQMessageListener(topic = "user_topic",consumerGroup = "user_consumer_group"
)
public class UserConsumer implements RocketMQListener<User> {@Overridepublic void onMessage(User user) {System.out.println("Received user message: " + user.getName() + ", age: " + user.getAge());}
}

4.1 消息生产（Producer）

一般会定义一个消息生产者类，然后其他类调用这个类包装好的消息发送方法。

发送普通消息

使用场景：发送简单的文本或对象消息。

实现方式：使用 RocketMQTemplate.convertAndSend(topic, message) 方法发送消息。

示例代码：

@Autowired
private RocketMQTemplate rocketMQTemplate;public void sendSimpleMessage(String topic, String message) {rocketMQTemplate.convertAndSend(topic, message);System.out.println("Sent simple message: " + message);
}

---

发送带 Tag 的消息

使用场景：通过 Tag 对消息进行分类，消费者可以根据 Tag 过滤消息。

实现方式：在 Topic 后添加 :Tag。消息目的地的格式为：topic:tag。

示例代码：

public void sendMessageWithTag(String topic, String tag, String message) {String destination = topic + ":" + tag; // 格式：topic:tagrocketMQTemplate.convertAndSend(destination, message);System.out.println("Sent message with tag: " + message);
}

---

发送同步消息

使用场景：需要确保消息成功发送到 Broker 的场景。

实现方式：使用 RocketMQTemplate.syncSend(destination, message) 方法。

示例代码：

public void sendSyncMessage(String topic, String message) {SendResult sendResult = rocketMQTemplate.syncSend(topic, message);System.out.println("Sync message sent, result: " + sendResult);
}

---

发送异步消息

使用场景：不需要立即确认消息是否发送成功，通过回调处理发送结果。

实现方式：使用 RocketMQTemplate.asyncSend(destination, message, sendCallback) 方法。

示例代码：

public void sendAsyncMessage(String topic, String message) {rocketMQTemplate.asyncSend(topic, message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("Async message sent successfully: " + sendResult);}@Overridepublic void onException(Throwable throwable) {System.err.println("Async message send failed: " + throwable.getMessage());}});
}

---

发送单向消息

使用场景：适用于不需要确认的日志、监控等场景。

实现方式：
使用 RocketMQTemplate.sendOneWay(destination, message) 方法。

示例代码：

public void sendOneWayMessage(String topic, String message) {rocketMQTemplate.sendOneWay(topic, message);System.out.println("One-way message sent: " + message);
}

---

4.2 消息消费（Consumer）

基础消费逻辑

实现方式：

• 实现 RocketMQListener 接口：

  • 消费者类必须实现 RocketMQListener<T> 接口，并指定泛型类型（如 String 或自定义 POJO）。

• 重写 onMessage 方法：

  • 在 onMessage 方法中编写消息处理逻辑。

• 使用 @RocketMQMessageListener 注解：

  • 通过注解指定监听的 Topic 和 Consumer Group。

示例代码：

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;@Service
@RocketMQMessageListener(topic = "demo_topic",          // 监听的 TopicconsumerGroup = "demo_consumer_group"  // 消费者组名称
)
public class DemoConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received message: " + message);}
}

---

消费消息的确认机制

自动确认：
RocketMQ 默认会自动确认消息（ACK），消费者成功消费消息后，Broker 会认为消息已处理。

手动确认：
如果需要手动确认消息，可以使用 RocketMQListener<MessageExt> 并调用 ack 方法。

---

消费失败的处理

重试机制：

• 如果消费者消费消息失败，RocketMQ 会尝试重新投递消息。

• 默认重试次数为 16 次，重试间隔逐渐增加。

死信队列：

• 如果消息重试多次后仍然失败，会被投递到死信队列（DLQ，Dead Letter Queue）。

• 死信队列的命名规则为：%DLQ% + Consumer Group。

• 自动创建：

  • RocketMQ 会自动为每个 Consumer Group 创建一个死信队列，无需手动配置。

  • 例如，如果 Consumer Group 为 demo_consumer_group，则死信队列的topic为 %DLQ%demo_consumer_group。

• 手动消费死信队列：

  • 可以创建一个消费者，专门消费死信队列中的消息。

@Service
@RocketMQMessageListener(topic = "%DLQ%demo_consumer_group",  // 死信队列名称consumerGroup = "dlq_consumer_group"  // 死信队列的消费者组
)
public class DLQConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received DLQ message: " + message);// 处理死信消息，例如记录日志或人工干预}
}

---

消费带 Tag 的消息

实现方式：
在 @RocketMQMessageListener 中指定 selectorExpression 来过滤 Tag。

示例代码：

@Service
@RocketMQMessageListener(topic = "demo_topic",consumerGroup = "demo_consumer_group",selectorExpression = "tagA || tagB"  // 过滤 Tag
)
public class TagConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received tagged message: " + message);}
}

---

顺序消费

• RocketMQ 默认是并发消费，不保证消息的顺序性。

• 顺序消费是指消息按照发送顺序被消费，适用于对消息顺序有严格要求的场景。

• 实现顺序消费需要：

  1. 生产者发送消息时指定相同的消息组。

  2. 消费者设置 consumeMode = ConsumeMode.ORDERLY。

• 顺序消费的实现依赖于队列的顺序性和消费者的单线程消费。

示例代码：

@Service
@RocketMQMessageListener(topic = "demo_topic",consumerGroup = "demo_consumer_group",consumeMode = ConsumeMode.ORDERLY  // 顺序消费
)
public class OrderlyConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received orderly message: " + message);}
}

5. Spring Boot 集成 RocketMQ 的高级功能

5.1 延时消息

延时消息是指消息发送后，不会立即被消费者消费，而是在指定的延迟时间后才会被投递。RocketMQ 提供了 18 个固定的延迟级别，每个级别对应不同的延迟时间。

延迟级别与时间对应关系

延迟级别	延迟时间
1	1 秒
2	5 秒
3	10 秒
4	30 秒
5	1 分钟
6	2 分钟
7	3 分钟
8	4 分钟
9	5 分钟
10	6 分钟
11	7 分钟
12	8 分钟
13	9 分钟
14	10 分钟
15	20 分钟
16	30 分钟
17	1 小时
18	2 小时

使用场景

• 订单超时未支付取消。

• 定时任务触发。

• 延迟通知。

实现步骤

• 构建消息：使用 MessageBuilder 构建消息，并设置 MessageConst.PROPERTY_DELAY_TIME_LEVEL 属性与等级值。

• 发送消息：使用 RocketMQTemplate.syncSend 发送消息。

示例代码

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;@Service
public class DelayedMessageProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;/*** 发送延迟消息** @param topic       主题* @param message     消息内容* @param delayLevel 延迟级别（1-18）*/public void sendDelayedMessage(String topic, String message, int delayLevel) {// 构建消息，设置延迟级别Message<String> msg = MessageBuilder.withPayload(message).setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(delayLevel)).build();// 发送延迟消息rocketMQTemplate.syncSend(topic, msg);System.out.println("Sent delayed message: " + message + ", delay level: " + delayLevel);}
}

调用消息生产者发送消息

delayedMessageProducer.sendDelayedMessage("my-topic", "Delayed Message", 3); // 延迟 10 秒

---

5.2 事务消息

事务消息用于实现分布式事务，确保业务操作和消息发送的一致性。RocketMQ 的事务消息分为两个阶段：

• 发送半消息：消息发送到 Broker，但暂时对消费者不可见。

• 提交或回滚：根据本地事务的执行结果，提交或回滚消息。

使用场景

• 分布式事务场景，例如订单创建和库存扣减。

• 需要保证业务操作和消息发送一致性的场景。

实现步骤

• 定义事务监听器：实现 RocketMQLocalTransactionListener 接口，定义本地事务逻辑和回查逻辑。

• 使用注解@RocketMQTransactionListener：通过 txProducerGroup 属性指定事务组的名称。事务组的名称必须与发送事务消息时指定的 txProducerGroup 一致。txProducerGroup 是事务组的唯一标识，用于关联生产者和事务监听器。

• 发送事务消息：使用 rocketMQTemplate.sendMessageInTransaction 方法发送事务消息。

示例代码

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;@Component
@RocketMQTransactionListener(txProducerGroup = "my-transaction-group")
public class MyTransactionListener implements RocketMQLocalTransactionListener {/*** 执行本地事务** @param msg 消息* @param arg 附加参数* @return 事务状态*/@Overridepublic RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {try {// 执行本地事务System.out.println("Executing local transaction: " + msg.getPayload());// 模拟业务操作成功return RocketMQLocalTransactionState.COMMIT;} catch (Exception e) {// 本地事务执行失败，回滚消息return RocketMQLocalTransactionState.ROLLBACK;}}/*** 回查本地事务状态** @param msg 消息* @return 事务状态*/@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(Message msg) {// 回查本地事务状态System.out.println("Checking local transaction: " + msg.getPayload());return RocketMQLocalTransactionState.COMMIT;}
}

checkLocalTransaction 是 RocketMQLocalTransactionListener 接口中的一个方法，用于 回查本地事务的状态。它的作用是：

事务状态回查：

• 如果生产者发送事务消息后，未及时返回本地事务的执行结果，RocketMQ 会调用 checkLocalTransaction 方法回查事务状态。

返回事务状态：根据回查结果，返回 RocketMQLocalTransactionState 枚举值：

• COMMIT：提交事务，消息对消费者可见。

• ROLLBACK：回滚事务，消息被丢弃。

• UNKNOWN：事务状态未知，RocketMQ 会继续回查。

定义消息生产者发送事务消息 

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;@Service
public class TransactionMessageProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;/*** 发送事务消息** @param topic   主题* @param message 消息内容*/public void sendTransactionMessage(String topic, String message) {// 构建消息Message<String> msg = MessageBuilder.withPayload(message).build();// 发送事务消息rocketMQTemplate.sendMessageInTransaction("my-transaction-group", topic, msg, null);System.out.println("Sent transaction message: " + message);}
}

• withPayload(message)：设置消息的有效负载（Payload），即消息的内容。

• message 可以是任意类型的对象，RocketMQ 会自动将其序列化为字节数组。

调用消息生产者发送消息

transactionMessageProducer.sendTransactionMessage("my-topic", "Transaction Message");

---

5.3 消息过滤

RocketMQ 支持通过 Tag 或 SQL92 表达式 过滤消息，消费者可以只接收符合条件的消息。

使用 Tag 过滤

• Tag 是消息的标签，用于对消息进行分类。

• 消费者可以订阅指定 Tag 的消息。

示例代码

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;@Service
@RocketMQMessageListener(topic = "my-topic",consumerGroup = "my-consumer-group",selectorExpression = "tagA || tagB"  // 过滤 Tag
)
public class TagFilterConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received filtered message: " + message);}
}

生产者发送带 Tag 的消息

public void sendMessageWithTag(String topic, String tag, String message) {String destination = topic + ":" + tag; // 格式：topic:tagrocketMQTemplate.syncSend(destination, message);System.out.println("Sent message with tag: " + message);
}

使用 SQL92 表达式过滤

• SQL92 表达式 可以根据消息的属性进行更复杂的过滤。

• 需要在 Broker 配置中启用 SQL92 过滤功能。

示例代码

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;@Service
@RocketMQMessageListener(topic = "my-topic",consumerGroup = "my-consumer-group",selectorType = SelectorType.SQL92,  // 使用 SQL92 过滤selectorExpression = "age > 18"     // 过滤条件
)
public class SqlFilterConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println("Received SQL filtered message: " + message);}
}

生产者发送消息时设置属性

public void sendMessageWithProperties(String topic, String message) {Message<String> msg = MessageBuilder.withPayload(message).setHeader("age", 20)  // 设置消息属性.build();rocketMQTemplate.syncSend(topic, msg);System.out.println("Sent message with properties: " + message);
}