SpringBoot接入阿里云RocketMQ的完整指南
更新時(shí)間：2026年07月08日 08:35:34 作者：霸道流氓氣質(zhì)
本文介紹了SpringBoot框架及其簡(jiǎn)化開(kāi)發(fā)的特點(diǎn),以及RocketMQ的分布式消息中間件功能,重點(diǎn)講述了如何在SpringBoot項(xiàng)目中集成RocketMQ,實(shí)現(xiàn)消息的高效生產(chǎn)和消費(fèi),適用于多種分布式應(yīng)用場(chǎng)景,需要的朋友可以參考下
一、核心概念
1.1 實(shí)例（Instance）
消息服務(wù)的頂層容器，按環(huán)境隔離。ID 格式：MQ_INST_<賬號(hào)ID>_<標(biāo)識(shí)>
阿里云賬號(hào) ├── 實(shí)例-dev (MQ_INST_xxx_BXsxZrYe) → 開(kāi)發(fā)環(huán)境 └── 實(shí)例-prod (MQ_INST_xxx_BX3z6RD9) → 生產(chǎn)環(huán)境
1.2 Topic
消息通道，Producer 發(fā)到 Topic，Consumer 從 Topic 消費(fèi)。
1.3 GroupId
消費(fèi)者組標(biāo)識(shí)。同組內(nèi)多實(shí)例競(jìng)爭(zhēng)消費(fèi)（一條消息只被一個(gè)實(shí)例消費(fèi)）。
阿里云硬性要求：同一 GroupId 下所有消費(fèi)者必須訂閱完全相同的 Topic 集合，否則報(bào) subscription not consistent。
1.4 Tag
消息二級(jí)標(biāo)簽，Consumer 可按 Tag 過(guò)濾。
1.5 Endpoint
接入地址。本地開(kāi)發(fā)用公網(wǎng)，服務(wù)器部署用內(nèi)網(wǎng)：
| 類型 | 格式 | 場(chǎng)景 |
|---|---|---|
| 公網(wǎng) | http://MQ_INST_xxx.mq-internet-access.mq-internet.aliyuncs.com:80 | 本地開(kāi)發(fā) |
| 內(nèi)網(wǎng) | http://MQ_INST_xxx.cn-xxxx.mq-internal.aliyuncs.com:8080 | 同地域ECS |
1.6 AccessKey / SecretKey
身份認(rèn)證憑證。在阿里云 RAM 中創(chuàng)建，建議用子賬號(hào)，最小權(quán)限原則。
二、阿里云控制臺(tái)操作步驟
2.1 創(chuàng)建實(shí)例
- 登錄阿里云 → 搜索"消息隊(duì)列 RocketMQ" → 創(chuàng)建實(shí)例
- 選擇地域（如華北1青島）、規(guī)格
- 獲取實(shí)例 ID 和 Endpoint
2.2 創(chuàng)建 Topic
實(shí)例詳情 → Topic 管理 → 創(chuàng)建 Topic：
- 名稱：
my-app-import-check-dev - 消息類型：普通消息
2.3 創(chuàng)建 GroupId
實(shí)例詳情 → Group 管理 → 創(chuàng)建 Group：
- 名稱必須以
GID-或GID_開(kāi)頭 - 如
GID-MY-APP-IMPORT-CHECK
2.4 獲取 AccessKey
右上角頭像 → AccessKey 管理 → 創(chuàng)建 RAM 子賬號(hào) → 授權(quán) AliyunMQFullAccess
三、YAML 配置：數(shù)組格式 vs Map 格式
3.1 數(shù)組格式
rocketmq:
consumers:
- topic: my-app-import-check-dev
groupId: GID-MY-APP-CHECK
tag: app-tag
- topic: my-app-import-save-dev
groupId: GID-MY-APP-SAVE
tag: app-tag代碼引用：${rocketmq.consumers[0].tag}、${rocketmq.consumers[1].tag}
特點(diǎn)：
- Java 中映射為
List<ConsumerProperties> - 適合程序動(dòng)態(tài)遍歷創(chuàng)建消費(fèi)者
- 順序依賴（
[0]/[1]必須和配置順序一致）
3.2 Map 格式
rocketmq:
consumers:
import-check:
topic: my-app-import-check-dev
groupId: GID-MY-APP-CHECK
tag: app-tag
import-save:
topic: my-app-import-save-dev
groupId: GID-MY-APP-SAVE
tag: app-tag代碼引用：${rocketmq.consumers.import-check.tag}
特點(diǎn)：
- Java 中映射為
Map<String, ConsumerProperties> - 按名稱引用，不依賴順序
- 語(yǔ)義更清晰
3.3 關(guān)鍵區(qū)別
| 維度 | 數(shù)組格式 | Map 格式 |
|---|---|---|
| 引用方式 | [0]、[1]（位置） | .name（名稱） |
| 可讀性 | 低（依賴順序） | 高（見(jiàn)名知義） |
| 擴(kuò)展性 | 插入/刪除影響索引 | 獨(dú)立不互相影響 |
| 框架注冊(cè) | 遍歷 List 逐個(gè)創(chuàng)建 | 遍歷 Map.entrySet 逐個(gè)創(chuàng)建 |
3.4 一個(gè)數(shù)組元素訂閱多 Topic vs 多個(gè)獨(dú)立 GroupId
方式A：一個(gè)消費(fèi)者訂閱多 Topic
consumers:
- topic: topicA|topicB|topicC
groupId: GID-ALL-IN-ONE- 優(yōu)點(diǎn)：配置簡(jiǎn)單，只需一個(gè) GroupId
- 缺點(diǎn)：無(wú)法對(duì)不同 Topic 設(shè)置不同的并發(fā)度/重試策略
方式B：多個(gè)消費(fèi)者各自獨(dú)立
consumers:
check:
topic: topicA
groupId: GID-CHECK
save:
topic: topicB
groupId: GID-SAVE
export:
topic: topicC
groupId: GID-EXPORT- 優(yōu)點(diǎn)：各自獨(dú)立控制，不會(huì)互相影響
- 缺點(diǎn)：需要?jiǎng)?chuàng)建多個(gè) GroupId
四、完整代碼示例
4.1 依賴
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>1.8.8.1.Final</version>
</dependency>4.2 配置屬性類
package com.example.mq.config;
import java.util.Map;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Data
@Component
@ConfigurationProperties(prefix = "rocketmq")
public class RocketMqProperties {
private ProducerConfig producer;
private Map<String, ConsumerConfig> consumers;
@Data
public static class ProducerConfig {
private String accessKey;
private String secretKey;
private String endpoint;
private String tag;
private String instanceId;
}
@Data
public static class ConsumerConfig {
private String accessKey;
private String secretKey;
private String endpoint;
private String tag;
private String topic;
private String groupId;
private String protocol;
private String instanceId;
}
}4.3 Producer 配置
package com.example.mq.config;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Properties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ProducerConfiguration {
@Bean(initMethod = "start", destroyMethod = "shutdown")
public Producer producer(RocketMqProperties config) {
Properties properties = new Properties();
properties.put(PropertyKeyConst.AccessKey, config.getProducer().getAccessKey());
properties.put(PropertyKeyConst.SecretKey, config.getProducer().getSecretKey());
properties.put(PropertyKeyConst.NAMESRV_ADDR, config.getProducer().getEndpoint());
if (config.getProducer().getInstanceId() != null) {
properties.put(PropertyKeyConst.INSTANCE_ID, config.getProducer().getInstanceId());
}
return ONSFactory.createProducer(properties);
}
}
4.4 Consumer 自動(dòng)注冊(cè)（核心——支持 Map 格式多消費(fèi)者）
package com.example.mq.config;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.example.mq.listener.MqListener;
import jakarta.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import jakarta.annotation.PostConstruct;
@Slf4j
@Configuration
public class ConsumerConfiguration {
@Autowired
private RocketMqProperties config;
@Autowired
private List<MqListener> listeners;
private List<Consumer> consumers = new ArrayList<>();
@PostConstruct
public void init() {
if (config.getConsumers() == null) return;
// 遍歷 Map 格式的消費(fèi)者配置
for (Map.Entry<String, RocketMqProperties.ConsumerConfig> entry
: config.getConsumers().entrySet()) {
String name = entry.getKey();
RocketMqProperties.ConsumerConfig cc = entry.getValue();
Properties properties = new Properties();
properties.put(PropertyKeyConst.AccessKey, cc.getAccessKey());
properties.put(PropertyKeyConst.SecretKey, cc.getSecretKey());
properties.put(PropertyKeyConst.NAMESRV_ADDR, cc.getEndpoint());
properties.put(PropertyKeyConst.GROUP_ID, cc.getGroupId());
if (cc.getInstanceId() != null) {
properties.put(PropertyKeyConst.INSTANCE_ID, cc.getInstanceId());
}
Consumer consumer = ONSFactory.createConsumer(properties);
// 訂閱 Topic（支持 | 分隔多個(gè)）
String[] topics = cc.getTopic().split("\\|");
for (String topic : topics) {
// 找到負(fù)責(zé)這個(gè) Topic 的 Listener
MqListener listener = findListener(topic.trim());
if (listener != null) {
consumer.subscribe(topic.trim(), cc.getTag(), (message, context) -> {
try {
listener.consume(message.getBody());
return com.aliyun.openservices.ons.api.Action.CommitMessage;
} catch (Exception e) {
log.error("消費(fèi)異常, topic={}, msgId={}", topic, message.getMsgID(), e);
return com.aliyun.openservices.ons.api.Action.ReconsumeLater;
}
});
log.info("消費(fèi)者[{}]訂閱 topic={}, tag={}, groupId={}",
name, topic.trim(), cc.getTag(), cc.getGroupId());
}
}
consumer.start();
consumers.add(consumer);
log.info("消費(fèi)者[{}]啟動(dòng)成功", name);
}
}
@PreDestroy
public void destroy() {
for (Consumer consumer : consumers) {
consumer.shutdown();
}
}
private MqListener findListener(String topic) {
for (MqListener listener : listeners) {
if (topic.equals(listener.topic())) {
return listener;
}
}
return null;
}
}
4.5 Listener 接口
package com.example.mq.listener;
/**
* 消息監(jiān)聽(tīng)器接口.
* 每個(gè)業(yè)務(wù)實(shí)現(xiàn)此接口，通過(guò) topic() 方法告訴框架自己負(fù)責(zé)哪個(gè) Topic.
*/
public interface MqListener {
/** 消費(fèi)消息. */
void consume(byte[] body);
/** 返回負(fù)責(zé)的 Topic 名稱. */
String topic();
}
4.6 具體 Listener 實(shí)現(xiàn)
package com.example.mq.listener;
import com.example.entity.TaskMessage;
import com.example.handler.TaskHandler;
import com.example.handler.TaskHandlerRegistry;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class ImportCheckListener implements MqListener {
@Autowired
private TaskHandlerRegistry registry;
@Value("${xxx.mq.import.check}")
private String topic;
private final ObjectMapper objectMapper = new ObjectMapper();
@Override
public void consume(byte[] body) {
try {
TaskMessage task = objectMapper.readValue(body, TaskMessage.class);
log.info("校驗(yàn)消費(fèi)開(kāi)始, bizType={}, id={}", task.getBizType(), task.getId());
TaskHandler handler = registry.getHandler(task.getBizType());
handler.check(task);
} catch (Exception e) {
throw new RuntimeException("校驗(yàn)消費(fèi)異常", e);
}
}
@Override
public String topic() {
return topic;
}
}
package com.example.mq.listener;
import com.example.entity.TaskMessage;
import com.example.handler.TaskHandlerRegistry;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class ImportSaveListener implements MqListener {
@Autowired
private TaskHandlerRegistry registry;
@Value("${xxx.mq.import.save}")
private String topic;
private final ObjectMapper objectMapper = new ObjectMapper();
@Override
public void consume(byte[] body) {
try {
TaskMessage task = objectMapper.readValue(body, TaskMessage.class);
log.info("導(dǎo)入消費(fèi)開(kāi)始, bizType={}, id={}", task.getBizType(), task.getId());
registry.getHandler(task.getBizType()).save(task);
} catch (Exception e) {
throw new RuntimeException("導(dǎo)入消費(fèi)異常", e);
}
}
@Override
public String topic() {
return topic;
}
}
4.7 Producer 發(fā)送工具
package com.example.mq;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.example.entity.TaskMessage;
import com.example.mq.config.RocketMqProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class TaskMqProducer {
@Autowired
private Producer producer;
@Autowired
private RocketMqProperties config;
@Value("${xxx.mq.import.check}")
private String checkTopic;
@Value("${xxx.mq.import.save}")
private String saveTopic;
private final ObjectMapper objectMapper = new ObjectMapper();
public void sendCheck(TaskMessage task) {
send(checkTopic, task);
}
public void sendConfirm(TaskMessage task) {
send(saveTopic, task);
}
private void send(String topic, TaskMessage task) {
try {
byte[] body = objectMapper.writeValueAsBytes(task);
Message message = new Message(topic, config.getProducer().getTag(), body);
message.setKey(String.valueOf(task.getId()));
producer.send(message);
log.info("MQ發(fā)送成功, topic={}, id={}", topic, task.getId());
} catch (Exception e) {
throw new RuntimeException("MQ發(fā)送失敗", e);
}
}
}
4.8 業(yè)務(wù)處理器（路由目標(biāo)）
package com.example.handler;
import com.example.entity.TaskMessage;
public interface TaskHandler {
String getBizType();
void check(TaskMessage task);
void save(TaskMessage task);
}
package com.example.handler;
import jakarta.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class TaskHandlerRegistry {
@Autowired
private List<TaskHandler> allHandlers;
private Map<String, TaskHandler> map = new HashMap<>();
@PostConstruct
public void init() {
for (TaskHandler h : allHandlers) {
map.put(h.getBizType(), h);
}
}
public TaskHandler getHandler(String bizType) {
TaskHandler h = map.get(bizType);
if (h == null) throw new RuntimeException("未找到處理器: " + bizType);
return h;
}
}
package com.example.handler.impl;
import com.example.entity.TaskMessage;
import com.example.handler.TaskHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class OrderHandler implements TaskHandler {
@Override
public String getBizType() { return "ORDER"; }
@Override
public void check(TaskMessage task) {
log.info("訂單校驗(yàn), id={}", task.getId());
}
@Override
public void save(TaskMessage task) {
log.info("訂單入庫(kù), id={}", task.getId());
}
}
五、常見(jiàn)問(wèn)題排查
| 錯(cuò)誤 | 原因 | 解決 |
|---|---|---|
| No route info of this topic | Topic 未創(chuàng)建/endpoint 不匹配 | 確認(rèn)阿里云上 Topic 存在且 instanceId 正確 |
| subscription not consistent | 同一 GroupId 的多個(gè)實(shí)例訂閱了不同 Topic | 確保所有實(shí)例代碼一致，或使用獨(dú)立 GroupId |
| 消費(fèi)者零消費(fèi) | Consumer 未注冊(cè)成功 | 檢查啟動(dòng)日志中是否有"訂閱成功"，確認(rèn) GroupId 已創(chuàng)建 |
| AUTH_FAILED | AK/SK 錯(cuò)誤或無(wú)權(quán)限 | 確認(rèn) RAM 用戶有 MQ 權(quán)限 |
@Value 注入失敗 | yml 格式和代碼路徑不匹配 | 數(shù)組用 [0]，Map 用 .key |
以上就是SpringBoot接入阿里云RocketMQ的完整指南的詳細(xì)內(nèi)容，更多關(guān)于SpringBoot接入阿里云RocketMQ的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
JFormDesigner是一款用于設(shè)計(jì)和創(chuàng)建圖形用戶界面的插件,本文主要介紹了IDEA中的JFormDesigner使用小結(jié),具有一定的參考價(jià)值,感興趣的可以了解一下2024-01-01
SpringData如何通過(guò)@Query注解支持JPA語(yǔ)句和原生SQL語(yǔ)句
這篇文章主要介紹了SpringData如何通過(guò)@Query注解支持JPA語(yǔ)句和原生SQL語(yǔ)句，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-11-11
JAVA+MySQL實(shí)現(xiàn)分庫(kù)分表的項(xiàng)目實(shí)踐
本文主要介紹了JAVA+MySQL實(shí)現(xiàn)分庫(kù)分表的項(xiàng)目實(shí)踐,包括水平分表、垂直分表和水平分庫(kù)等策略,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2025-05-05
這篇文章主要介紹了Spring中的依賴注入DI詳解,組件之間依賴關(guān)系由容器在運(yùn)行期決定,形象的說(shuō),即由容器動(dòng)態(tài)的將依賴關(guān)系注入到組件之中,依賴注入的目的并非為軟件系統(tǒng)帶來(lái)更多功能,是為了提升組件重用的頻率,并為系統(tǒng)搭建一個(gè)靈活、可擴(kuò)展的平臺(tái),需要的朋友可以參考下2024-01-01
Spring Boot 4 與 Spring Framework&nb
SpringBoot4和SpringFramework7帶來(lái)了大量新特性,包括升級(jí)到JakartaEE11、支持Java21、增強(qiáng)模塊化、優(yōu)化測(cè)試、增加彈性功能等,本文介紹Spring Boot 4 與 Spring Framework 7新特性、升級(jí)要點(diǎn)與實(shí)戰(zhàn)指南,感興趣的朋友跟隨小編一起看看吧2026-03-03
Java OOM 異常場(chǎng)景與排查過(guò)程(堆、棧、方法區(qū))
這篇文章主要介紹了Java OOM 異常場(chǎng)景與排查過(guò)程(堆、棧、方法區(qū)),具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-03-03
Java枚舉類的規(guī)范設(shè)計(jì)與常見(jiàn)錯(cuò)誤規(guī)避指南
在Java開(kāi)發(fā)中,枚舉（enum）是一種強(qiáng)大的工具,用于定義一組固定常量集合,然而,許多開(kāi)發(fā)者在使用枚舉時(shí)容易陷入設(shè)計(jì)誤區(qū),導(dǎo)致代碼可維護(hù)性差、運(yùn)行時(shí)錯(cuò)誤頻發(fā),甚至引發(fā)生產(chǎn)事故,所以本文給大家介紹了Java枚舉類的規(guī)范設(shè)計(jì)與常見(jiàn)錯(cuò)誤規(guī)避,需要的朋友可以參考下2025-06-06
JCrontab簡(jiǎn)單入門(mén)實(shí)例詳解
這篇文章主要為大家詳細(xì)介紹了JCrontab簡(jiǎn)單入門(mén)實(shí)例，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2017-12-12
使用hibernate和struts2實(shí)現(xiàn)分頁(yè)功能的示例
本篇文章主要介紹了使用hibernate和struts2實(shí)現(xiàn)分頁(yè)功能，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-01-01
這篇文章主要介紹了Java自動(dòng)拆箱空指針異常的解決，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03