首页
/ 正文

rocketmq springboot（长文解析）

更新时间: 2025-03-17 17:44:09

💡一则或许对你有用的小广告

欢迎加入小哈的星球 ，你将获得：专属的项目实战(已更新的所有项目都能学习) / 1v1 提问 / Java 学习路线 / 学习打卡 / 每月赠书 / 社群讨论

新开坑项目:《Spring AI 项目实战》 正在持续爆肝中，基于 Spring AI + Spring Boot 3.x + JDK 21...， 点击查看 ;

《从零手撸：仿小红书（微服务架构）》 已完结，基于 Spring Cloud Alibaba + Spring Boot 3.x + JDK 17...，点击查看项目介绍 ;演示链接： http://116.62.199.48:7070 ;

《从零手撸：前后端分离博客项目（全栈开发）》 2 期已完结，演示链接： http://116.62.199.48/ ;

截止目前，星球内专栏累计输出 90w+ 字，讲解图 3441+ 张，还在持续爆肝中.. 后续还会上新更多项目，目标是将 Java 领域典型的项目都整一波，如秒杀系统, 在线商城, IM 即时通讯，权限管理，Spring Cloud Alibaba 微服务等等，已有 3100+ 小伙伴加入学习，欢迎点击围观

在分布式系统开发中，消息队列技术扮演着至关重要的角色。RocketMQ 作为阿里巴巴开源的高性能分布式消息中间件，凭借其低延迟、高吞吐和高可靠性的特性，逐渐成为企业级应用的首选方案之一。而SpringBoot 作为 Java 生态中广受欢迎的快速开发框架，二者结合能够极大提升消息系统的开发效率。本文将从基础概念到实战案例，分步骤讲解如何在 SpringBoot 项目中集成 RocketMQ，并通过形象的比喻和代码示例，帮助读者快速掌握这一技术组合的核心能力。

一、RocketMQ 的核心概念与 SpringBoot 的结合逻辑

1.1 RocketMQ 的核心角色

RocketMQ 的基本架构包括以下核心角色：

Producer（生产者）：负责将消息发送到指定 Topic。
Broker（消息服务器）：作为消息的中转站，负责存储、转发消息。
Consumer（消费者）：从 Broker 拉取消息并进行处理。
Name Server（路由注册中心）：维护 Broker 的路由信息，帮助生产者和消费者快速定位目标节点。

比喻：可以将 RocketMQ 想象为一个快递系统。生产者是寄件人，将包裹（消息）投递到快递网点（Broker），Name Server 则是快递公司的调度中心，告知寄件人和收件人最近的网点位置，而消费者则是收件人，最终接收包裹并处理。

1.2 SpringBoot 的整合优势

SpringBoot 通过 @EnableRocketMQ 等注解简化了 RocketMQ 的集成流程，同时提供了统一的配置管理能力。开发者无需手动编写复杂的服务发现或连接逻辑，只需通过声明式注解即可快速搭建消息通信链路。

二、环境搭建与基础配置

2.1 项目依赖配置

在 SpringBoot 项目中集成 RocketMQ，需在 pom.xml 中添加以下依赖：

<dependency>  
    <groupId>org.apache.rocketmq</groupId>  
    <artifactId>rocketmq-spring-boot-starter</artifactId>  
    <version>2.2.3</version>  
</dependency>

2.2 核心配置项说明

在 application.yml 中配置 RocketMQ 的基础参数：

rocketmq:  
  name-server: localhost:9876  # Name Server 地址  
  producer:  
    group: demo-producer      # 生产者组名  
  consumer:  
    group: demo-consumer      # 消费者组名

关键点解释：

name-server 是 RocketMQ 的路由中心地址，需确保服务已启动。
group 用于标识生产者和消费者的身份，避免不同业务线的消息混淆。

三、核心功能实现

3.1 生产者：消息发送的两种模式

3.1.1 同步发送模式

同步发送会等待 Broker 的响应，确保消息成功写入后才继续执行后续代码。适合对可靠性要求高的场景。

@Autowired  
private RocketMQTemplate rocketMQTemplate;  

public void sendMessageSync() {  
    String msg = "Hello RocketMQ!";  
    SendResult result = rocketMQTemplate.syncSend("demo-topic", msg);  
    System.out.println("Message sent with status: " + result.getSendStatus());  
}

3.1.2 异步发送模式

异步发送不会阻塞当前线程，通过回调函数处理结果。适合对性能要求高的场景。

public void sendMessageAsync() {  
    rocketMQTemplate.asyncSend("demo-topic", "Async Message", (sendResult) -> {  
        System.out.println("Message ID: " + sendResult.getMsgId());  
    });  
}

比喻：同步发送像“先付款再拿货”，异步发送则像“先拿货，稍后付款”，前者更可靠但效率较低，后者效率高但需处理可能的失败情况。

3.2 消费者：消息监听与处理

通过 @RocketMQMessageListener 注解声明消费者，结合 MessageListener 接口实现消息处理逻辑。

@Component  
@RocketMQMessageListener(  
    topic = "demo-topic",  
    consumerGroup = "demo-consumer"  
)  
public class DemoConsumer implements MessageListener {  

    @Override  
    public Action executeOnMessage(Message msg) {  
        String content = new String(msg.getBody());  
        System.out.println("Received message: " + content);  
        return Action.CommitMessage;  // 提交消息，避免重复消费  
    }  
}

四、进阶用法与最佳实践

4.1 消息过滤：通过标签（Tag）精准路由

RocketMQ 支持通过 Tag 对消息进行分类，消费者可通过 MessageSelector 过滤特定标签的消息。

// 生产端指定 Tag  
rocketMQTemplate.syncSend("demo-topic", "Order Message", MessageKeys.TAG, "ORDER");  

// 消费端配置过滤规则  
@Component  
@RocketMQMessageListener(  
    topic = "demo-topic",  
    consumerGroup = "order-consumer",  
    messageModel = MessageModel.CLUSTERING,  
    selectorExpression = "TAG == 'ORDER'",  
    selectorType = SelectorType.TAG  
)  
public class OrderConsumer implements MessageListener {  
    // 处理订单相关消息  
}

4.2 事务消息：保证业务与消息的一致性

事务消息用于解决“数据库操作与消息发送”之间的最终一致性问题。通过 @RocketMQTransactionListener 注解实现本地事务与消息的双重确认。

@Component  
@RocketMQTransactionListener  
public class TxProducer {  

    @Autowired  
    private OrderService orderService;  

    // 本地事务执行方法  
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {  
        String orderId = new String(msg.getBody());  
        boolean success = orderService.createOrder(orderId);  
        return success ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;  
    }  

    // 二次检查回调（仅在本地事务状态为 UNKNOW 时触发）  
    public LocalTransactionState checkLocalTransaction(Message msg) {  
        String orderId = new String(msg.getBody());  
        OrderStatus status = orderService.checkOrderStatus(orderId);  
        return status == OrderStatus.COMPLETED ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;  
    }  
}

4.3 性能优化：批量发送与消费

通过配置 BatchMessageSender 实现批量发送，减少网络开销：

@Configuration  
public class RocketMQConfig {  

    @Bean  
    public RocketMQTemplate rocketMQTemplate(RocketMQProducer rocketMQProducer) {  
        rocketMQProducer.setBatch(true); // 开启批量发送  
        return new RocketMQTemplate(rocketMQProducer);  
    }  
}

五、实战案例：订单系统中的消息解耦

5.1 场景描述

在电商系统中，下单成功后需同步触发库存扣减、积分赠送、短信通知等多个异步操作。通过 RocketMQ 将订单状态变更与下游服务解耦，避免因下游服务故障导致下单失败。

5.2 代码实现

5.2.1 订单生产者

@Service  
public class OrderService {  

    @Autowired  
    private RocketMQTemplate rocketMQTemplate;  

    public void createOrder(String orderId) {  
        // 执行订单创建逻辑  
        // ...  

        // 发送订单状态变更消息  
        rocketMQTemplate.syncSend("order-topic", orderId, MessageKeys.TAG, "CREATE");  
    }  
}

5.2.2 库存消费者

@Component  
@RocketMQMessageListener(  
    topic = "order-topic",  
    consumerGroup = "stock-consumer",  
    selectorExpression = "TAG == 'CREATE'",  
    selectorType = SelectorType.TAG  
)  
public class StockConsumer implements MessageListener {  

    @Override  
    public Action executeOnMessage(Message msg) {  
        String orderId = new String(msg.getBody());  
        stockService.deductStock(orderId);  
        return Action.CommitMessage;  
    }  
}

5.3 优势总结

解耦性：订单服务与库存服务无直接依赖，降低系统复杂度。
容错性：即使库存服务暂时不可用，订单仍可正常提交，待恢复后继续处理消息。

六、结论

通过本文的讲解，我们系统梳理了 RocketMQ 在 SpringBoot 项目中的核心用法，从基础配置到进阶场景，逐步展示了如何利用这一技术组合构建高可靠的消息系统。无论是同步/异步发送、消息过滤，还是事务消息的复杂场景，开发者均可通过 SpringBoot 的注解驱动和 RocketMQ 的强大能力快速实现。

在实际开发中，建议根据业务需求选择合适的消息模式，并结合监控工具（如 RocketMQ 控制台）实时跟踪消息状态。随着分布式系统的规模扩大，RocketMQ 与 SpringBoot 的结合将成为构建弹性架构的重要基石。

（全文约 1800 字）

rocketmq springboot（长文解析）

一、RocketMQ 的核心概念与 SpringBoot 的结合逻辑

1.1 RocketMQ 的核心角色

1.2 SpringBoot 的整合优势

二、环境搭建与基础配置

2.1 项目依赖配置

2.2 核心配置项说明

三、核心功能实现

3.1 生产者：消息发送的两种模式

3.1.1 同步发送模式

3.1.2 异步发送模式

3.2 消费者：消息监听与处理

四、进阶用法与最佳实践

4.1 消息过滤：通过标签（Tag）精准路由

4.2 事务消息：保证业务与消息的一致性

4.3 性能优化：批量发送与消费

五、实战案例：订单系统中的消息解耦

5.1 场景描述

5.2 代码实现

5.2.1 订单生产者

5.2.2 库存消费者

5.3 优势总结

六、结论

最新发布

RubyMine 2025.1.4.1 最新激活码，破解版安装教程（亲测有效~）

Clion 2025.1.4 最新激活码，破解版安装教程（至2099年~）

Pycharm 2025.1.3.1 最新激活码，破解版安装教程（至2099年~）

GoLand 2025.1.4 最新激活码，破解版安装教程（至2099年~）

PhpStorm 2025.1.4.1 最新激活码，破解版安装教程（亲测有效~）

Webstorm 2025.1.4.1 最新激活码，破解版安装教程（至2099年~）

IDEA 2025.1.4.1 最新激活码，破解版安装教程（至2099年~）

DBeaver 23.2 破解安装教程（附激活码，亲测有效~）

SQLyog Ultimate 破解版安装教程（附证书秘钥注册码）

Navicat 16 破解版安装教程_激活码(永久激活,亲测好用)