RocketMQ在Spring Boot上的基礎(chǔ)使用

更新時(shí)間：2026年03月03日 08:27:14 作者：造輪子的豬

本文主要介紹了RocketMQ在Spring Boot上的基礎(chǔ)使用,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

一、整合前置準(zhǔn)備

1. 環(huán)境依賴

Spring Boot 版本：2.x 或 3.x
RocketMQ 版本：4.9.x 或 5.x
JDK 版本：1.8 及以上

2. 引入 Maven 依賴

在 pom.xml 中添加 RocketMQ 官方 Spring Boot Starter 依賴：

<!-- RocketMQ Spring Boot Starter 核心依賴 -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version> <!-- 與 RocketMQ 服務(wù)端版本匹配 -->
</dependency>

<!-- 可選：測試依賴 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

版本匹配建議：
RocketMQ 4.9.x → starter 2.2.3
RocketMQ 5.x → starter 2.3.0+

二、核心配置

1. 基礎(chǔ)配置（application.yml）

spring:
  application:
    name: rocketmq-spring-boot-demo

# RocketMQ 核心配置
rocketmq:
  # NameServer 地址（集群用分號分隔）
  name-server: 127.0.0.1:9876
  # 生產(chǎn)者配置
  producer:
    # 生產(chǎn)者組名（必填，建議按業(yè)務(wù)命名）
    group: demo-producer-group
    # 發(fā)送超時(shí)時(shí)間（默認(rèn) 3000ms）
    send-message-timeout: 3000
    # 同步發(fā)送失敗重試次數(shù)（默認(rèn) 2）
    retry-times-when-send-failed: 2
    # 異步發(fā)送失敗重試次數(shù)（默認(rèn) 2）
    retry-times-when-send-async-failed: 2
    # 消息最大長度（默認(rèn) 4194304 字節(jié) = 4MB）
    max-message-size: 4194304
    # 壓縮閾值（默認(rèn) 4096 字節(jié)，超過自動(dòng)壓縮）
    compress-message-body-threshold: 4096
  # 消費(fèi)者配置（全局默認(rèn)，可在消費(fèi)端注解覆蓋）
  consumer:
    group: demo-consumer-group
    # 消費(fèi)線程數(shù)（默認(rèn) 20）
    consume-thread-min: 10
    consume-thread-max: 20
    # 批量消費(fèi)最大條數(shù)（默認(rèn) 1）
    consume-message-batch-max-size: 1
    # 最大重試次數(shù)（默認(rèn) -1 表示 16 次）
    max-reconsume-times: 3

2. 核心配置項(xiàng)說明

配置項(xiàng)	作用	生產(chǎn)建議
name-server	NameServer 地址	集群環(huán)境配置多個(gè)（用分號分隔），避免單點(diǎn)
producer.group	生產(chǎn)者組名	按業(yè)務(wù)劃分（如 order-producer-group）
producer.send-message-timeout	發(fā)送超時(shí)	核心業(yè)務(wù)設(shè)為 5000ms，避免超時(shí)過短
consumer.group	消費(fèi)者組名	一個(gè)業(yè)務(wù)邏輯對應(yīng)一個(gè)組，不可復(fù)用
consumer.max-reconsume-times	最大重試次數(shù)	非核心業(yè)務(wù)設(shè)為 3 次，核心業(yè)務(wù)設(shè)為 5 次

三、各消息類型代碼示例

1. 普通消息（最基礎(chǔ)）

1.1 生產(chǎn)者（同步發(fā)送）

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

@Component
public class NormalMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步發(fā)送普通消息
     * @param topic 消息主題
     * @param msgContent 消息內(nèi)容
     * @param msgKey 消息唯一鍵（用于追蹤/冪等）
     */
    public void sendNormalMessage(String topic, String msgContent, String msgKey) {
        // 構(gòu)建消息（支持自定義 Header）
        Message<String> message = MessageBuilder
                .withPayload(msgContent)
                // 設(shè)置消息 Key（必填，用于冪等/追蹤）
                .setHeader(RocketMQHeaders.KEYS, msgKey)
                // 可選：設(shè)置 Tag
                .setHeader(RocketMQHeaders.TAGS, "normal_tag")
                .build();

        // 同步發(fā)送（topic:tag 格式指定 Tag）
        rocketMQTemplate.syncSend(topic + ":normal_tag", message);
        System.out.println("普通消息發(fā)送成功：" + msgKey);
    }
}

1.2 消費(fèi)者

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * 普通消息消費(fèi)者
 * - consumerGroup：消費(fèi)者組（必填）
 * - topic：訂閱主題（必填）
 * - selectorExpression：Tag 過濾（* 表示所有）
 * - messageModel：消費(fèi)模式（CLUSTERING 集群/ BROADCASTING 廣播）
 * - consumeMode：消費(fèi)模式（CONCURRENTLY 并發(fā)/ ORDERLY 順序）
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "demo-consumer-group",
        topic = "normal_topic",
        selectorExpression = "normal_tag",
        messageModel = MessageModel.CLUSTERING,
        consumeMode = ConsumeMode.CONCURRENTLY
)
public class NormalMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        // 消費(fèi)邏輯（需保證冪等）
        System.out.println("收到普通消息：" + msg);
        // 業(yè)務(wù)處理...
    }
}

2. 順序消息

2.1 生產(chǎn)者

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class OrderedMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 發(fā)送順序消息（按業(yè)務(wù)鍵哈希選擇隊(duì)列）
     * @param topic 主題
     * @param msgContent 內(nèi)容
     * @param orderId 業(yè)務(wù)唯一鍵（如訂單ID，保證同ID入同隊(duì)列）
     */
    public void sendOrderedMessage(String topic, String msgContent, String orderId) {
        // 構(gòu)建消息
        String message = "訂單" + orderId + "：" + msgContent;
        
        // 發(fā)送順序消息（指定 hashKey 為 orderId）
        rocketMQTemplate.syncSendOrderly(
                topic + ":order_tag", 
                MessageBuilder.withPayload(message).build(),
                orderId // 關(guān)鍵：hashKey，保證同值入同隊(duì)列
        );
        System.out.println("順序消息發(fā)送成功：" + orderId);
    }
}

2.2 消費(fèi)者（必須設(shè)為順序消費(fèi)）

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

@Component
@RocketMQMessageListener(
        consumerGroup = "order-consumer-group",
        topic = "order_topic",
        selectorExpression = "order_tag",
        // 核心：順序消費(fèi)必須設(shè)為 ORDERLY
        consumeMode = ConsumeMode.ORDERLY
)
public class OrderedMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        // 單線程消費(fèi)，保證順序
        System.out.println("收到順序消息：" + msg);
        // 業(yè)務(wù)處理（如訂單創(chuàng)建→支付→發(fā)貨）
    }
}

3. 延遲消息

3.1 生產(chǎn)者

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class DelayMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 發(fā)送延遲消息（固定等級）
     * @param topic 主題
     * @param msgContent 內(nèi)容
     * @param delayLevel 延遲等級（1=1s，5=1m，18=2h）
     */
    public void sendDelayMessage(String topic, String msgContent, int delayLevel) {
        // 構(gòu)建消息
        org.apache.rocketmq.common.message.Message rocketMsg = new org.apache.rocketmq.common.message.Message(
                topic,
                "delay_tag",
                msgContent.getBytes()
        );
        // 設(shè)置延遲等級
        rocketMsg.setDelayTimeLevel(delayLevel);

        // 發(fā)送延遲消息
        rocketMQTemplate.getProducer().send(rocketMsg);
        System.out.println("延遲消息發(fā)送成功（等級" + delayLevel + "）：" + msgContent);
    }
}

3.2 消費(fèi)者（與普通消息一致）

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

@Component
@RocketMQMessageListener(
        consumerGroup = "delay-consumer-group",
        topic = "delay_topic",
        selectorExpression = "delay_tag"
)
public class DelayMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        System.out.println("收到延遲消息：" + msg);
        // 業(yè)務(wù)處理（如訂單超時(shí)取消）
    }
}

4. 事務(wù)消息

4.1 事務(wù)監(jiān)聽器（核心）

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

/**
 * 事務(wù)消息監(jiān)聽器
 * - txProducerGroup：對應(yīng)生產(chǎn)者組名
 */
@RocketMQTransactionListener(txProducerGroup = "tx-producer-group")
@Component
public class TransactionListener implements RocketMQLocalTransactionListener {

    /**
     * 執(zhí)行本地事務(wù)
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 獲取消息內(nèi)容
        String msgContent = new String((byte[]) msg.getPayload());
        String orderId = msg.getHeaders().get("KEYS").toString();
        
        try {
            // 執(zhí)行本地事務(wù)（如扣減庫存、創(chuàng)建訂單）
            boolean success = executeLocalDBTransaction(orderId);
            if (success) {
                // 提交消息
                return RocketMQLocalTransactionState.COMMIT;
            } else {
                // 回滾消息
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            // 未知狀態(tài)，等待回查
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    /**
     * 事務(wù)回查（Broker 主動(dòng)調(diào)用）
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String orderId = msg.getHeaders().get("KEYS").toString();
        // 查詢本地事務(wù)狀態(tài)
        boolean isSuccess = queryLocalTransactionStatus(orderId);
        return isSuccess ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
    }

    // 模擬本地事務(wù)執(zhí)行
    private boolean executeLocalDBTransaction(String orderId) {
        // 實(shí)際業(yè)務(wù)邏輯：操作數(shù)據(jù)庫/緩存等
        return true;
    }

    // 模擬查詢本地事務(wù)狀態(tài)
    private boolean queryLocalTransactionStatus(String orderId) {
        // 實(shí)際查詢數(shù)據(jù)庫狀態(tài)
        return true;
    }
}

4.2 事務(wù)消息生產(chǎn)者

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class TransactionMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 發(fā)送事務(wù)消息
     * @param topic 主題
     * @param msgContent 內(nèi)容
     * @param orderId 訂單ID（消息Key）
     */
    public void sendTransactionMessage(String topic, String msgContent, String orderId) {
        // 構(gòu)建消息
        Message<String> message = MessageBuilder
                .withPayload(msgContent)
                .setHeader("KEYS", orderId)
                .setHeader("TAGS", "tx_tag")
                .build();

        // 發(fā)送事務(wù)消息（最后一個(gè)參數(shù)為 arg，會傳給監(jiān)聽器）
        rocketMQTemplate.sendMessageInTransaction(
                topic + ":tx_tag",
                message,
                null // 自定義參數(shù)，可傳業(yè)務(wù)對象
        );
        System.out.println("事務(wù)消息發(fā)送成功（半消息）：" + orderId);
    }
}

4.3 事務(wù)消息消費(fèi)者（與普通消息一致）

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

@Component
@RocketMQMessageListener(
        consumerGroup = "tx-consumer-group",
        topic = "tx_topic",
        selectorExpression = "tx_tag"
)
public class TransactionMessageConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        System.out.println("收到事務(wù)消息：" + msg);
        // 業(yè)務(wù)處理（如通知物流、更新積分）
    }
}

5. 批量消息

5.1 生產(chǎn)者

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Component
public class BatchMessageProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 發(fā)送批量消息
     * @param topic 主題
     */
    public void sendBatchMessage(String topic) {
        // 構(gòu)建批量消息列表（必須同Topic、同Tag、無延遲/事務(wù)）
        List<Message> msgList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message msg = new Message(
                    topic,
                    "batch_tag",
                    ("批量消息" + i).getBytes()
            );
            msgList.add(msg);
        }

        // 發(fā)送批量消息
        SendResult result = rocketMQTemplate.getProducer().send(msgList);
        System.out.println("批量消息發(fā)送成功：" + result.getMsgId());
    }
}

5.2 批量消息消費(fèi)者（支持批量消費(fèi)）

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQBatchListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@RocketMQMessageListener(
        consumerGroup = "batch-consumer-group",
        topic = "batch_topic",
        selectorExpression = "batch_tag",
        // 批量消費(fèi)最大條數(shù)（需與配置項(xiàng)一致）
        consumeMessageBatchMaxSize = 10
)
public class BatchMessageConsumer implements RocketMQBatchListener<String> {

    @Override
    public void onMessage(List<String> msgs) {
        // 批量處理消息
        System.out.println("收到批量消息，共" + msgs.size() + "條：");
        msgs.forEach(msg -> System.out.println("- " + msg));
    }
}

四、測試用例（完整示例）

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;

@SpringBootTest
public class RocketMQTest {

    @Resource
    private NormalMessageProducer normalMessageProducer;

    @Resource
    private OrderedMessageProducer orderedMessageProducer;

    @Resource
    private DelayMessageProducer delayMessageProducer;

    @Resource
    private TransactionMessageProducer transactionMessageProducer;

    @Resource
    private BatchMessageProducer batchMessageProducer;

    // 測試普通消息
    @Test
    public void testNormalMessage() {
        normalMessageProducer.sendNormalMessage("normal_topic", "Hello RocketMQ", "MSG_001");
    }

    // 測試順序消息
    @Test
    public void testOrderedMessage() {
        String orderId = "ORDER_1001";
        orderedMessageProducer.sendOrderedMessage("order_topic", "創(chuàng)建", orderId);
        orderedMessageProducer.sendOrderedMessage("order_topic", "支付", orderId);
        orderedMessageProducer.sendOrderedMessage("order_topic", "發(fā)貨", orderId);
    }

    // 測試延遲消息（等級5=1分鐘）
    @Test
    public void testDelayMessage() {
        delayMessageProducer.sendDelayMessage("delay_topic", "訂單超時(shí)取消", 5);
    }

    // 測試事務(wù)消息
    @Test
    public void testTransactionMessage() {
        transactionMessageProducer.sendTransactionMessage("tx_topic", "訂單支付成功", "ORDER_10086");
    }

    // 測試批量消息
    @Test
    public void testBatchMessage() {
        batchMessageProducer.sendBatchMessage("batch_topic");
    }
}

五、生產(chǎn)級優(yōu)化建議

1. 冪等性保障

消費(fèi)端必須基于 msgKey 做冪等（如 Redis 分布式鎖、數(shù)據(jù)庫唯一鍵）；
避免重復(fù)消費(fèi)導(dǎo)致業(yè)務(wù)異常。

2. 異常處理

生產(chǎn)者：捕獲 MQClientException，實(shí)現(xiàn)失敗重試/降級；
消費(fèi)者：消費(fèi)失敗時(shí)拋出異常，觸發(fā)重試（或手動(dòng)返回失敗）。

3. 監(jiān)控告警

監(jiān)控消息堆積量（brokerOffset - consumerOffset）；
監(jiān)控發(fā)送/消費(fèi)失敗率、延遲時(shí)間。

4. 死信隊(duì)列處理

配置死信隊(duì)列（默認(rèn) %DLQ%{consumerGroup}）；
定期處理死信消息，避免消息丟失。

總結(jié)

核心關(guān)鍵點(diǎn)

配置核心：name-server 和 producer/consumer.group 是必填項(xiàng)，需按業(yè)務(wù)規(guī)范命名；
消息類型：
- 普通消息：直接發(fā)送，適配大部分場景；
- 順序消息：需指定 hashKey，消費(fèi)端設(shè)為 ORDERLY；
- 延遲消息：僅支持固定等級，需設(shè)置 delayTimeLevel；
- 事務(wù)消息：需實(shí)現(xiàn) RocketMQLocalTransactionListener，處理本地事務(wù)和回查；
- 批量消息：需保證同Topic/Tag，消費(fèi)端實(shí)現(xiàn) RocketMQBatchListener；
生產(chǎn)規(guī)范：消息 Key 必設(shè)、消費(fèi)冪等必做、重試次數(shù)合理配置。

到此這篇關(guān)于RocketMQ在Spring Boot上的基礎(chǔ)使用的文章就介紹到這了,更多相關(guān)SpringBoot RocketMQ使用內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

RocketMQ在Spring Boot上的基礎(chǔ)使用

目錄

一、整合前置準(zhǔn)備

1. 環(huán)境依賴

2. 引入 Maven 依賴

二、核心配置

1. 基礎(chǔ)配置（application.yml）

2. 核心配置項(xiàng)說明

三、各消息類型代碼示例

1. 普通消息（最基礎(chǔ)）

1.1 生產(chǎn)者（同步發(fā)送）

1.2 消費(fèi)者

2. 順序消息

2.1 生產(chǎn)者

2.2 消費(fèi)者（必須設(shè)為順序消費(fèi)）

3. 延遲消息

3.1 生產(chǎn)者

3.2 消費(fèi)者（與普通消息一致）

4. 事務(wù)消息

4.1 事務(wù)監(jiān)聽器（核心）

4.2 事務(wù)消息生產(chǎn)者

4.3 事務(wù)消息消費(fèi)者（與普通消息一致）

5. 批量消息

5.1 生產(chǎn)者

5.2 批量消息消費(fèi)者（支持批量消費(fèi)）

四、測試用例（完整示例）

五、生產(chǎn)級優(yōu)化建議

1. 冪等性保障

2. 異常處理

3. 監(jiān)控告警

4. 死信隊(duì)列處理

總結(jié)

核心關(guān)鍵點(diǎn)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

RocketMQ在Spring Boot上的基礎(chǔ)使用

目錄

一、整合前置準(zhǔn)備

1. 環(huán)境依賴

2. 引入 Maven 依賴

二、核心配置

1. 基礎(chǔ)配置（application.yml）

2. 核心配置項(xiàng)說明

三、各消息類型代碼示例

1. 普通消息（最基礎(chǔ)）

1.1 生產(chǎn)者（同步發(fā)送）

1.2 消費(fèi)者

2. 順序消息

2.1 生產(chǎn)者

2.2 消費(fèi)者（必須設(shè)為順序消費(fèi)）

3. 延遲消息

3.1 生產(chǎn)者

3.2 消費(fèi)者（與普通消息一致）

4. 事務(wù)消息

4.1 事務(wù)監(jiān)聽器（核心）

4.2 事務(wù)消息生產(chǎn)者

4.3 事務(wù)消息消費(fèi)者（與普通消息一致）

5. 批量消息

5.1 生產(chǎn)者

5.2 批量消息消費(fèi)者（支持批量消費(fèi)）

四、測試用例（完整示例）

五、生產(chǎn)級優(yōu)化建議

1. 冪等性保障

2. 異常處理

3. 監(jiān)控告警

4. 死信隊(duì)列處理

總結(jié)

核心關(guān)鍵點(diǎn)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

一、整合前置準(zhǔn)備

二、核心配置

三、各消息類型代碼示例

四、測試用例（完整示例）