SpringBoot 接入 RocketMQ 全教程：Tag 过滤、批量发送、事务消息一站式实现

青年逐梦

125人浏览 · 2026-06-14 12:11:25

青年逐梦 · 2026-06-14 12:11:25 发布

一、组件说明

RocketMQ 分4个核心角色：

NameServer：注册中心，Broker路由管理，无状态集群；
Broker：真正存消息的服务，分Master/Slave；单机部署只用单个Master；
Producer生产者：发送消息；
Consumer消费者：订阅Topic、拉取/监听消息。

二、Windows单机部署

2个cmd窗口分别启动

1、官方下载地址

二进制压缩包：
下载地址
Windows选 rocketmq-all-x.x.x-bin-release.zip

解压路径无中文、空格，示例：D:\soft\rocketmq-5.3.0

2、修改JVM内存

可选

bin/runserver.cmd

set JAVA_OPT=-Xms128m -Xmx128m

bin/runbroker.cmd

set JAVA_OPT=-Xms128m -Xmx128m

3、启动NameServer

cd D:\soft\rocketmq-5.3.0\bin
start mqnamesrv.cmd

默认端口：9876

4、启动Broker

mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

autoCreateTopicEnable=true 测试环境自动创建Topic，不用手动建。

5、验证安装

工具脚本测试发送/消费

# 发送
tools.cmd org.apache.rocketmq.example.quickstart.Producer
# 消费
tools.cmd org.apache.rocketmq.example.quickstart.Consumer

三、SpringBoot 接入RocketMQ

Maven依赖

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

application.yml 基础配置

rocketmq:
  name-server: 127.0.0.1:9876
  # 生产者组
  producer:
    group: order-producer-group

四、生产者发送消息

1）普通同步消息

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MqProducerService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 目标Topic：order_topic
    public void sendNormalMsg() {
        String msg = "订单创建消息，订单号：10001";
        rocketMQTemplate.syncSend("order_topic", msg);
    }
}

2）带Tag过滤消息

消费者可以只订阅指定Tag，实现消息分类过滤

// tag=pay
rocketMQTemplate.syncSend("order_topic:pay", "支付消息");
// tag=cancel
rocketMQTemplate.syncSend("order_topic:cancel", "取消订单消息");

3）延迟消息（常用：超时未支付自动关单）

// 延迟等级 1~18，等级3=10s
rocketMQTemplate.syncSend("order_topic", MessageBuilder.withPayload("超时未支付").build(), 3);

4）批量发送消息

List<Message<String>> msgList = new ArrayList<>();
msgList.add(MessageBuilder.withPayload("msg1").build());
msgList.add(MessageBuilder.withPayload("msg2").build());
rocketMQTemplate.syncSendBatch("order_topic", msgList);

五、消费者监听代码

@RocketMQMessageListener

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

@Service
@RocketMQMessageListener(
        topic = "order_topic",
        consumerGroup = "order-consumer-group",
        // 只消费tag=pay的消息
        selectorExpression = "pay"
)
public class OrderConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("收到消息：" + message);
        // 业务处理：更新订单、扣库存等
    }
}

方法正常执行完毕：自动ACK确认，Broker删除这条消息；
抛出异常：消息重试投递。

六、核心高级场景：事务消息

分布式事务可靠投递

下单+发消息扣库存，保证原子性：

半消息发送成功；
本地订单事务执行；
事务成功：Commit，消息投递给消费者；
事务失败：Rollback，消息丢弃。

// 发送事务消息
TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
        "tx-producer-group",
        "order_topic",
        MessageBuilder.withPayload("订单事务消息").build(),
        null
);