〓 RocketMQ 基础概念
什么是消息队列？
消息队列（Message Queue，简称MQ）是一种应用对应用的通信方法，应用程序通过写和检索出入列队的针对应用程序的数据（即消息）来通信，而无需经过连接到的服务。消息队列主要解决应用耦合、异步消息、流量削锋等问题，实现高性能、高可用、可伸缩和最终一致性架构。
RocketMQ 简介
RocketMQ 是阿里巴巴开源的一款分布式消息中间件，后来捐献给Apache基金会，成为Apache的顶级项目。它具备高吞吐量、低延迟、高可靠性、无限扩展性等特点，广泛应用于金融、电商、物流等多个领域。
⚡ 高吞吐量
RocketMQ 支持每秒处理百万级消息，满足高并发场景需求。
⚡ 低延迟
消息端到端延迟低至毫秒级，确保实时性要求高的业务场景。
⚡ 高可靠性
支持主从架构、多副本机制，确保消息不丢失，数据高可用。
⚡ 无限扩展性
支持水平扩展，轻松应对业务增长带来的压力。
〓 RocketMQ 架构解析
核心组件
- Producer（生产者）：负责发送消息到Broker，支持集群发送、广播发送等模式。
- Consumer（消费者）：负责从Broker接收并处理消息，支持集群消费和广播消费。
- Broker：消息存储和转发中心，负责接收、存储和转发消息。
- NameServer：服务发现组件，负责管理Broker的路由信息，生产者与消费者通过NameServer获取路由信息。
架构工作流程
Broker启动后向NameServer注册，定期发送心跳包维持连接。
Producer和Consumer从NameServer获取Topic的路由信息，包括Broker地址、队列分布等。
Producer根据路由信息选择队列，将消息发送到对应的Broker。
Broker将消息持久化到磁盘，并返回发送结果给Producer。
Consumer拉取或被动接收消息，处理业务逻辑后提交消费结果。
〓 RocketMQ 核心特性
消息可靠性保障机制
RocketMQ 通过多种机制保障消息不丢失：
- 同步刷盘：消息写入Broker后，必须同步刷盘成功才返回成功，确保数据持久化。
- 异步刷盘：消息写入内存后异步刷盘，性能更高但有一定丢失风险。
- 主从同步：Master节点接收消息后，同步复制到Slave节点，确保数据冗余。
- ACK机制：Consumer消费成功后需向Broker发送ACK，确保消息被正确处理。
// 同步刷盘配置
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setSendMsgTimeout(3000);
producer.setVipChannelEnabled(false);
producer.start();事务消息应用场景
事务消息用于保证本地事务与消息发送的最终一致性，典型应用场景包括：
- 订单支付：支付成功后发送消息通知库存扣减。
- 数据同步：数据库更新后同步消息到搜索引擎。
- 日志记录：业务操作后记录审计日志。
// 事务消息示例
TransactionMQProducer producer = new TransactionMQProducer("transaction_group");
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 执行本地事务
return LocalTransactionState.COMMIT_MESSAGE;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 回查本地事务状态
return LocalTransactionState.COMMIT_MESSAGE;
}
});顺序消息实现原理
RocketMQ 通过消息队列和消费者单线程消费实现顺序消息：
- 全局顺序：所有消息只有一个队列，串行处理，吞吐量低。
- 局部顺序：相同业务标识的消息发送到同一队列，保证局部顺序。
- 消费者单线程：每个队列由一个消费者线程处理，确保顺序。
// 局部顺序消息示例
MessageQueueSelector selector = (mqs, msg, arg) -> {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
};
producer.send(msg, selector, orderId);延迟消息使用场景
延迟消息允许消息在指定时间后被消费者处理，典型应用包括：
- 订单超时取消：下单后30分钟未支付自动取消。
- 重试机制：失败任务延迟重试。
- 定时任务：替代定时任务调度系统。
// 延迟消息示例
Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
msg.setDelayTimeLevel(3); // 延迟等级3，对应10秒
producer.send(msg);〓 RocketMQ 面试真题
Q1：RocketMQ 如何保证消息不丢失？
RocketMQ 从生产者、Broker、消费者三个层面保证消息不丢失：
| 环节 | 保障措施 | 说明 |
|---|---|---|
| 生产者 | 同步发送 + ACK确认 | 发送消息后等待Broker确认，失败则重试 |
| Broker | 同步刷盘 + 主从同步 | 消息持久化到磁盘，主从同步确保数据冗余 |
| 消费者 | 手动ACK + 消费重试 | 消费成功后才ACK，失败则重试或进入死信队列 |
Q2：RocketMQ 与 Kafka 的区别？
⚡ 消息模型
RocketMQ：支持点对点（P2P）和发布订阅（Pub/Sub）两种模型。
Kafka：仅支持发布订阅模型。
⚡ 消息可靠性
RocketMQ：支持事务消息、顺序消息，可靠性更高。
Kafka：主要面向日志收集，事务支持较弱。
⚡ 吞吐量
RocketMQ：百万级TPS，适合金融级场景。
Kafka：千万级TPS，适合大数据场景。
⚡ 生态集成
RocketMQ：与阿里生态深度集成，适合Java应用。
Kafka：大数据生态核心组件，支持多语言。
Q3：RocketMQ 消息堆积如何处理？
- 增加消费者：横向扩展消费者实例，提高消费能力。
- 优化消费逻辑：减少单次消费处理时间，提升吞吐量。
- 临时扩容：紧急情况下临时增加Broker和Consumer实例。
- 丢弃非重要消息：对于非核心业务，可适当丢弃部分消息。
// 批量消费示例
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
// 批量处理消息
processMessage(msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}); Q4：RocketMQ 如何实现分布式事务？
RocketMQ 通过半消息（Half Message）和回查机制实现分布式事务：
- 发送半消息：Producer发送半消息到Broker，Broker不立即投递。
- 执行本地事务：Producer执行本地数据库操作。
- 提交或回滚：根据本地事务结果，向Broker提交或回滚半消息。
- 回查机制：Broker定期回查本地事务状态，确保最终一致性。
〓 RocketMQ 常见问题排查
问题1：消息发送失败
可能原因：
- 网络不通：检查Producer与Broker的网络连接。
- Topic不存在：确认Topic已创建且权限正确。
- 队列满：检查Broker磁盘空间是否充足。
解决方案：
// 设置发送超时和重试次数
producer.setSendMsgTimeout(5000);
producer.setRetryTimesWhenSendFailed(3);问题2：消息消费延迟
可能原因：
- 消费逻辑复杂：优化业务代码，减少耗时操作。
- 数据库瓶颈：检查数据库连接池和SQL性能。
- 资源不足：增加CPU、内存、网络带宽。
解决方案：
// 调整消费线程池
consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(50);问题3：消息重复消费
可能原因：
- 网络超时：Producer发送成功但ACK丢失。
- 消费失败重试：Consumer处理失败触发重试。
解决方案：
- 幂等性设计：消费者实现幂等逻辑，如使用唯一ID去重。
- 数据库唯一约束：利用数据库唯一索引防止重复插入。
- 缓存去重：使用Redis记录已处理消息ID。