<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.15</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>spring-boot-rocketmq-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>spring-boot-rocketmq-demo</name>
    <description>Spring Boot RocketMQ 整合示例</description>
    <properties>
        <java.version>1.8</java.version>
        <rocketmq-spring-boot-starter.version>2.2.3</rocketmq-spring-boot-starter.version>
    </properties>
    <dependencies>
        <!-- Spring Web 依賴，用于提供接口測試 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- RocketMQ Spring Boot Starter -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>${rocketmq-spring-boot-starter.version}</version>
        </dependency>
        <!-- 測試依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

spring:
  application:
    name: spring-boot-rocketmq-demo
# RocketMQ 配置
rocketmq:
  # NameServer 地址，多個地址用分號分隔
  name-server: localhost:9876
  # 生產者配置
  producer:
    # 生產者組名，必須唯一
    group: demo-producer-group
    # 發(fā)送消息的超時時間，默認 3000ms
    send-message-timeout: 3000
    # 消息體最大長度，默認 4MB
    max-message-size: 4194304
    # 壓縮消息的閾值，默認 4KB
    compress-message-body-threshold: 4096
    # 重試次數，默認 2 次
    retry-times-when-send-failed: 2
    # 異步發(fā)送失敗時是否重試其他 Broker，默認 false
    retry-next-server: false

package com.example.demo.service;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
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.Service;
import javax.annotation.Resource;
/**
 * RocketMQ 生產者服務
 */
@Service
public class RocketMQProducerService {
    // 注入 RocketMQ 模板類
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    /**
     * 同步發(fā)送消息
     * @param topic 主題名（可攜帶標簽，格式：topic:tag）
     * @param message 消息內容
     * @return 發(fā)送結果
     */
    public SendResult sendSyncMessage(String topic, String message) {
        // 構建消息（可添加消息頭，如自定義 key）
        Message<String> msg = MessageBuilder.withPayload(message)
                .setHeader(RocketMQHeaders.KEYS, "sync-key-" + System.currentTimeMillis())
                .build();
        // 同步發(fā)送
        return rocketMQTemplate.syncSend(topic, msg);
    }
    /**
     * 異步發(fā)送消息
     * @param topic 主題名
     * @param message 消息內容
     */
    public void sendAsyncMessage(String topic, String message) {
        Message<String> msg = MessageBuilder.withPayload(message)
                .setHeader(RocketMQHeaders.KEYS, "async-key-" + System.currentTimeMillis())
                .build();
        // 異步發(fā)送，通過 SendCallback 處理回調
        rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 發(fā)送成功處理
                System.out.println("異步發(fā)送消息成功：" + sendResult);
            }
            @Override
            public void onException(Throwable e) {
                // 發(fā)送失敗處理
                System.err.println("異步發(fā)送消息失敗：" + e.getMessage());
            }
        });
    }
    /**
     * 單向發(fā)送消息（不關心發(fā)送結果）
     * @param topic 主題名
     * @param message 消息內容
     */
    public void sendOneWayMessage(String topic, String message) {
        Message<String> msg = MessageBuilder.withPayload(message)
                .setHeader(RocketMQHeaders.KEYS, "oneway-key-" + System.currentTimeMillis())
                .build();
        // 單向發(fā)送
        rocketMQTemplate.sendOneWay(topic, msg);
    }
}

package com.example.demo.controller;
import org.apache.rocketmq.client.producer.SendResult;
import com.example.demo.service.RocketMQProducerService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
 * 消息發(fā)送測試控制器
 */
@RestController
@RequestMapping("/message")
public class MessageSendController {
    @Resource
    private RocketMQProducerService rocketMQProducerService;
    // 測試同步發(fā)送
    @GetMapping("/sync/send")
    public String sendSyncMessage(@RequestParam String msg) {
        // 主題名：demo-topic，標簽：demo-tag（標簽用于消息過濾）
        String topic = "demo-topic:demo-tag";
        SendResult sendResult = rocketMQProducerService.sendSyncMessage(topic, msg);
        return "同步發(fā)送消息成功：" + sendResult;
    }
    // 測試異步發(fā)送
    @GetMapping("/async/send")
    public String sendAsyncMessage(@RequestParam String msg) {
        String topic = "demo-topic:demo-tag";
        rocketMQProducerService.sendAsyncMessage(topic, msg);
        return "異步發(fā)送消息請求已提交";
    }
    // 測試單向發(fā)送
    @GetMapping("/oneway/send")
    public String sendOneWayMessage(@RequestParam String msg) {
        String topic = "demo-topic:demo-tag";
        rocketMQProducerService.sendOneWayMessage(topic, msg);
        return "單向發(fā)送消息完成";
    }
}

package com.example.demo.consumer;
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 消費者服務
 * 注解說明：
 * - consumerGroup：消費者組名，必須唯一
 * - topic：監(jiān)聽的主題名
 * - selectorExpression：標簽表達式，* 表示所有標簽，也可指定具體標簽（如 demo-tag）
 * - messageModel：消息模式，CLUSTERING（集群模式）/BROADCASTING（廣播模式），默認集群模式
 * - consumeMode：消費模式，CONCURRENTLY（并發(fā)消費）/ORDERLY（順序消費），默認并發(fā)消費
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "demo-consumer-group",
        topic = "demo-topic",
        selectorExpression = "*",
        messageModel = MessageModel.CLUSTERING,
        consumeMode = ConsumeMode.CONCURRENTLY
)
public class RocketMQConsumerService implements RocketMQListener<String> {
    /**
     * 消息消費邏輯
     * @param message 消息內容
     */
    @Override
    public void onMessage(String message) {
        System.out.println("接收到消息：" + message);
        // 模擬業(yè)務處理
        // 注意：消費端拋出異常會觸發(fā)消息重試
        // handleBusiness(message);
    }
    /**
     * 模擬業(yè)務處理
     * @param message 消息內容
     */
    private void handleBusiness(String message) {
        // 業(yè)務邏輯代碼
    }
}

# 消費者重試配置（可在注解中覆蓋）
rocketmq:
  consumer:
    # 消費線程數
    consume-thread-max: 20
    # 批量消費的最大消息數
    consume-message-batch-max-size: 1
    # 消費超時時間
    consume-timeout: 15

package com.example.demo.consumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;
import java.util.List;
/**
 * 自定義重試邏輯的消費者
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "demo-consumer-group-retry",
        topic = "demo-topic",
        selectorExpression = "*"
)
public class RetryRocketMQConsumerService implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(MessageExt messageExt) {
        String message = new String(messageExt.getBody());
        System.out.println("接收到消息（帶重試邏輯）：" + message + "，重試次數：" + messageExt.getReconsumeTimes());
        try {
            // 模擬業(yè)務處理失敗
            int a = 1 / 0;
            // 業(yè)務處理成功，無需重試
        } catch (Exception e) {
            System.err.println("消息消費失?。? + e.getMessage());
            // 若重試次數超過 3 次，直接返回成功（不再重試），否則拋出異常觸發(fā)重試
            if (messageExt.getReconsumeTimes() >= 3) {
                System.out.println("消息重試次數已達上限，不再重試");
                // 可將消息記錄到數據庫，后續(xù)人工處理
                return;
            }
            // 拋出異常，觸發(fā)重試
            throw new RuntimeException("消費失敗，觸發(fā)重試");
        }
    }
    @Override
    public void prepareStart(org.apache.rocketmq.client.consumer.DefaultMQPushConsumer consumer) {
        // 可自定義消費者配置，如設置重試次數
        try {
            // 設置消費線程數
            consumer.setConsumeThreadMax(20);
            // 設置消息監(jiān)聽器（若需要更細粒度的控制）
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                for (MessageExt msg : msgs) {
                    onMessage(msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
    }
}

package com.example.demo.consumer;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
 * 死信隊列消費者
 * 死信隊列名稱：%DLQ% + 原消費者組名
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "dlq-consumer-group",
        topic = "%DLQ%demo-consumer-group",
        selectorExpression = "*"
)
public class DlqRocketMQConsumerService implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("接收到死信隊列消息：" + message);
        // 人工處理邏輯，如記錄日志、通知運維、手動重試等
    }
}