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 | 消费者消费消息的线程数,默认值为 |
rocketmq.consumer.maxReconsumeTimes | 自定义重试次数,消息消费失败会自动重试,默认最大重试次数为 16 次(包括第一次消费) |
4. Spring Boot 集成 RocketMQ 的消息生产与消费
RocketMQTemplate 是否需要定义为 Bean?
-
不需要手动定义:在 Spring Boot 项目中,
RocketMQTemplate
已经由rocketmq-spring-boot-starter
自动配置为 Bean,因此可以直接通过@Autowired
注入使用。 -
自动配置原理:
发送非 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 默认是并发消费,不保证消息的顺序性。
-
顺序消费是指消息按照发送顺序被消费,适用于对消息顺序有严格要求的场景。
-
实现顺序消费需要:
-
生产者发送消息时指定相同的消息组。
-
消费者设置
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);
}