1. RocketMQ核心概念与入门准备RocketMQ作为阿里巴巴开源的分布式消息中间件已经成为企业级应用架构中不可或缺的组件。我第一次接触RocketMQ是在2016年参与一个电商平台重构项目当时我们需要解决峰值10万QPS的订单消息处理问题。经过多轮技术选型最终RocketMQ以其出色的稳定性、可扩展性和丰富的功能特性胜出。1.1 核心组件解析RocketMQ的架构设计非常精妙主要由四个核心组件构成NameServer轻量级的注册中心负责Broker的注册与发现。与Zookeeper不同NameServer采用无状态设计各节点之间互不通信这使得它的性能极高且不存在单点故障。在实际部署时建议至少部署2-3个节点以保证高可用。Broker消息存储和转发的核心节点负责消息的接收、存储和投递。Broker采用主从架构Master-Slave支持同步/异步复制模式。生产环境中我们通常会为每个Master配置至少一个Slave并在不同的机房部署以实现容灾。Producer消息生产者负责产生消息并发送到Broker。Producer支持三种发送模式同步、异步和单向发送我们会在第3章详细讨论这些模式的选择策略。Consumer消息消费者从Broker拉取消息并进行处理。Consumer支持集群消费和广播消费两种模式前者适用于负载均衡场景后者适用于全量通知场景。1.2 开发环境搭建在开始编码前我们需要准备好开发环境。以下是基于Java生态的推荐配置!-- Maven依赖配置 -- dependency groupIdorg.apache.rocketmq/groupId artifactIdrocketmq-client/artifactId version4.9.4/version /dependency对于本地测试可以快速启动一个RocketMQ实例# 下载并解压RocketMQ wget https://archive.apache.org/dist/rocketmq/4.9.4/rocketmq-all-4.9.4-bin-release.zip unzip rocketmq-all-4.9.4-bin-release.zip # 启动NameServer nohup sh bin/mqnamesrv # 启动Broker nohup sh bin/mqbroker -n localhost:9876 注意生产环境部署需要考虑持久化、资源隔离、监控告警等要素建议参考官方部署手册进行配置。我曾遇到过因为磁盘空间不足导致Broker异常的情况因此强烈建议配置磁盘使用率监控。2. Topic注册与管理实践2.1 Topic的创建与配置在RocketMQ中Topic是消息的逻辑分类单元。与Kafka不同RocketMQ的Topic创建更加灵活支持自动创建和手动创建两种方式。手动创建Topic生产环境推荐sh bin/mqadmin updateTopic -c DefaultCluster -t OrderTopic -n 127.0.0.1:9876这个命令会在DefaultCluster集群中创建一个名为OrderTopic的主题默认包含8个读写队列。队列数量需要根据实际吞吐量需求进行调整一般建议低吞吐场景1k QPS4-8个队列中吞吐场景1k-10k QPS16-32个队列高吞吐场景10k QPS32-64个队列自动创建Topic仅限开发测试 在broker.conf中配置autoCreateTopicEnabletrue踩坑提醒自动创建Topic虽然方便但在生产环境可能导致大量无效Topic占用系统资源。我曾处理过一个故障由于误配置导致系统自动创建了上千个无用Topic最终导致NameServer内存溢出。2.2 Topic的最佳实践命名规范建议采用业务域_数据类型的命名方式如Order_Create、Payment_Notify。这样既便于管理也方便后续监控和排查问题。权限控制通过perm参数设置Topic的读写权限6读写权限默认4只读权限2只写权限标签(Tag)设计Tag是消息的二级分类合理使用Tag可以显著提升消费端的过滤效率。例如在订单Topic中可以使用TagA表示普通订单TagB表示秒杀订单。3. 消息发送模式深度解析3.1 同步发送模式同步发送是最常用的模式其特点是发送线程会阻塞等待Broker返回结果具有强一致性保证吞吐量相对较低典型代码实现public class SyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer new DefaultMQProducer(OrderProducerGroup); producer.setNamesrvAddr(localhost:9876); producer.start(); try { Message msg new Message(OrderTopic, TagA, (OrderID:1001).getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult producer.send(msg); System.out.printf(Message sent: %s%n, sendResult); } finally { producer.shutdown(); } } }性能优化技巧设置合理的发送超时时间默认3秒producer.setSendMsgTimeout(5000); // 单位毫秒启用压缩减少网络传输msg.setCompressed(true); // 默认压缩阈值4KB批量发送减少IO次数但单批次不宜超过1MB3.2 异步发送模式异步发送适用于对延迟敏感的场景发送线程不阻塞通过回调处理结果吞吐量较高需要自行处理失败重试实现示例public class AsyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer new DefaultMQProducer(AsyncProducerGroup); producer.setNamesrvAddr(localhost:9876); producer.start(); for (int i 0; i 100; i) { Message msg new Message(OrderTopic, TagA, (OrderID: i).getBytes()); producer.send(msg, new SendCallback() { Override public void onSuccess(SendResult sendResult) { System.out.println(Send success: sendResult.getMsgId()); } Override public void onException(Throwable e) { e.printStackTrace(); // 实际项目中应该记录日志并触发告警 } }); } // 等待所有回调完成 Thread.sleep(5000); producer.shutdown(); } }经验分享异步发送虽然性能好但如果回调处理不当可能导致消息丢失。我们曾经因为回调中未正确处理异常导致大量消息发送失败却未被发现。建议在回调中至少记录错误日志并考虑实现死信队列机制。3.3 单向发送模式单向发送适用于日志收集等可靠性要求不高的场景不关心发送结果吞吐量最高可能丢失消息代码示例public class OnewayProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer new DefaultMQProducer(OnewayProducerGroup); producer.setNamesrvAddr(localhost:9876); producer.start(); for (int i 0; i 1000; i) { Message msg new Message(LogTopic, TagA, (Log content i).getBytes()); producer.sendOneway(msg); // 无返回值 } producer.shutdown(); } }4. 生产环境问题排查指南4.1 常见错误代码解析错误代码含义解决方案FLUSH_DISK_TIMEOUT刷盘超时检查磁盘IO性能考虑使用SSDFLUSH_SLAVE_TIMEOUT主从同步超时检查网络状况适当增大超时时间SLAVE_NOT_AVAILABLE从节点不可用检查Slave节点状态必要时重启SYSTEM_BUSY系统繁忙降低发送频率或扩容Broker4.2 性能调优参数客户端参数producer.setCompressMsgBodyOverHowmuch(1024 * 4); // 压缩阈值 producer.setRetryTimesWhenSendFailed(3); // 同步发送重试次数 producer.setRetryTimesWhenSendAsyncFailed(2); // 异步发送重试次数Broker参数broker.confsendMessageThreadPoolNums16 # 发送线程数 flushDiskTypeASYNC_FLUSH # 异步刷盘性能更好 mapedFileSizeCommitLog1073741824 # CommitLog文件大小1GB4.3 监控与告警完善的监控体系应该包括基础指标CPU、内存、磁盘、网络业务指标堆积消息数、发送/消费TPS端到端延迟从生产到消费的延迟时间推荐使用PrometheusGrafana搭建监控平台关键指标包括rocketmq_producer_tpsrocketmq_consumer_tpsrocketmq_message_accumulation我在实际项目中曾通过监控发现了一个隐蔽的性能问题某个Consumer Group的消费速度突然下降最终排查发现是因为下游系统升级导致处理逻辑变慢。如果没有完善的监控这种问题可能需要很长时间才能发现。
