<!-- Spring Boot 基礎(chǔ)依賴 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- RocketMQ Spring Boot Starter（官方推薦） -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version> <!-- 適配 RocketMQ 5.x，兼容 Spring Boot 2/3 -->
</dependency>
<!-- 可選：測(cè)試依賴 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

spring:
  application:
    name: rocketmq-demo # 應(yīng)用名稱
# RocketMQ 核心配置
rocketmq:
  name-server: 127.0.0.1:9876 # NameServer 地址（集群用逗號(hào)分隔）
  producer:
    group: demo-producer-group # 生產(chǎn)者組（必填，標(biāo)識(shí)同一類生產(chǎn)者）
    send-message-timeout: 3000 # 發(fā)送超時(shí)時(shí)間，默認(rèn)3000ms
    compress-message-body-threshold: 4096 # 消息壓縮閾值，默認(rèn)4096字節(jié)
    max-message-size: 4194304 # 最大消息大小，默認(rèn)4MB
    retry-times-when-send-failed: 2 # 同步發(fā)送失敗重試次數(shù)
    retry-times-when-send-async-failed: 2 # 異步發(fā)送失敗重試次數(shù)

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class RocketMQProducer {
    // 注入官方封裝的 RocketMQ 模板（類似 RabbitTemplate）
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    /**
     * 同步發(fā)送消息（阻塞等待結(jié)果）
     * @param topic 消息主題（必填，需提前創(chuàng)建）
     * @param message 消息內(nèi)容
     * @return 發(fā)送結(jié)果
     */
    public SendResult sendSyncMessage(String topic, String message) {
        try {
            // 發(fā)送格式："topic:tag"（tag可選，用于消息過(guò)濾）
            SendResult sendResult = rocketMQTemplate.syncSend(topic + ":demoTag", message);
            System.out.println("同步發(fā)送成功，消息ID：" + sendResult.getMsgId());
            return sendResult;
        } catch (Exception e) {
            System.err.println("同步發(fā)送失?。? + e.getMessage());
            // 業(yè)務(wù)異常處理（如重試、記錄日志、告警）
            throw new RuntimeException("消息發(fā)送失敗", e);
        }
    }
    /**
     * 異步發(fā)送消息（非阻塞，回調(diào)通知結(jié)果）
     * @param topic 主題
     * @param message 消息內(nèi)容
     */
    public void sendAsyncMessage(String topic, String message) {
        rocketMQTemplate.asyncSend(
            topic + ":demoTag",
            message,
            // 發(fā)送成功回調(diào)
            sendResult -> System.out.println("異步發(fā)送成功，消息ID：" + sendResult.getMsgId()),
            // 發(fā)送失敗回調(diào)
            throwable -> System.err.println("異步發(fā)送失?。? + throwable.getMessage())
        );
    }
    /**
     * 單向發(fā)送消息（無(wú)回調(diào)，適用于日志、埋點(diǎn)等不關(guān)心結(jié)果的場(chǎng)景）
     * @param topic 主題
     * @param message 消息內(nèi)容
     */
    public void sendOneWayMessage(String topic, String message) {
        rocketMQTemplate.sendOneWay(topic + ":demoTag", message);
        System.out.println("單向消息發(fā)送請(qǐng)求已提交");
    }
}

// 自定義消息實(shí)體（實(shí)現(xiàn) Serializable）
public class OrderMessage implements Serializable {
    private Long orderId;
    private String orderNo;
    private BigDecimal amount;
    // 省略 getter/setter/toString
}
// 生產(chǎn)者中新增方法
public SendResult sendObjectMessage(String topic, OrderMessage orderMessage) {
    return rocketMQTemplate.syncSend(topic + ":orderTag", orderMessage);
}

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;
/**
 * RocketMQ 消費(fèi)者
 * - topic：訂閱的主題（需與生產(chǎn)者一致）
 * - consumerGroup：消費(fèi)者組（必填，同一組消費(fèi)同一主題）
 * - messageModel：消費(fèi)模式（CLUSTERING 集群模式，BROADCASTING 廣播模式）
 * - consumeMode：消費(fèi)方式（CONCURRENTLY 并發(fā)消費(fèi)，ORDERLY 順序消費(fèi)）
 */
@Component
@RocketMQMessageListener(
    topic = "demo_topic", // 訂閱主題
    consumerGroup = "demo-consumer-group", // 消費(fèi)者組
    messageModel = MessageModel.CLUSTERING, // 集群模式（默認(rèn)）
    consumeMode = ConsumeMode.CONCURRENTLY // 并發(fā)消費(fèi)（默認(rèn)）
)
public class RocketMQConsumer implements RocketMQListener<String> {
    /**
     * 消息消費(fèi)邏輯（接收到消息時(shí)觸發(fā)）
     * @param message 消息內(nèi)容（與生產(chǎn)者發(fā)送類型一致）
     */
    @Override
    public void onMessage(String message) {
        try {
            // 核心業(yè)務(wù)邏輯：如解析消息、處理訂單、更新庫(kù)存等
            System.out.println("接收到消息：" + message);
            // 消費(fèi)成功無(wú)需返回，拋出異常則會(huì)觸發(fā)重試
        } catch (Exception e) {
            System.err.println("消息消費(fèi)失?。? + e.getMessage());
            // 異常拋出后，RocketMQ 會(huì)自動(dòng)重試（默認(rèn)最多16次）
            throw new RuntimeException("消費(fèi)失敗", e);
        }
    }
}

@Component
@RocketMQMessageListener(
    topic = "demo_topic",
    consumerGroup = "order-consumer-group"
)
public class OrderMessageConsumer implements RocketMQListener<OrderMessage> {
    @Override
    public void onMessage(OrderMessage orderMessage) {
        System.out.println("接收到訂單消息：" + orderMessage);
        // 處理訂單業(yè)務(wù)邏輯
    }
}

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
public class RocketMQDemoTest {
    @Autowired
    private RocketMQProducer rocketMQProducer;
    @Test
    public void testSendSyncMessage() {
        // 發(fā)送消息到 demo_topic 主題
        rocketMQProducer.sendSyncMessage("demo_topic", "Hello RocketMQ + Spring Boot!");
        // 暫停3秒，確保消費(fèi)者能接收到消息
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}