Spring Boot 4.0 整合 RocketMQ：消息发送与消费治理

作者：苏三2026-11-07Spring Boot 4.0Spring Boot 4

前言

订单创建后还要通知积分、库存和搜索服务。同步调用会让订单接口依赖越来越多下游，消息队列把主链路和后续动作拆开。

这篇教程用 RocketMQ 4.x 原生客户端发送订单事件，并启动一个并发消费者。

配套源码：dvsusan/spring-boot4-demo。

最终结果

订单消息生产和消费流程

图 1：生产者通过 NameServer 找到 Broker，消费者成功处理后确认，失败进入重试。

第 1 步：启动 RocketMQ

docker compose up -d rocketmq-namesrv rocketmq-broker
docker compose ps

当前配置启用自动创建 Topic，仅用于本地教学。生产环境应提前规划 Topic、队列数、权限和清理策略。

第 2 步：创建并启动生产者

// 开启 MQ 章节时才创建并启动生产者，避免默认启动依赖 Broker。
@Bean(destroyMethod = "shutdown")
public DefaultMQProducer rocketMqProducer(
        @Value("${demo.chapter4.rocketmq.namesrv:127.0.0.1:9876}")
        String namesrv,
        @Value("${demo.chapter4.rocketmq.producer-group:chapter4-producer}")
        String group
) throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer(group);
    producer.setNamesrvAddr(namesrv);
    producer.start();
    return producer;
}

生产者组用于标识同类生产者。应用关闭时必须 shutdown，否则本地运行多次会留下连接和后台线程。

第 3 步：发送订单事件

public SendResult publish(String tag, String payload) throws Exception {
    Message message = new Message(
            topic,
            tag,
            payload.getBytes(StandardCharsets.UTF_8)
    );
    // 生产环境还应传入业务唯一键，便于查询和幂等处理。
    return producer.send(message);
}

标签 ORDER_CREATED 用于订阅过滤。消息体建议使用带 eventId、eventType、occurredAt 和版本的 JSON。

第 4 步：消费并确认

// RocketMQ 会交给线程池并发调用该方法。
@Override
public ConsumeConcurrentlyStatus consumeMessage(
        List<MessageExt> messages,
        ConsumeConcurrentlyContext context
) {
    for (MessageExt message : messages) {
        String body = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("received order event: msgId={}, body={}",
                message.getMsgId(), body);
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

只有业务处理完成后才返回 CONSUME_SUCCESS。异常、数据库超时或下游失败应返回重试状态。重试意味着至少一次投递，因此消费者必须幂等。

第 5 步：启动并验证

mvn clean package -DskipTests

java -jar demo-web/target/demo-web-1.0.0-SNAPSHOT.jar \
  --server.port=18094 \
  --demo.chapter4.rocketmq.enabled=true \
  --demo.chapter4.rocketmq.consume=true

发送消息：

curl -X POST \
  "http://localhost:18094/api/v1/chapter4/messages/orders/SO1001"

预期响应包含 RocketMQ msgId，控制台能看到相同订单号的事件。

消息治理边界

消息体不要放超大对象，只放标识和必要快照。
重试必须幂等，以 eventId 或业务唯一键去重。
设置最大重试次数和死信处理，不让毒消息无限循环。
消费端失败要可观测：日志、重试次数、死信数量和消费延迟。
顺序消息只在确有顺序要求时使用，避免牺牲吞吐。

常见问题排查

现象	原因	处理方式
找不到 Broker	NameServer 地址错误	检查 `127.0.0.1:9876`
发送超时	Broker 未就绪或端口不通	查看 Compose 日志和 10911 端口
反复消费	消费逻辑抛异常	修复业务并保证幂等
应用启动失败	开启 MQ 但服务未启动	关闭开关或先启动 RocketMQ
消息丢失	只看发送返回值，没有重试机制	使用 Outbox 和发送确认

练习题

增加 eventId 消费去重表，同一个事件重复消费时直接返回成功。

总结

RocketMQ 解决异步通信，不自动解决业务幂等。生产者确认、消费者重试、死信和唯一键必须一起设计。

下一篇解决“数据库成功、消息发送失败”的最终一致性问题。

上一节：Redisson 分布式锁
 下一节：事务消息与 Outbox

13 个企业级实战项目 · 源码 + 教程 + 答疑

简历包装指导AI Agent 实战高并发·微服务Offer 导向持续更新3000+ 同行交流1V1 答疑3 天包退

覆盖高并发、微服务、分布式与 AI Agent：100万QPS短链、企业智能知识库、商城微服务、SaaS点餐、秒杀、刷题小程序、代码生成器、AI Agent 等。券后 ¥129（原价 ¥159），不满意 3 天包退。

立即获取 13 个项目源码 →