RocketMQ 入门第一篇：从概述到实战（环境搭建 + Java API+Spring Boot 集成）- 惊觉

RocketMQ 入门第一篇：从概述到实战（环境搭建 + Java API+Spring Boot 集成）

在分布式系统中，消息中间件是 “解耦、削峰、异步” 的关键组件，而 RocketMQ 作为 Apache 顶级项目，凭借高可用、高吞吐、支持多种消息类型的特性，成为很多企业的首选。这篇作为 RocketMQ 系列的第一篇，会从 “是什么→怎么搭环境→怎么写代码” 三个维度，一起学习快速入门 RocketMQ，零基础也能跟着做。

一、RocketMQ 概述：它是什么？为什么选它？

在动手之前，先搞清楚 RocketMQ 的 “出身” 和核心能力，避免盲目上手。

1. RocketMQ 的 “前世今生”

RocketMQ 并非凭空出现，它的发展历程体现了其成熟度：

2012 年： 由阿里巴巴开源，最初用于阿里内部的电商场景（如双 11 的消息流转）；
2016 年： 捐赠给 Apache 软件基金会，进入孵化阶段；
2017 年： 成为 Apache 顶级项目，正式走向开源社区，目前已更新到 5.x 版本（支持云原生、gRPC 协议）。

核心定位：纯 Java 编写的分布式消息中间件，基于队列模型，专注于解决分布式系统中的消息通信问题。

2. RocketMQ 的核心特点：为什么选它？

相比 RabbitMQ、Kafka，RocketMQ 有自己独特的优势，尤其适合企业级场景：

类别	核心特性
消息顺序	单队列内严格保证 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/（从入门到进阶的完整指南）；
官方 FAQ：http://rocketmq.apache.org/docs/faq/（解决常见问题，如启动失败、消息丢失）；
商业版：阿里云RocketMQ：https://www.aliyun.com/product/rocketmq，开箱即用，适合企业生产环境。

二、RocketMQ 环境搭建：从单机到集群

环境搭建是入门的第一步，先从简单的单机部署入手，再了解集群方案，最后实战双主双从集群（生产环境常用）。

1. 前置准备：环境要求

RocketMQ 对环境的要求不高，确保满足以下条件：

JDK： 1.8 及以上（推荐 1.8，避免高版本兼容问题）；
操作系统： Linux（推荐，生产环境）、Windows（开发环境，注意路径不含中文）；
内存： 至少 2GB（默认启动脚本占用内存较大，可修改）；
Maven： 3.5 及以上（编译源码用，若直接下载二进制包可不用）。

2. 单机部署：3 步启动（以 Linux 为例）

单机部署适合开发和测试，步骤简单，5 分钟就能搞定。

步骤 1：下载并解压二进制包

推荐直接下载二进制包（无需编译），地址：https://rocketmq.apache.org/zh/download，选择稳定版本（如 4.9.5，兼容性好）。

# 1. 下载（也可手动下载后上传到Linux）
wget https://archive.apache.org/dist/rocketmq/4.9.5/rocketmq-all-4.9.5-bin-release.zip

# 2. 解压到/usr/local目录
unzip rocketmq-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.sh）
vim /usr/local/rocketmq/bin/runserver.sh
# 找到以下行，修改为小内存（如Xms512m Xmx512m Xmn256m）
# 原配置：JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

# 2. 修改Broker的JVM参数（runbroker.sh）
vim /usr/local/rocketmq/bin/runbroker.sh
# 原配置：JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m"

步骤 3：启动 NameServer 和 Broker

RocketMQ 的核心组件是NameServer（路由注册中心）和Broker（消息存储和转发），必须先启动 NameServer，再启动 Broker。

# 1. 启动NameServer（后台启动，日志输出到nohup.out）
cd /usr/local/rocketmq
nohup sh bin/mqnamesrv &

# 2. 查看NameServer启动日志（确认是否成功）
tail -f ~/logs/rocketmqlogs/namesrv.log
# 出现以下日志表示启动成功：The Name Server boot success...

# 3. 启动Broker（指定NameServer地址为本地，后台启动）
# -n：指定NameServer地址（格式：ip:port，默认端口9876）
# autoCreateTopicEnable=true：允许自动创建Topic（开发环境用，生产环境建议关闭）
nohup sh bin/mqbroker -n localhost:9876 autoCreateTopicEnable=true &

# 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？
原因：防火墙未开放 9876（NameServer 端口）和 10911（Broker 默认端口），执行以下命令开放端口：

# 开放9876和10911端口
firewall-cmd --zone=public --add-port=9876/tcp --permanent
firewall-cmd --zone=public --add-port=10911/tcp --permanent
# 重新加载防火墙规则
firewall-cmd --reload

3. 集群方案分析：生产环境怎么部署？

单机部署仅适合开发，生产环境必须用集群，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：

# 服务器1（192.168.1.101）：修改broker-a.properties（Master）和broker-a-s.properties（Slave）
vim /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties
# 修改以下参数：
brokerIP1=192.168.1.101 # 当前服务器IP
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876 # 两个NameServer地址

vim /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties
# 修改以下参数：
brokerIP1=192.168.1.101
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876

# 服务器2（192.168.1.102）：修改broker-b.properties（Master）和broker-b-s.properties（Slave）
vim /usr/local/rocketmq/conf/2m-2s-async/broker-b.properties
brokerIP1=192.168.1.102
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876

vim /usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties
brokerIP1=192.168.1.102
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876

步骤 2：启动集群节点

按以下顺序启动（先启动所有 NameServer，再启动 Master，最后启动 Slave）：

# 1. 启动两台服务器的NameServer
# 服务器1
nohup sh /usr/local/rocketmq/bin/mqnamesrv &
# 服务器2
nohup sh /usr/local/rocketmq/bin/mqnamesrv &

# 2. 启动Master节点
# 服务器1：启动broker-a（Master）
nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties &
# 服务器2：启动broker-b（Master）
nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b.properties &

# 3. 启动Slave节点
# 服务器1：启动broker-a-s（Slave）
nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties &
# 服务器2：启动broker-b-s（Slave）
nohup sh /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 -n 192.168.1.101:9876
# 输出结果包含4个节点（2个Master+2个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 依赖：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.5</version> <!-- 与部署的RocketMQ版本一致 -->
</dependency>

（1）生产者：发送普通消息

生产者负责发送消息到 Broker，核心步骤：创建生产者→设置 NameServer→启动生产者→发送消息→关闭生产者。

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class SimpleProducer {
    public static void main(String[] args) throws Exception {
        // 1. 创建生产者（指定生产者组名，同一组的生产者共享负载均衡）
        DefaultMQProducer producer = new DefaultMQProducer("simple-producer-group");
        
        // 2. 设置NameServer地址（单机：localhost:9876；集群：192.168.1.101:9876;192.168.1.102:9876）
        producer.setNamesrvAddr("localhost:9876");
        
        // 3. 启动生产者（必须在发送消息前启动）
        producer.start();
        
        // 4. 构建消息（参数：Topic名、Tag（过滤用）、消息内容）
        String topic = "simple-topic";
        String tag = "tag1";
        String content = "Hello RocketMQ! This is a simple message.";
        Message message = new Message(
                topic,
                tag,
                content.getBytes(RemotingHelper.DEFAULT_CHARSET) // 消息内容转字节数组
        );
        
        // 5. 发送消息（同步发送，等待Broker返回结果）
        SendResult sendResult = producer.send(message);
        System.out.println("消息发送成功！结果：" + sendResult);
        
        // 6. 关闭生产者（建议在应用关闭时调用，避免资源泄漏）
        producer.shutdown();
    }
}

（2）消费者：订阅并消费消息

消费者负责从 Broker 订阅并消费消息，核心步骤：创建消费者→设置 NameServer→订阅 Topic→注册消息监听→启动消费者。

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class SimpleConsumer {
    public static void main(String[] args) throws MQClientException {
        // 1. 创建Push消费者（指定消费者组名，同一组的消费者共享消息）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("simple-consumer-group");
        
        // 2. 设置NameServer地址
        consumer.setNamesrvAddr("localhost:9876");
        
        // 3. 订阅Topic（*表示订阅所有Tag，也可指定具体Tag，如"tag1||tag2"）
        consumer.subscribe("simple-topic", "*");
        
        // 4. 注册消息监听器（并发消费，默认一次消费一条）
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 遍历处理消息
                for (MessageExt msg : msgs) {
                    try {
                        // 解析消息内容
                        String content = new String(msg.getBody(), "UTF-8");
                        System.out.println("消费者收到消息：Topic=" + msg.getTopic() + ", Tag=" + msg.getTags() + ", Content=" + content);
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 消费失败，返回RECONSUME_LATER，Broker会重试（默认重试16次）
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 消费成功，返回CONSUME_SUCCESS
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        // 5. 启动消费者（启动后会持续监听消息）
        consumer.start();
        System.out.println("消费者启动成功，等待接收消息...");
    }
}

运行顺序：

先启动消费者（确保消费者已订阅 Topic，避免消息发送后消费者未启动导致消息堆积）；
再启动生产者，发送消息；
查看消费者控制台，会输出收到的消息内容。

2. Spring Boot 集成 RocketMQ（实战）

实际开发中，更常用 Spring Boot 集成，通过spring-boot-starter-rocketmq简化开发。

步骤 1：引入依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version> <!-- 与RocketMQ版本匹配，4.9.5对应2.2.3 -->
</dependency>

步骤 2：配置 application.yml

rocketmq:
  name-server: localhost:9876 # NameServer地址（集群用分号分隔）
  producer:
    group: springboot-producer-group # 生产者组名
    send-message-timeout: 3000 # 发送消息超时时间（毫秒）

步骤 3：生产者（使用 RocketMQTemplate）

Spring Boot 提供RocketMQTemplate，封装了原生 API，简化发送逻辑：

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class RocketMQProducerController {
    // 注入RocketMQTemplate
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 发送普通消息（通过HTTP接口触发）
    @GetMapping("/sendMsg")
    public String sendMsg(@RequestParam String content) {
        String topic = "springboot-topic";
        String tag = "tag1";
        // 发送消息（格式：Topic:Tag）
        rocketMQTemplate.convertAndSend(topic + ":" + tag, content);
        return "消息发送成功！内容：" + content;
    }
}

步骤 4：消费者（使用 @RocketMQMessageListener）

通过注解@RocketMQMessageListener指定消费者组和订阅的 Topic/Tag，无需手动注册监听器：

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

// 标注为消费者，指定消费者组、Topic、Tag
@Component
@RocketMQMessageListener(
        consumerGroup = "springboot-consumer-group",
        topic = "springboot-topic",
        selectorExpression = "tag1" // 只消费tag1的消息，*表示所有
)
public class RocketMQConsumer implements RocketMQListener<String> {
    // 消息消费逻辑（参数为消息内容，自动反序列化为String）
    @Override
    public void onMessage(String message) {
        System.out.println("Spring Boot消费者收到消息：" + message);
    }
}

测试：

启动 Spring Boot 应用；
访问http://localhost:8080/sendMsg?content=Hello RocketMQ Spring Boot；
查看控制台，消费者会输出收到的消息内容。

四、总结与预告

本次学习完成了 RocketMQ 的入门：

了解了 RocketMQ 的发展历程、核心特点和 CAP 定位，知道了它适合哪些场景；
实战了单机部署和双主双从集群搭建，解决了启动时的常见问题；
用原生 Java API 和 Spring Boot 集成两种方式，实现了消息的发送和消费。

下一次将学习RocketMQ系统架构与核心组件。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2403_88625492/article/details/153960260

RocketMQ 入门第一篇：从概述到实战（环境搭建 + Java API+Spring Boot 集成）