当前位置：首页 > news >正文

RocketMQ-技术详解

news 2026/5/12 15:06:01

----------------------------------------------------------------------------------------

RocketMQ 集成基础知识（新手友好版）

RocketMQ 是阿里开源的分布式消息中间件，核心用于解决系统间异步通信、削峰填谷、解耦等问题（比如电商下单后通知物流、支付回调、日志异步采集等场景）。下面从「核心概念→环境准备→基础集成→核心操作」一步步讲，新手能直接跟着做。

一、核心概念（先懂这些，集成不懵）

表格

概念	通俗解释
Producer	消息生产者：负责发送消息到 RocketMQ（比如下单系统发送 “订单创建” 消息）
Consumer	消息消费者：负责接收并处理消息（比如物流系统接收 “订单创建” 消息开始发货）
Topic	消息主题：消息的 “分类标签”，生产者发消息到指定 Topic，消费者订阅 Topic 收消息
Queue	队列：Topic 下的细分队列，用于负载均衡（一个 Topic 可包含多个 Queue）
Broker	RocketMQ 服务端节点：存储消息、转发消息（核心服务）
NameServer	注册中心：管理 Broker 地址，生产者 / 消费者通过它找到 Broker
Group	分组：Producer Group（生产者分组）/ Consumer Group（消费者分组），用于标识一组生产 / 消费实例
消息模式	集群模式（同组消费者分摊消息）、广播模式（同组消费者都收到所有消息）

二、环境准备（以 Java 为例，最常用）

安装 RocketMQ：先部署 RocketMQ 服务端（本地测试可装单机版），步骤：
- 下载安装包（https://rocketmq.apache.org/zh/download/）；
- 启动 NameServer：nohup sh bin/mqnamesrv &；
- 启动 Broker：nohup sh bin/mqbroker -n localhost:9876 &。
项目依赖：Maven 引入核心依赖（Spring Boot 项目推荐用 rocketmq-spring-boot-starter）：

xml

<!-- Spring Boot 集成依赖（推荐） -->
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version>
</dependency>
<!-- 原生依赖（非 Spring Boot 项目） -->
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.7</version>
</dependency>

配置文件（application.yml）：Spring Boot 项目配置 RocketMQ 连接信息：

yaml

rocketmq:name-server: localhost:9876  # NameServer 地址producer:group: demo-producer-group  # 生产者分组send-message-timeout: 3000  # 发送超时时间consumer:group: demo-consumer-group  # 消费者分组

三、基础集成：生产 / 消费消息（开箱即用）

1. 消息生产者（发送消息）

java

运行

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class DemoProducer {// 注入 RocketMQ 模板（Spring Boot 封装好的工具）@Autowiredprivate RocketMQTemplate rocketMQTemplate;// 发送普通同步消息（最常用）public void sendNormalMessage(String topic, String msg) {try {// 参数1：Topic名称（可加Tag，格式：topic:tag）；参数2：消息内容rocketMQTemplate.convertAndSend(topic, msg);System.out.println("消息发送成功：" + msg);} catch (Exception e) {System.out.println("消息发送失败：" + e.getMessage());// 失败可重试/记录日志}}
}

2. 消息消费者（接收消息）

java

运行

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;// 注解配置：topic=订阅的主题，consumerGroup=消费者分组
@Component
@RocketMQMessageListener(topic = "demo-topic", consumerGroup = "demo-consumer-group")
public class DemoConsumer implements RocketMQListener<String> {// 接收到消息后执行的方法@Overridepublic void onMessage(String msg) {try {// 处理消息（比如业务逻辑：更新订单状态、通知物流等）System.out.println("收到消息：" + msg);// 无需手动确认，默认自动确认（集群模式下消息会分摊给同组消费者）} catch (Exception e) {System.out.println("消息处理失败：" + e.getMessage());// 处理失败可抛异常，RocketMQ 会重试（默认重试16次）}}
}

3. 测试调用

java

运行

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;@SpringBootApplication
public class RocketMQDemoApplication {public static void main(String[] args) {SpringApplication.run(RocketMQDemoApplication.class, args);}// 项目启动后发送测试消息@Beanpublic CommandLineRunner testSendMessage(DemoProducer producer) {return args -> {producer.sendNormalMessage("demo-topic", "这是一条测试纠纷案件消息");};}
}