RocketMQ 入门第一篇从概述到实战环境搭建 Java APISpring Boot 集成在分布式系统中消息中间件是 “解耦、削峰、异步” 的关键组件而 RocketMQ 作为 Apache 顶级项目凭借高可用、高吞吐、支持多种消息类型的特性成为很多企业的首选。这篇作为 RocketMQ 系列的第一篇会从 “是什么→怎么搭环境→怎么写代码” 三个维度一起学习快速入门 RocketMQ零基础也能跟着做。一、RocketMQ 概述它是什么为什么选它在动手之前先搞清楚 RocketMQ 的 “出身” 和核心能力避免盲目上手。1. RocketMQ 的 “前世今生”RocketMQ 并非凭空出现它的发展历程体现了其成熟度2012 年由阿里巴巴开源最初用于阿里内部的电商场景如双 11 的消息流转2016 年捐赠给 Apache 软件基金会进入孵化阶段2017 年成为 Apache 顶级项目正式走向开源社区目前已更新到 5.x 版本支持云原生、gRPC 协议。核心定位纯 Java 编写的分布式消息中间件基于队列模型专注于解决分布式系统中的消息通信问题。2. RocketMQ 的核心特点为什么选它相比 RabbitMQ、KafkaRocketMQ 有自己独特的优势尤其适合企业级场景类别核心特性消息顺序单队列内严格保证 FIFO先进先出支持全局顺序消息特定场景消费模式支持 Push推和 Pull拉两种消费模式Push 基于 Pull 封装兼顾实时性和灵活性堆积能力单一队列可堆积百万级消息不会因堆积导致性能下降消息类型支持普通消息、事务消息、顺序消息、批量消息、定时延迟消息、消息回溯高可用分布式部署架构支持多主多从自动故障转移无单点风险其他优势支持 18 级延迟消息、失败消息重发可配置次数和间隔、Tag 过滤减少网络传输举个实际场景电商下单后需要同步扣库存、异步发物流通知、异步记录日志 —— 用 RocketMQ 的普通消息就能实现异步解耦避免下单接口因后续流程耗时过长而超时。3. RocketMQ 的 CAP 定位牺牲一致性换可用性消息中间件在 CAP 定理中通常会做取舍RocketMQ 的选择是AP 优先兼顾最终一致性。CAP 定义回顾一致性Consistency所有节点同一时间看到的数据一致可用性Availability服务始终可用即使部分节点故障分区容错性Partition tolerance网络分区时服务仍能继续运行。RocketMQ 的选择优先保证可用性A和分区容错性P允许短暂的数据不一致如主从同步延迟通过 “同步双写”“异步复制” 等配置可在一致性和可用性之间做权衡如金融场景选同步双写提升一致性。4. 必备学习资源后续深入靠这些学习 RocketMQ官方资源最权威整理如下源码地址https://github.com/apache/rocketmq可查看最新特性和 bug 修复中文官方文档https://rocketmq.apache.org/zh/从入门到进阶的完整指南官方 FAQhttp://rocketmq.apache.org/docs/faq/解决常见问题如启动失败、消息丢失商业版阿里云RocketMQhttps://www.aliyun.com/product/rocketmq开箱即用适合企业生产环境。二、RocketMQ 环境搭建从单机到集群环境搭建是入门的第一步先从简单的单机部署入手再了解集群方案最后实战双主双从集群生产环境常用。1. 前置准备环境要求RocketMQ 对环境的要求不高确保满足以下条件JDK1.8 及以上推荐 1.8避免高版本兼容问题操作系统Linux推荐生产环境、Windows开发环境注意路径不含中文内存至少 2GB默认启动脚本占用内存较大可修改Maven3.5 及以上编译源码用若直接下载二进制包可不用。2. 单机部署3 步启动以 Linux 为例单机部署适合开发和测试步骤简单5 分钟就能搞定。步骤 1下载并解压二进制包推荐直接下载二进制包无需编译地址https://rocketmq.apache.org/zh/download选择稳定版本如 4.9.5兼容性好。# 1. 下载也可手动下载后上传到Linuxwgethttps://archive.apache.org/dist/rocketmq/4.9.5/rocketmq-all-4.9.5-bin-release.zip# 2. 解压到/usr/local目录unziprocketmq-all-4.9.5-bin-release.zip-d/usr/local/# 3. 重命名方便后续操作mv/usr/local/rocketmq-all-4.9.5-bin-release /usr/local/rocketmq步骤 2修改 JVM 参数关键避免内存不足RocketMQ 默认的 JVM 参数内存占用较高单机环境可能启动失败需修改runserver.sh和runbroker.sh# 1. 修改NameServer的JVM参数runserver.shvim/usr/local/rocketmq/bin/runserver.sh# 找到以下行修改为小内存如Xms512m Xmx512m Xmn256m# 原配置JAVA_OPT${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize128m -XX:MaxMetaspaceSize320mJAVA_OPT${JAVA_OPT}-server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize128m -XX:MaxMetaspaceSize320m# 2. 修改Broker的JVM参数runbroker.shvim/usr/local/rocketmq/bin/runbroker.sh# 原配置JAVA_OPT${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4gJAVA_OPT${JAVA_OPT}-server -Xms512m -Xmx512m -Xmn256m步骤 3启动 NameServer 和 BrokerRocketMQ 的核心组件是NameServer路由注册中心和Broker消息存储和转发必须先启动 NameServer再启动 Broker。# 1. 启动NameServer后台启动日志输出到nohup.outcd/usr/local/rocketmqnohupshbin/mqnamesrv# 2. 查看NameServer启动日志确认是否成功tail-f~/logs/rocketmqlogs/namesrv.log# 出现以下日志表示启动成功The Name Server boot success...# 3. 启动Broker指定NameServer地址为本地后台启动# -n指定NameServer地址格式ip:port默认端口9876# autoCreateTopicEnabletrue允许自动创建Topic开发环境用生产环境建议关闭nohupshbin/mqbroker-nlocalhost:9876autoCreateTopicEnabletrue# 4. 查看Broker启动日志确认是否成功tail-f~/logs/rocketmqlogs/broker.log# 出现以下日志表示启动成功The broker[xxx, ip:port] boot success...常见问题启动失败怎么办问题 1启动 Broker 时报 “Cannot allocate memory”原因JVM 参数设置过大按步骤 2 修改runbroker.sh的内存参数即可。问题 2客户端连接不上 NameServer原因防火墙未开放 9876NameServer 端口和 10911Broker 默认端口执行以下命令开放端口# 开放9876和10911端口firewall-cmd--zonepublic --add-port9876/tcp--permanentfirewall-cmd--zonepublic --add-port10911/tcp--permanent# 重新加载防火墙规则firewall-cmd--reload3. 集群方案分析生产环境怎么部署单机部署仅适合开发生产环境必须用集群RocketMQ 常见的集群架构有两种1双主双从架构推荐架构图2 个 Master 节点 2 个 Slave 节点每个 Master 对应一个 Slave数据同步Master 和 Slave 之间支持 “同步双写”Master 写成功后Slave 也写成功才返回或 “异步复制”Master 写成功后直接返回Slave 异步拉取优势高可用即使一个 Master 故障Slave 可切换为 Master避免消息丢失。2多主多从架构大规模场景架构图N 个 Master 节点 N 个 Slave 节点N≥3优势更高的吞吐量可按 Topic 分区到不同 Master分担压力适用场景日均消息量千万级以上的场景如电商双 11。4. 双主双从集群搭建实战这里以 “双主双从 异步复制” 为例基于两台 Linux 服务器IP 分别为 192.168.1.101 和 192.168.1.102。步骤 1修改 Broker 配置文件在两台服务器的/usr/local/rocketmq/conf/2m-2s-async目录下已有默认配置文件需修改 IP 为实际服务器 IP# 服务器1192.168.1.101修改broker-a.propertiesMaster和broker-a-s.propertiesSlavevim/usr/local/rocketmq/conf/2m-2s-async/broker-a.properties# 修改以下参数brokerIP1192.168.1.101# 当前服务器IPnamesrvAddr192.168.1.101:9876;192.168.1.102:9876# 两个NameServer地址vim/usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties# 修改以下参数brokerIP1192.168.1.101namesrvAddr192.168.1.101:9876;192.168.1.102:9876# 服务器2192.168.1.102修改broker-b.propertiesMaster和broker-b-s.propertiesSlavevim/usr/local/rocketmq/conf/2m-2s-async/broker-b.propertiesbrokerIP1192.168.1.102namesrvAddr192.168.1.101:9876;192.168.1.102:9876vim/usr/local/rocketmq/conf/2m-2s-async/broker-b-s.propertiesbrokerIP1192.168.1.102namesrvAddr192.168.1.101:9876;192.168.1.102:9876步骤 2启动集群节点按以下顺序启动先启动所有 NameServer再启动 Master最后启动 Slave# 1. 启动两台服务器的NameServer# 服务器1nohupsh/usr/local/rocketmq/bin/mqnamesrv# 服务器2nohupsh/usr/local/rocketmq/bin/mqnamesrv# 2. 启动Master节点# 服务器1启动broker-aMasternohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-a.properties# 服务器2启动broker-bMasternohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-b.properties# 3. 启动Slave节点# 服务器1启动broker-a-sSlavenohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties# 服务器2启动broker-b-sSlavenohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties步骤 3验证集群状态通过mqadmin工具查看集群信息# 查看集群节点状态sh/usr/local/rocketmq/bin/mqadmin clusterList-n192.168.1.101:9876# 输出结果包含4个节点2个Master2个Slave表示集群搭建成功5. RocketMQ 默认配置表常用参数部署时经常需要调整配置整理常用默认参数如下配置项默认值说明NameServer端口 9876客户端连接 NameServer 的端口Broker监听端口 10911客户端连接 Broker 的端口Broker 角色ASYNC_MASTER默认为异步 Master可改为 SYNC_MASTER同步消息存储路径~/store消息、索引、commitLog 的存储目录消息过期时间72 小时过期消息自动清理最大消息大小4MB超过会报错可修改 maxMessageSize 参数三、RocketMQ Java 开发API 与 Spring Boot 集成环境搭好后就可以写代码了先讲原生 Java API再讲 Spring Boot 集成更符合实际开发。1. 原生 Java API 开发基础先创建一个 Maven 项目引入 RocketMQ 依赖dependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.9.5/version!-- 与部署的RocketMQ版本一致 --/dependency1生产者发送普通消息生产者负责发送消息到 Broker核心步骤创建生产者→设置 NameServer→启动生产者→发送消息→关闭生产者。importorg.apache.rocketmq.client.exception.MQClientException;importorg.apache.rocketmq.client.producer.DefaultMQProducer;importorg.apache.rocketmq.client.producer.SendResult;importorg.apache.rocketmq.common.message.Message;importorg.apache.rocketmq.remoting.common.RemotingHelper;publicclassSimpleProducer{publicstaticvoidmain(String[]args)throwsException{// 1. 创建生产者指定生产者组名同一组的生产者共享负载均衡DefaultMQProducerproducernewDefaultMQProducer(simple-producer-group);// 2. 设置NameServer地址单机localhost:9876集群192.168.1.101:9876;192.168.1.102:9876producer.setNamesrvAddr(localhost:9876);// 3. 启动生产者必须在发送消息前启动producer.start();// 4. 构建消息参数Topic名、Tag过滤用、消息内容Stringtopicsimple-topic;Stringtagtag1;StringcontentHello RocketMQ! This is a simple message.;MessagemessagenewMessage(topic,tag,content.getBytes(RemotingHelper.DEFAULT_CHARSET)// 消息内容转字节数组);// 5. 发送消息同步发送等待Broker返回结果SendResultsendResultproducer.send(message);System.out.println(消息发送成功结果sendResult);// 6. 关闭生产者建议在应用关闭时调用避免资源泄漏producer.shutdown();}}2消费者订阅并消费消息消费者负责从 Broker 订阅并消费消息核心步骤创建消费者→设置 NameServer→订阅 Topic→注册消息监听→启动消费者。importorg.apache.rocketmq.client.consumer.DefaultMQPushConsumer;importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;importorg.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;importorg.apache.rocketmq.common.message.MessageExt;importjava.util.List;publicclassSimpleConsumer{publicstaticvoidmain(String[]args)throwsMQClientException{// 1. 创建Push消费者指定消费者组名同一组的消费者共享消息DefaultMQPushConsumerconsumernewDefaultMQPushConsumer(simple-consumer-group);// 2. 设置NameServer地址consumer.setNamesrvAddr(localhost:9876);// 3. 订阅Topic*表示订阅所有Tag也可指定具体Tag如tag1||tag2consumer.subscribe(simple-topic,*);// 4. 注册消息监听器并发消费默认一次消费一条consumer.registerMessageListener(newMessageListenerConcurrently(){OverridepublicConsumeConcurrentlyStatusconsumeMessage(ListMessageExtmsgs,ConsumeConcurrentlyContextcontext){// 遍历处理消息for(MessageExtmsg:msgs){try{// 解析消息内容StringcontentnewString(msg.getBody(),UTF-8);System.out.println(消费者收到消息Topicmsg.getTopic(), Tagmsg.getTags(), Contentcontent);}catch(Exceptione){e.printStackTrace();// 消费失败返回RECONSUME_LATERBroker会重试默认重试16次returnConsumeConcurrentlyStatus.RECONSUME_LATER;}}// 消费成功返回CONSUME_SUCCESSreturnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 5. 启动消费者启动后会持续监听消息consumer.start();System.out.println(消费者启动成功等待接收消息...);}}运行顺序先启动消费者确保消费者已订阅 Topic避免消息发送后消费者未启动导致消息堆积再启动生产者发送消息查看消费者控制台会输出收到的消息内容。2. Spring Boot 集成 RocketMQ实战实际开发中更常用 Spring Boot 集成通过spring-boot-starter-rocketmq简化开发。步骤 1引入依赖dependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-spring-boot-starter/artifactIdversion2.2.3/version!-- 与RocketMQ版本匹配4.9.5对应2.2.3 --/dependency步骤 2配置 application.ymlrocketmq:name-server:localhost:9876# NameServer地址集群用分号分隔producer:group:springboot-producer-group# 生产者组名send-message-timeout:3000# 发送消息超时时间毫秒步骤 3生产者使用 RocketMQTemplateSpring Boot 提供RocketMQTemplate封装了原生 API简化发送逻辑importorg.apache.rocketmq.spring.core.RocketMQTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestParam;importorg.springframework.web.bind.annotation.RestController;RestControllerpublicclassRocketMQProducerController{// 注入RocketMQTemplateAutowiredprivateRocketMQTemplaterocketMQTemplate;// 发送普通消息通过HTTP接口触发GetMapping(/sendMsg)publicStringsendMsg(RequestParamStringcontent){Stringtopicspringboot-topic;Stringtagtag1;// 发送消息格式Topic:TagrocketMQTemplate.convertAndSend(topic:tag,content);return消息发送成功内容content;}}步骤 4消费者使用 RocketMQMessageListener通过注解RocketMQMessageListener指定消费者组和订阅的 Topic/Tag无需手动注册监听器importorg.apache.rocketmq.spring.annotation.RocketMQMessageListener;importorg.apache.rocketmq.spring.core.RocketMQListener;importorg.springframework.stereotype.Component;// 标注为消费者指定消费者组、Topic、TagComponentRocketMQMessageListener(consumerGroupspringboot-consumer-group,topicspringboot-topic,selectorExpressiontag1// 只消费tag1的消息*表示所有)publicclassRocketMQConsumerimplementsRocketMQListenerString{// 消息消费逻辑参数为消息内容自动反序列化为StringOverridepublicvoidonMessage(Stringmessage){System.out.println(Spring Boot消费者收到消息message);}}测试启动 Spring Boot 应用访问http://localhost:8080/sendMsg?contentHello RocketMQ Spring Boot查看控制台消费者会输出收到的消息内容。四、总结与预告本次学习完成了 RocketMQ 的入门了解了 RocketMQ 的发展历程、核心特点和 CAP 定位知道了它适合哪些场景实战了单机部署和双主双从集群搭建解决了启动时的常见问题用原生 Java API 和 Spring Boot 集成两种方式实现了消息的发送和消费。下一次将学习RocketMQ系统架构与核心组件。 
长河君 · 2026/10/19 2:46:21 · 阅读 1.2k RocketMQ 入门第一篇：从概述到实战（环境搭建 + Java API+Spring Boot 集成）
RocketMQ 入门第一篇从概述到实战环境搭建 Java APISpring Boot 集成在分布式系统中消息中间件是 “解耦、削峰、异步” 的关键组件而 RocketMQ 作为 Apache 顶级项目凭借高可用、高吞吐、支持多种消息类型的特性成为很多企业的首选。这篇作为 RocketMQ 系列的第一篇会从 “是什么→怎么搭环境→怎么写代码” 三个维度一起学习快速入门 RocketMQ零基础也能跟着做。一、RocketMQ 概述它是什么为什么选它在动手之前先搞清楚 RocketMQ 的 “出身” 和核心能力避免盲目上手。1. RocketMQ 的 “前世今生”RocketMQ 并非凭空出现它的发展历程体现了其成熟度2012 年由阿里巴巴开源最初用于阿里内部的电商场景如双 11 的消息流转2016 年捐赠给 Apache 软件基金会进入孵化阶段2017 年成为 Apache 顶级项目正式走向开源社区目前已更新到 5.x 版本支持云原生、gRPC 协议。核心定位纯 Java 编写的分布式消息中间件基于队列模型专注于解决分布式系统中的消息通信问题。2. RocketMQ 的核心特点为什么选它相比 RabbitMQ、KafkaRocketMQ 有自己独特的优势尤其适合企业级场景类别核心特性消息顺序单队列内严格保证 FIFO先进先出支持全局顺序消息特定场景消费模式支持 Push推和 Pull拉两种消费模式Push 基于 Pull 封装兼顾实时性和灵活性堆积能力单一队列可堆积百万级消息不会因堆积导致性能下降消息类型支持普通消息、事务消息、顺序消息、批量消息、定时延迟消息、消息回溯高可用分布式部署架构支持多主多从自动故障转移无单点风险其他优势支持 18 级延迟消息、失败消息重发可配置次数和间隔、Tag 过滤减少网络传输举个实际场景电商下单后需要同步扣库存、异步发物流通知、异步记录日志 —— 用 RocketMQ 的普通消息就能实现异步解耦避免下单接口因后续流程耗时过长而超时。3. RocketMQ 的 CAP 定位牺牲一致性换可用性消息中间件在 CAP 定理中通常会做取舍RocketMQ 的选择是AP 优先兼顾最终一致性。CAP 定义回顾一致性Consistency所有节点同一时间看到的数据一致可用性Availability服务始终可用即使部分节点故障分区容错性Partition tolerance网络分区时服务仍能继续运行。RocketMQ 的选择优先保证可用性A和分区容错性P允许短暂的数据不一致如主从同步延迟通过 “同步双写”“异步复制” 等配置可在一致性和可用性之间做权衡如金融场景选同步双写提升一致性。4. 必备学习资源后续深入靠这些学习 RocketMQ官方资源最权威整理如下源码地址https://github.com/apache/rocketmq可查看最新特性和 bug 修复中文官方文档https://rocketmq.apache.org/zh/从入门到进阶的完整指南官方 FAQhttp://rocketmq.apache.org/docs/faq/解决常见问题如启动失败、消息丢失商业版阿里云RocketMQhttps://www.aliyun.com/product/rocketmq开箱即用适合企业生产环境。二、RocketMQ 环境搭建从单机到集群环境搭建是入门的第一步先从简单的单机部署入手再了解集群方案最后实战双主双从集群生产环境常用。1. 前置准备环境要求RocketMQ 对环境的要求不高确保满足以下条件JDK1.8 及以上推荐 1.8避免高版本兼容问题操作系统Linux推荐生产环境、Windows开发环境注意路径不含中文内存至少 2GB默认启动脚本占用内存较大可修改Maven3.5 及以上编译源码用若直接下载二进制包可不用。2. 单机部署3 步启动以 Linux 为例单机部署适合开发和测试步骤简单5 分钟就能搞定。步骤 1下载并解压二进制包推荐直接下载二进制包无需编译地址https://rocketmq.apache.org/zh/download选择稳定版本如 4.9.5兼容性好。# 1. 下载也可手动下载后上传到Linuxwgethttps://archive.apache.org/dist/rocketmq/4.9.5/rocketmq-all-4.9.5-bin-release.zip# 2. 解压到/usr/local目录unziprocketmq-all-4.9.5-bin-release.zip-d/usr/local/# 3. 重命名方便后续操作mv/usr/local/rocketmq-all-4.9.5-bin-release /usr/local/rocketmq步骤 2修改 JVM 参数关键避免内存不足RocketMQ 默认的 JVM 参数内存占用较高单机环境可能启动失败需修改runserver.sh和runbroker.sh# 1. 修改NameServer的JVM参数runserver.shvim/usr/local/rocketmq/bin/runserver.sh# 找到以下行修改为小内存如Xms512m Xmx512m Xmn256m# 原配置JAVA_OPT${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize128m -XX:MaxMetaspaceSize320mJAVA_OPT${JAVA_OPT}-server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize128m -XX:MaxMetaspaceSize320m# 2. 修改Broker的JVM参数runbroker.shvim/usr/local/rocketmq/bin/runbroker.sh# 原配置JAVA_OPT${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4gJAVA_OPT${JAVA_OPT}-server -Xms512m -Xmx512m -Xmn256m步骤 3启动 NameServer 和 BrokerRocketMQ 的核心组件是NameServer路由注册中心和Broker消息存储和转发必须先启动 NameServer再启动 Broker。# 1. 启动NameServer后台启动日志输出到nohup.outcd/usr/local/rocketmqnohupshbin/mqnamesrv# 2. 查看NameServer启动日志确认是否成功tail-f~/logs/rocketmqlogs/namesrv.log# 出现以下日志表示启动成功The Name Server boot success...# 3. 启动Broker指定NameServer地址为本地后台启动# -n指定NameServer地址格式ip:port默认端口9876# autoCreateTopicEnabletrue允许自动创建Topic开发环境用生产环境建议关闭nohupshbin/mqbroker-nlocalhost:9876autoCreateTopicEnabletrue# 4. 查看Broker启动日志确认是否成功tail-f~/logs/rocketmqlogs/broker.log# 出现以下日志表示启动成功The broker[xxx, ip:port] boot success...常见问题启动失败怎么办问题 1启动 Broker 时报 “Cannot allocate memory”原因JVM 参数设置过大按步骤 2 修改runbroker.sh的内存参数即可。问题 2客户端连接不上 NameServer原因防火墙未开放 9876NameServer 端口和 10911Broker 默认端口执行以下命令开放端口# 开放9876和10911端口firewall-cmd--zonepublic --add-port9876/tcp--permanentfirewall-cmd--zonepublic --add-port10911/tcp--permanent# 重新加载防火墙规则firewall-cmd--reload3. 集群方案分析生产环境怎么部署单机部署仅适合开发生产环境必须用集群RocketMQ 常见的集群架构有两种1双主双从架构推荐架构图2 个 Master 节点 2 个 Slave 节点每个 Master 对应一个 Slave数据同步Master 和 Slave 之间支持 “同步双写”Master 写成功后Slave 也写成功才返回或 “异步复制”Master 写成功后直接返回Slave 异步拉取优势高可用即使一个 Master 故障Slave 可切换为 Master避免消息丢失。2多主多从架构大规模场景架构图N 个 Master 节点 N 个 Slave 节点N≥3优势更高的吞吐量可按 Topic 分区到不同 Master分担压力适用场景日均消息量千万级以上的场景如电商双 11。4. 双主双从集群搭建实战这里以 “双主双从 异步复制” 为例基于两台 Linux 服务器IP 分别为 192.168.1.101 和 192.168.1.102。步骤 1修改 Broker 配置文件在两台服务器的/usr/local/rocketmq/conf/2m-2s-async目录下已有默认配置文件需修改 IP 为实际服务器 IP# 服务器1192.168.1.101修改broker-a.propertiesMaster和broker-a-s.propertiesSlavevim/usr/local/rocketmq/conf/2m-2s-async/broker-a.properties# 修改以下参数brokerIP1192.168.1.101# 当前服务器IPnamesrvAddr192.168.1.101:9876;192.168.1.102:9876# 两个NameServer地址vim/usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties# 修改以下参数brokerIP1192.168.1.101namesrvAddr192.168.1.101:9876;192.168.1.102:9876# 服务器2192.168.1.102修改broker-b.propertiesMaster和broker-b-s.propertiesSlavevim/usr/local/rocketmq/conf/2m-2s-async/broker-b.propertiesbrokerIP1192.168.1.102namesrvAddr192.168.1.101:9876;192.168.1.102:9876vim/usr/local/rocketmq/conf/2m-2s-async/broker-b-s.propertiesbrokerIP1192.168.1.102namesrvAddr192.168.1.101:9876;192.168.1.102:9876步骤 2启动集群节点按以下顺序启动先启动所有 NameServer再启动 Master最后启动 Slave# 1. 启动两台服务器的NameServer# 服务器1nohupsh/usr/local/rocketmq/bin/mqnamesrv# 服务器2nohupsh/usr/local/rocketmq/bin/mqnamesrv# 2. 启动Master节点# 服务器1启动broker-aMasternohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-a.properties# 服务器2启动broker-bMasternohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-b.properties# 3. 启动Slave节点# 服务器1启动broker-a-sSlavenohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties# 服务器2启动broker-b-sSlavenohupsh/usr/local/rocketmq/bin/mqbroker-c/usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties步骤 3验证集群状态通过mqadmin工具查看集群信息# 查看集群节点状态sh/usr/local/rocketmq/bin/mqadmin clusterList-n192.168.1.101:9876# 输出结果包含4个节点2个Master2个Slave表示集群搭建成功5. RocketMQ 默认配置表常用参数部署时经常需要调整配置整理常用默认参数如下配置项默认值说明NameServer端口 9876客户端连接 NameServer 的端口Broker监听端口 10911客户端连接 Broker 的端口Broker 角色ASYNC_MASTER默认为异步 Master可改为 SYNC_MASTER同步消息存储路径~/store消息、索引、commitLog 的存储目录消息过期时间72 小时过期消息自动清理最大消息大小4MB超过会报错可修改 maxMessageSize 参数三、RocketMQ Java 开发API 与 Spring Boot 集成环境搭好后就可以写代码了先讲原生 Java API再讲 Spring Boot 集成更符合实际开发。1. 原生 Java API 开发基础先创建一个 Maven 项目引入 RocketMQ 依赖dependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.9.5/version!-- 与部署的RocketMQ版本一致 --/dependency1生产者发送普通消息生产者负责发送消息到 Broker核心步骤创建生产者→设置 NameServer→启动生产者→发送消息→关闭生产者。importorg.apache.rocketmq.client.exception.MQClientException;importorg.apache.rocketmq.client.producer.DefaultMQProducer;importorg.apache.rocketmq.client.producer.SendResult;importorg.apache.rocketmq.common.message.Message;importorg.apache.rocketmq.remoting.common.RemotingHelper;publicclassSimpleProducer{publicstaticvoidmain(String[]args)throwsException{// 1. 创建生产者指定生产者组名同一组的生产者共享负载均衡DefaultMQProducerproducernewDefaultMQProducer(simple-producer-group);// 2. 设置NameServer地址单机localhost:9876集群192.168.1.101:9876;192.168.1.102:9876producer.setNamesrvAddr(localhost:9876);// 3. 启动生产者必须在发送消息前启动producer.start();// 4. 构建消息参数Topic名、Tag过滤用、消息内容Stringtopicsimple-topic;Stringtagtag1;StringcontentHello RocketMQ! This is a simple message.;MessagemessagenewMessage(topic,tag,content.getBytes(RemotingHelper.DEFAULT_CHARSET)// 消息内容转字节数组);// 5. 发送消息同步发送等待Broker返回结果SendResultsendResultproducer.send(message);System.out.println(消息发送成功结果sendResult);// 6. 关闭生产者建议在应用关闭时调用避免资源泄漏producer.shutdown();}}2消费者订阅并消费消息消费者负责从 Broker 订阅并消费消息核心步骤创建消费者→设置 NameServer→订阅 Topic→注册消息监听→启动消费者。importorg.apache.rocketmq.client.consumer.DefaultMQPushConsumer;importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;importorg.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;importorg.apache.rocketmq.common.message.MessageExt;importjava.util.List;publicclassSimpleConsumer{publicstaticvoidmain(String[]args)throwsMQClientException{// 1. 创建Push消费者指定消费者组名同一组的消费者共享消息DefaultMQPushConsumerconsumernewDefaultMQPushConsumer(simple-consumer-group);// 2. 设置NameServer地址consumer.setNamesrvAddr(localhost:9876);// 3. 订阅Topic*表示订阅所有Tag也可指定具体Tag如tag1||tag2consumer.subscribe(simple-topic,*);// 4. 注册消息监听器并发消费默认一次消费一条consumer.registerMessageListener(newMessageListenerConcurrently(){OverridepublicConsumeConcurrentlyStatusconsumeMessage(ListMessageExtmsgs,ConsumeConcurrentlyContextcontext){// 遍历处理消息for(MessageExtmsg:msgs){try{// 解析消息内容StringcontentnewString(msg.getBody(),UTF-8);System.out.println(消费者收到消息Topicmsg.getTopic(), Tagmsg.getTags(), Contentcontent);}catch(Exceptione){e.printStackTrace();// 消费失败返回RECONSUME_LATERBroker会重试默认重试16次returnConsumeConcurrentlyStatus.RECONSUME_LATER;}}// 消费成功返回CONSUME_SUCCESSreturnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 5. 启动消费者启动后会持续监听消息consumer.start();System.out.println(消费者启动成功等待接收消息...);}}运行顺序先启动消费者确保消费者已订阅 Topic避免消息发送后消费者未启动导致消息堆积再启动生产者发送消息查看消费者控制台会输出收到的消息内容。2. Spring Boot 集成 RocketMQ实战实际开发中更常用 Spring Boot 集成通过spring-boot-starter-rocketmq简化开发。步骤 1引入依赖dependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-spring-boot-starter/artifactIdversion2.2.3/version!-- 与RocketMQ版本匹配4.9.5对应2.2.3 --/dependency步骤 2配置 application.ymlrocketmq:name-server:localhost:9876# NameServer地址集群用分号分隔producer:group:springboot-producer-group# 生产者组名send-message-timeout:3000# 发送消息超时时间毫秒步骤 3生产者使用 RocketMQTemplateSpring Boot 提供RocketMQTemplate封装了原生 API简化发送逻辑importorg.apache.rocketmq.spring.core.RocketMQTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestParam;importorg.springframework.web.bind.annotation.RestController;RestControllerpublicclassRocketMQProducerController{// 注入RocketMQTemplateAutowiredprivateRocketMQTemplaterocketMQTemplate;// 发送普通消息通过HTTP接口触发GetMapping(/sendMsg)publicStringsendMsg(RequestParamStringcontent){Stringtopicspringboot-topic;Stringtagtag1;// 发送消息格式Topic:TagrocketMQTemplate.convertAndSend(topic:tag,content);return消息发送成功内容content;}}步骤 4消费者使用 RocketMQMessageListener通过注解RocketMQMessageListener指定消费者组和订阅的 Topic/Tag无需手动注册监听器importorg.apache.rocketmq.spring.annotation.RocketMQMessageListener;importorg.apache.rocketmq.spring.core.RocketMQListener;importorg.springframework.stereotype.Component;// 标注为消费者指定消费者组、Topic、TagComponentRocketMQMessageListener(consumerGroupspringboot-consumer-group,topicspringboot-topic,selectorExpressiontag1// 只消费tag1的消息*表示所有)publicclassRocketMQConsumerimplementsRocketMQListenerString{// 消息消费逻辑参数为消息内容自动反序列化为StringOverridepublicvoidonMessage(Stringmessage){System.out.println(Spring Boot消费者收到消息message);}}测试启动 Spring Boot 应用访问http://localhost:8080/sendMsg?contentHello RocketMQ Spring Boot查看控制台消费者会输出收到的消息内容。四、总结与预告本次学习完成了 RocketMQ 的入门了解了 RocketMQ 的发展历程、核心特点和 CAP 定位知道了它适合哪些场景实战了单机部署和双主双从集群搭建解决了启动时的常见问题用原生 Java API 和 Spring Boot 集成两种方式实现了消息的发送和消费。下一次将学习RocketMQ系统架构与核心组件。