华为云分布式消息服务RocketMQ版从入门到对接实战指南
1. 华为云分布式消息服务RocketMQ版概述
分布式消息服务RocketMQ版是华为云提供的一款完全托管的消息队列服务，兼容开源Apache RocketMQ客户端。它具备低延迟、弹性高可靠、高吞吐、动态扩展等特性，为分布式应用系统提供异步通信、应用解耦、流量削峰、数据同步等能力。RocketMQ实例采用物理隔离的方式部署，租户独占实例资源，保障了消息通信的稳定与安全。
华为云RocketMQ版兼容开源RocketMQ客户端，支持顺序消息、延迟消息、定时消息、重投消息、死信消息、事务消息等丰富的消息类型。同时提供消息轨迹追踪、消息溯源、链路诊断、死信导出、监控告警等运维能力，帮助开发者全方位掌握服务运行状况。
需要先登录华为云控制台，点击：华为云控制台，还没有账号，点击：注册并关联，已有账号点击：登录后关联
2. 准备工作与实例创建
2.1 账号准备与权限配置
在使用华为云RocketMQ服务之前，需要完成以下准备工作：
注册华为云账号并完成实名认证。如果已有账号可跳过此步骤。为操作人员添加RocketMQ的管理员权限DMS FullAccess，具体操作是在IAM控制台创建用户并授权。RocketMQ实例运行于虚拟私有云VPC中，需要提前创建VPC和子网，且VPC与RocketMQ实例必须在同一区域。创建安全组并配置相应的入方向规则，允许客户端访问RocketMQ实例的端口。准备一台弹性云服务器ECS用于部署客户端程序，ECS需要与RocketMQ实例在同一个VPC内以便内网访问。
2.2 购买RocketMQ实例
登录华为云控制台，进入分布式消息服务RocketMQ版管理页面。单击右上角购买RocketMQ实例按钮，根据业务需求选择实例规格。配置参数包括实例名称、消息引擎版本、存储空间、VPC与子网、安全组、可用区、产品规格、存储IO规格、代理个数等。创建实例时可以开启SSL加密传输，也可以开启ACL访问控制以实现用户级别的权限隔离。
以下是通过API创建RocketMQ实例的请求示例：
{
"name": "rocketmq-instance",
"engine": "rocketmq",
"engine_version": "4.8.0",
"storage_space": 600,
"vpc_id": "vpc-id-xxxxx",
"subnet_id": "subnet-id-xxxxx",
"security_group_id": "sg-id-xxxxx",
"available_zones": ["az-id-xxxxx"],
"product_id": "c6.4u8g.cluster",
"storage_spec_code": "dms.physical.storage.high.v2",
"broker_num": 1
}2.3 创建Topic与消费组
实例创建成功后，需要创建Topic用于存储消息。在RocketMQ实例控制台，进入实例概览页面，在左侧导航栏单击实例管理下的Topic管理，进入Topic管理页面。单击创建Topic按钮，填写Topic名称并选择消息类型普通、顺序、事务等。
消费组用于管理消费者的订阅关系，同样需要在控制台创建。在左侧导航栏单击实例管理下的消费组管理，进入消费组列表页面，单击创建消费组并填写相关信息。
3. 客户端环境配置
3.1 ECS环境准备
在ECS上部署RocketMQ客户端需要配置Java运行环境。以Linux系统为例，使用root用户登录ECS，安装JDK 1.8.111及以上版本并配置环境变量。下载JDK安装包后解压到指定目录，编辑~/.bash_profile文件添加Java环境变量配置。
3.2 Maven依赖配置
在Java项目中通过Maven引入RocketMQ客户端依赖。推荐使用5.1.4及以上版本的客户端。在pom.xml中添加以下依赖：
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>5.1.4</version>
</dependency>如果使用Spring Boot框架，可以引入RocketMQ-Spring依赖：
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>4. Java原生客户端对接
4.1 连接信息收集
对接前需要收集以下连接信息：RocketMQ实例的内网或公网连接地址和端口、Topic名称、消费组名称、如果开启了ACL还需要用户名和密钥、如果开启了SSL还需要证书相关配置。
4.2 普通消息生产与消费
普通消息是最基本的消息类型，支持同步发送和异步发送两种方式。同步发送是指消息发送方发出一条消息到服务端，服务端接收并处理消息后返回响应，发送方收到响应后才发送下一条消息。
生产者示例代码：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
public class RocketMQProducer {
public static void main(String[] args) throws Exception {
// 创建生产者实例，指定生产者组名
DefaultMQProducer producer = new DefaultMQProducer("producer-group");
// 设置NameServer地址，即RocketMQ实例的连接地址
producer.setNamesrvAddr("192.168.0.1:8100;192.168.0.2:8100");
// 启动生产者
producer.start();
// 创建消息对象，指定Topic、Tag和消息体
Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
// 发送消息并获取发送结果
SendResult sendResult = producer.send(msg);
System.out.println("发送结果：" + sendResult);
// 关闭生产者
producer.shutdown();
}
}消费者示例代码：
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
public class RocketMQConsumer {
public static void main(String[] args) throws Exception {
// 创建消费者实例，指定消费组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
// 设置NameServer地址
consumer.setNamesrvAddr("192.168.0.1:8100;192.168.0.2:8100");
// 订阅Topic，支持使用Tag进行消息过滤
consumer.subscribe("TopicTest", "*");
// 设置消费起始位置
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
// 注册消息监听器
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
for (var msg : msgs) {
System.out.println("收到消息：" + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 启动消费者
consumer.start();
System.out.println("消费者已启动");
}
}4.3 顺序消息收发
顺序消息是RocketMQ提供的一种严格按照顺序发布和消费的消息类型。分为全局顺序消息和分区顺序消息两种。全局顺序消息将Topic的队列数量设置为1，所有消息按照严格的FIFO顺序发布和订阅。分区顺序消息中，同一个队列内的消息按照FIFO顺序发布和订阅，生产者需要指定分区选择算法。使用gRPC协议连接时，消费者是否顺序消费取决于消费组中是否开启顺序消费。
顺序消息生产者示例：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
public class OrderedProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("ordered-producer-group");
producer.setNamesrvAddr("192.168.0.1:8100");
producer.start();
for (int i = 0; i < 10; i++) {
Message msg = new Message("OrderedTopic", "TagA", ("Order-" + i).getBytes());
// 使用订单ID作为选择Key，保证同一订单的消息进入同一队列
producer.send(msg, (MessageQueueSelector) (mqs, msg1, arg) -> {
Integer id = (Integer) arg;
return mqs.get(id % mqs.size());
}, i % 3);
}
producer.shutdown();
}
}4.4 事务消息收发
事务消息支持在业务逻辑与消息发送之间提供事务保证，通过两阶段方式提供对事务消息的支持。事务消息生产者首先发送半消息，然后执行本地事务。如果本地事务执行成功则提交事务，允许消费者消费消息。如果执行失败则回滚事务。
事务消息生产者需要实现两个回调函数：ExecuteLocalTransaction在发送半消息后被调用，CheckLocalTransaction在收到回查时调用。两个回调函数均可返回三种事务状态：COMMIT提交、ROLLBACK回滚、UNKNOWN未知待查。
事务消息生产者示例：
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
public class TransactionProducer {
public static void main(String[] args) throws Exception {
TransactionMQProducer producer = new TransactionMQProducer("transaction-producer-group");
producer.setNamesrvAddr("192.168.0.1:8100");
// 设置事务监听器
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 执行本地事务逻辑
System.out.println("执行本地事务");
return LocalTransactionState.COMMIT_MESSAGE;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 检查本地事务状态
System.out.println("检查本地事务状态");
return LocalTransactionState.COMMIT_MESSAGE;
}
});
producer.start();
Message msg = new Message("TransactionTopic", "TagA", "Transaction Message".getBytes());
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
System.out.println("事务发送结果：" + result);
producer.shutdown();
}
}4.5 定时消息与延迟消息
定时消息是指生产者发送消息到RocketMQ后，消息不会立即被消费，而是延迟到设定的时间点后才会发送给消费者进行消费。通过设置消息的投递时间戳可以实现定时消息。
定时消息发送示例：
import org.apache.rocketmq.common.message.Message;
import java.time.Instant;
public class ScheduledMessageProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("scheduled-producer-group");
producer.setNamesrvAddr("192.168.0.1:8100");
producer.start();
// 设置消息10秒后投递
long deliverTimestamp = Instant.now().plusSeconds(10).toEpochMilli();
Message msg = new Message("ScheduledTopic", "TagA", "Scheduled Message".getBytes());
// 设置定时投递时间戳
msg.setDeliverTimeMs(deliverTimestamp);
producer.send(msg);
System.out.println("定时消息已发送，将在10秒后投递");
producer.shutdown();
}
}5. Spring Boot框架集成
5.1 配置文件
在Spring Boot项目中集成RocketMQ，首先在application.yml或application.properties中配置连接参数：
rocketmq:
name-server: 192.168.0.1:8100;192.168.0.2:8100
producer:
group: spring-producer-group
send-message-timeout: 3000
retry-times-when-send-failed: 25.2 生产者实现
使用RocketMQ-Spring提供的RocketMQTemplate发送消息：
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class RocketMQProducerService {
@Autowired
private RocketMQTemplate rocketMQTemplate;
public void sendMessage(String topic, String message) {
rocketMQTemplate.convertAndSend(topic, message);
}
public void sendMessageWithTag(String topic, String tag, String message) {
rocketMQTemplate.send(topic + ":" + tag, message);
}
}5.3 消费者实现
使用@RocketMQMessageListener注解定义消费者：
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
@Component
@RocketMQMessageListener(
topic = "TopicTest",
consumerGroup = "spring-consumer-group",
selectorExpression = "*"
)
public class RocketMQConsumerService implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.println("接收到消息：" + message);
// 处理业务逻辑
}
}6. 安全配置与高级特性
6.1 SSL加密传输
华为云RocketMQ支持SSL加密传输，保障消息在传输过程中的安全性。在创建实例时可以选择开启SSL，也可以在实例创建后修改SSL配置。开启SSL后，客户端连接需要配置SSL相关参数。
6.2 ACL权限控制
ACL访问控制可以实现用户级别的权限隔离。开启ACL后需要创建用户并分配权限。用户名和密钥建议在配置文件或环境变量中密文存放，避免直接编辑到代码中带来安全风险。
6.3 消息轨迹与链路诊断
华为云RocketMQ提供消息轨迹追踪功能，可以追踪消息从生产到消费的完整链路。通过控制台可以查看消息的发送时间、存储时间、消费时间等关键信息，便于问题排查。
6.4 监控告警配置
通过云监控服务可以配置RocketMQ的监控告警规则。在RocketMQ实例控制台，单击实例名称进入详情页，选择监控与告警下的监控页面。在监控指标页面中找到需要创建告警的指标项，创建告警规则。建议配置消息堆积数、生产速率、消费速率等关键指标的告警策略。
7. 最佳实践
7.1 消息幂等处理
为了防止消息重复消费导致业务处理异常，消费者在接收到消息后有必要根据业务上的唯一Key对消息做幂等处理。常见的幂等处理方式包括：使用数据库唯一键约束防止重复插入、使用Redis记录已处理的消息ID、使用分布式锁保证同一消息只处理一次。
7.2 Topic与Tag的合理使用
在RocketMQ中，Topic与Tag都是业务上用来归类的标识，Topic是一级分类，Tag是二级分类。合理搭配使用Topic和Tag可以实现精准的消息过滤。对于不同业务模块使用不同的Topic，同一业务模块内使用不同的Tag区分消息类型。
7.3 订阅关系一致性
订阅关系不一致可能导致消息消费逻辑混乱，消息被重复消费或遗漏。同一个消费组内的所有消费者必须订阅相同的Topic和Tag。如果不同消费者需要不同的订阅关系，应该使用不同的消费组。
7.4 避免ClientId冲突
客户端的ClientId默认为进程号、本机IP、instanceName的组合。如果同一个进程内同一个消费组下启动两个消费者，会导致ClientId相同，从而出现部分队列重复消费、部分队列无法消费的问题。建议手动设置instanceName来避免冲突：consumer.setInstanceName(String.valueOf(System.nanoTime()))
7.5 开启生产者异常检测
Java客户端5.1.4及以上版本支持生产者异常检测机制。建议开启该功能，当客户端连接Broker超时后能快速剔除超时的Broker，使消息生产到其他正常的Broker上：producer.setSendLatencyFaultEnable(true)
7.6 消息堆积处理建议
当出现消息堆积时，可以从以下几个方面进行排查和处理：检查消费者的消费能力是否充足，适当增加消费者实例或提升消费处理速度。检查是否有消息消费失败导致反复重试，排查消费逻辑中的异常。检查Topic的分区数量是否合理，适当增加分区数提升并行消费能力。
8. 常见问题解答
问题1：华为云RocketMQ与开源RocketMQ兼容吗？
华为云分布式消息服务RocketMQ版完全兼容开源Apache RocketMQ客户端。开发者可以直接使用开源RocketMQ的SDK进行连接和消息收发，无需修改代码逻辑。
问题2：内网连接和公网连接有什么区别？
内网连接要求客户端与RocketMQ实例在同一个VPC内，网络延迟低、安全性高且不产生公网流量费用。公网连接需要开启实例的公网访问功能并绑定弹性IP地址，适用于外部系统接入场景，但会产生公网流量费用。
问题3：如何保证消息不丢失？
可以通过以下方式保障消息可靠性：使用同步发送模式确保消息成功写入Broker；开启实例的持久化存储；配置消息发送重试机制；消费端处理消息时使用手动确认机制；开启消息轨迹追踪便于问题定位。
问题4：消息堆积了怎么办？
消息堆积时首先检查消费者的消费能力是否充足；查看消费者日志确认是否有消费异常；适当增加消费者实例数量；检查Topic分区数是否合理；通过云监控查看实例的消费速率和积压量指标。
问题5：如何配置告警规则？
在RocketMQ实例控制台进入监控页面，找到需要监控的指标项如消息堆积数、生产速率等，创建告警规则。设置告警阈值和通知方式，当指标超过阈值时系统自动发送告警通知。
问题6：SSL和ACL可以同时开启吗？
可以。SSL保障传输层加密安全，ACL保障用户级别的访问控制权限。两者可以同时开启，为RocketMQ实例提供双重安全保障。
