一、RocketMQ 概述
1.1 什么是 RocketMQ
RocketMQ 是阿里巴巴开源的一款分布式消息中间件，后捐赠给 Apache 基金会成为顶级项目。它具有低延迟、高并发、高可用、高可靠等特点，广泛应用于订单交易、消息推送、流计算、IoT 等场景。
1.2 核心特性
- 高吞吐量：单机支持 10 万级 TPS
- 低延迟：毫秒级消息投递
- 高可用：主从架构，支持多副本
- 消息可靠：支持消息持久化、事务消息
- 扩展性强：支持集群部署，可水平扩展
- 丰富的消息类型：顺序消息、定时消息、事务消息等
二、RocketMQ 核心架构
2.1 架构组成
RocketMQ 由四个核心组件构成：
- NameServer：轻量级注册中心，负责 Broker 的注册与发现
- Broker：消息存储与转发服务器
- Producer：消息生产者
- Consumer：消息消费者
[Producer] → [NameServer] ← [Consumer]
↓ ↑
[Broker] ←→ [Broker]
2.2 核心概念
- Topic：消息主题，一级消息类型
- Message Queue：消息队列，Topic 的分区单位
- Tag：消息标签，二级消息类型
- Group：生产者/消费者组
- Offset：消息在队列中的位置标识
三、消息生产与消费
3.1 消息发送模式
3.1.1 同步发送
public class SyncProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();
Message msg = new Message("TopicTest", "TagA",
"Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 同步发送，等待Broker返回结果
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
producer.shutdown();
}
}
3.1.2 异步发送
public class AsyncProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();
Message msg = new Message("TopicTest", "TagA",
"Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 异步发送，设置回调函数
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.printf("Send success: %s%n", sendResult);
}
@Override
public void onException(Throwable e) {
System.out.printf("Send failed: %s%n", e);
}
});
Thread.sleep(5000); // 等待回调完成
producer.shutdown();
}
}
3.1.3 单向发送
// 只发送消息，不关心结果
producer.sendOneway(msg);
3.2 消息消费模式
3.2.1 集群消费（CLUSTERING）
public class ClusterConsumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("TopicTest", "*");
// 集群模式（默认）
consumer.setMessageModel(MessageModel.CLUSTERING);
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started.");
}
}
3.2.2 广播消费（BROADCASTING）
// 广播模式（所有消费者都收到全量消息）
consumer.setMessageModel(MessageModel.BROADCASTING);
四、高级特性
4.1 顺序消息
生产者：
// 确保相同订单ID的消息发送到同一个队列
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId); // orderId作为选择队列的依据
消费者：
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(
List<MessageExt> msgs, ConsumeOrderlyContext context) {
// 保证顺序消费
return ConsumeOrderlyStatus.SUCCESS;
}
});
4.2 事务消息
public class TransactionProducer {
public static void main(String[] args) throws Exception {
TransactionMQProducer producer = new TransactionMQProducer("transaction_group");
producer.setNamesrvAddr("localhost:9876");
// 设置事务监听器
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 执行本地事务
try {
// 模拟业务处理
System.out.println("Executing local transaction: " + msg);
Thread.sleep(1000);
return LocalTransactionState.COMMIT_MESSAGE;
} catch (Exception e) {
return LocalTransactionState.ROLLBACK_MESSAGE;
}
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 检查本地事务状态
return LocalTransactionState.COMMIT_MESSAGE;
}
});
producer.start();
Message msg = new Message("TransactionTopic", null,
"Transaction Message".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 发送事务消息
TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
System.out.printf("%s%n", sendResult);
Thread.sleep(5000);
producer.shutdown();
}
}
4.3 延迟消息
// 设置延迟级别（1-18分别对应1s,5s,10s,30s,1m...2h）
msg.setDelayTimeLevel(3); // 10秒后投递
五、RocketMQ 集群部署
5.1 集群模式
- 单Master模式：开发测试用，不可靠
- 多Master模式：高性能，无单点故障
- 多Master多Slave模式（异步复制）：性能与可靠性的平衡
- 多Master多Slave模式（同步双写）：高可靠性，性能略低
5.2 部署建议
- NameServer：至少2台，保证高可用
- Broker：
- 生产环境推荐多Master多Slave
- 每个Master配置至少1个Slave
- 主从分布在不同的物理机器
六、性能优化
6.1 生产者优化
-
批量发送：
List<Message> messages = new ArrayList<>(); // 添加多条消息 producer.send(messages); -
合理设置发送超时：
producer.setSendMsgTimeout(3000); // 3秒 -
关闭VIP通道（非阿里云环境）：
producer.setVipChannelEnabled(false);
6.2 消费者优化
-
增加消费线程数：
consumer.setConsumeThreadMin(20); consumer.setConsumeThreadMax(64); -
设置批量消费：
consumer.setConsumeMessageBatchMaxSize(10); // 每次最多消费10条 -
跳过堆积消息（特殊场景）：
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
七、常见问题解决方案
7.1 消息重复消费
解决方案：
- 消费端实现幂等处理
- 使用数据库唯一键约束
- 使用Redis等缓存记录已处理消息ID
7.2 消息堆积
解决方案：
- 增加消费者实例
- 提高消费者并行度
- 优化消费逻辑，减少处理时间
- 临时扩容Topic队列数
7.3 消息丢失
预防措施：
- 生产端使用同步发送+重试机制
- Broker配置同步刷盘
flushDiskType=SYNC_FLUSH - 主从同步双写模式
八、监控与管理
8.1 控制台部署
RocketMQ 提供可视化控制台，可监控：
- 消息堆积情况
- 消费者延迟
- Broker运行状态
- Topic/Queue分布
8.2 关键监控指标
- 生产消费TPS
- 消息堆积量
- 消费延迟时间
- Broker CPU/Memory
- 磁盘IO使用率
九、最佳实践
- Topic命名规范：业务线_子系统_功能，如"Trade_Order_Notify"
- Tag使用原则：细化消息分类，如"PaySuccess", “PayFailed”
- 消息大小控制：建议不超过1MB
- Producer/Consumer分组：按业务功能划分
- 消息Key设置：便于问题追踪
msg.setKeys("ORDER_10086");
十、总结
RocketMQ 作为一款成熟的分布式消息中间件，在电商、金融、IoT等领域有着广泛应用。掌握其核心原理、部署架构和优化技巧，能够帮助开发者构建高性能、高可靠的消息系统。在实际应用中，需要根据业务场景合理选择消息模式，并做好监控与运维工作，确保消息系统的稳定运行。
