<?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 依賴，用于提供接口測(cè)試 -->
        <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>
        <!-- 測(cè)試依賴 -->
        <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 地址，多個(gè)地址用分號(hào)分隔
  name-server: localhost:9876
  # 生產(chǎn)者配置
  producer:
    # 生產(chǎn)者組名，必須唯一
    group: demo-producer-group
    # 發(fā)送消息的超時(shí)時(shí)間，默認(rèn) 3000ms
    send-message-timeout: 3000
    # 消息體最大長(zhǎng)度，默認(rèn) 4MB
    max-message-size: 4194304
    # 壓縮消息的閾值，默認(rèn) 4KB
    compress-message-body-threshold: 4096
    # 重試次數(shù)，默認(rèn) 2 次
    retry-times-when-send-failed: 2
    # 異步發(fā)送失敗時(shí)是否重試其他 Broker，默認(rèn) 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 生產(chǎn)者服務(wù)
 */
@Service
public class RocketMQProducerService {
    // 注入 RocketMQ 模板類
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    /**
     * 同步發(fā)送消息
     * @param topic 主題名（可攜帶標(biāo)簽，格式：topic:tag）
     * @param message 消息內(nèi)容
     * @return 發(fā)送結(jié)果
     */
    public SendResult sendSyncMessage(String topic, String message) {
        // 構(gòu)建消息（可添加消息頭，如自定義 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 消息內(nèi)容
     */
    public void sendAsyncMessage(String topic, String message) {
        Message<String> msg = MessageBuilder.withPayload(message)
                .setHeader(RocketMQHeaders.KEYS, "async-key-" + System.currentTimeMillis())
                .build();
        // 異步發(fā)送，通過 SendCallback 處理回調(diào)
        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ā)送消息（不關(guān)心發(fā)送結(jié)果）
     * @param topic 主題名
     * @param message 消息內(nèi)容
     */
    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ā)送測(cè)試控制器
 */
@RestController
@RequestMapping("/message")
public class MessageSendController {
    @Resource
    private RocketMQProducerService rocketMQProducerService;
    // 測(cè)試同步發(fā)送
    @GetMapping("/sync/send")
    public String sendSyncMessage(@RequestParam String msg) {
        // 主題名：demo-topic，標(biāo)簽：demo-tag（標(biāo)簽用于消息過濾）
        String topic = "demo-topic:demo-tag";
        SendResult sendResult = rocketMQProducerService.sendSyncMessage(topic, msg);
        return "同步發(fā)送消息成功：" + sendResult;
    }
    // 測(cè)試異步發(fā)送
    @GetMapping("/async/send")
    public String sendAsyncMessage(@RequestParam String msg) {
        String topic = "demo-topic:demo-tag";
        rocketMQProducerService.sendAsyncMessage(topic, msg);
        return "異步發(fā)送消息請(qǐng)求已提交";
    }
    // 測(cè)試單向發(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 消費(fèi)者服務(wù)
 * 注解說明：
 * - consumerGroup：消費(fèi)者組名，必須唯一
 * - topic：監(jiān)聽的主題名
 * - selectorExpression：標(biāo)簽表達(dá)式，* 表示所有標(biāo)簽，也可指定具體標(biāo)簽（如 demo-tag）
 * - messageModel：消息模式，CLUSTERING（集群模式）/BROADCASTING（廣播模式），默認(rèn)集群模式
 * - consumeMode：消費(fèi)模式，CONCURRENTLY（并發(fā)消費(fèi)）/ORDERLY（順序消費(fèi)），默認(rèn)并發(fā)消費(fèi)
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "demo-consumer-group",
        topic = "demo-topic",
        selectorExpression = "*",
        messageModel = MessageModel.CLUSTERING,
        consumeMode = ConsumeMode.CONCURRENTLY
)
public class RocketMQConsumerService implements RocketMQListener<String> {
    /**
     * 消息消費(fèi)邏輯
     * @param message 消息內(nèi)容
     */
    @Override
    public void onMessage(String message) {
        System.out.println("接收到消息：" + message);
        // 模擬業(yè)務(wù)處理
        // 注意：消費(fèi)端拋出異常會(huì)觸發(fā)消息重試
        // handleBusiness(message);
    }
    /**
     * 模擬業(yè)務(wù)處理
     * @param message 消息內(nèi)容
     */
    private void handleBusiness(String message) {
        // 業(yè)務(wù)邏輯代碼
    }
}

# 消費(fèi)者重試配置（可在注解中覆蓋）
rocketmq:
  consumer:
    # 消費(fèi)線程數(shù)
    consume-thread-max: 20
    # 批量消費(fèi)的最大消息數(shù)
    consume-message-batch-max-size: 1
    # 消費(fèi)超時(shí)時(shí)間
    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;
/**
 * 自定義重試邏輯的消費(fèi)者
 */
@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 + "，重試次數(shù)：" + messageExt.getReconsumeTimes());
        try {
            // 模擬業(yè)務(wù)處理失敗
            int a = 1 / 0;
            // 業(yè)務(wù)處理成功，無需重試
        } catch (Exception e) {
            System.err.println("消息消費(fèi)失敗：" + e.getMessage());
            // 若重試次數(shù)超過 3 次，直接返回成功（不再重試），否則拋出異常觸發(fā)重試
            if (messageExt.getReconsumeTimes() >= 3) {
                System.out.println("消息重試次數(shù)已達(dá)上限，不再重試");
                // 可將消息記錄到數(shù)據(jù)庫(kù)，后續(xù)人工處理
                return;
            }
            // 拋出異常，觸發(fā)重試
            throw new RuntimeException("消費(fèi)失敗，觸發(fā)重試");
        }
    }
    @Override
    public void prepareStart(org.apache.rocketmq.client.consumer.DefaultMQPushConsumer consumer) {
        // 可自定義消費(fèi)者配置，如設(shè)置重試次數(shù)
        try {
            // 設(shè)置消費(fèi)線程數(shù)
            consumer.setConsumeThreadMax(20);
            // 設(shè)置消息監(jiān)聽器（若需要更細(xì)粒度的控制）
            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;
/**
 * 死信隊(duì)列消費(fèi)者
 * 死信隊(duì)列名稱：%DLQ% + 原消費(fèi)者組名
 */
@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("接收到死信隊(duì)列消息：" + message);
        // 人工處理邏輯，如記錄日志、通知運(yùn)維、手動(dòng)重試等
    }
}