SpringBoot 整合 RocketMQ 实现消息生产与消费的全过程

# 拉取 RocketMQ 镜像（官方推荐版本，稳定兼容）
docker pull rocketmqinc/rocketmq:4.9.4
# 启动容器（映射端口，设置环境变量，简化配置）
docker run -d --name rocketmq -p 9876:9876 -p 10911:10911 -e "ROCKETMQ_NAMESRV_ADDR=localhost:9876" -e "ROCKETMQ_BROKER_ADDR=localhost:10911" rocketmqinc/rocketmq:4.9.4

<?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.18</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>springboot-rocketmq-intro</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <properties>
        <java.version>1.8</java.version>
        <rocketmq.version>4.9.4</rocketmq.version>
    </properties>
    <dependencies>
        <!-- SpringBoot Web 依赖（用于编写接口测试生产者） -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- RocketMQ 整合依赖（核心） -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>${rocketmq.version}</version>
        </dependency>
        <!-- Lombok 简化代码（可选，与全系列风格统一） -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>
</project>

spring:
  application:
    name: springboot-rocketmq-intro # 项目名称，可自定义
# RocketMQ 核心配置
rocketmq:
  name-server: localhost:9876 # NameServer 地址（本地默认，服务器填对应IP）
  producer:
    group: rocketmq-producer-group # 生产者组（自定义，同一组生产者共享配置）
    send-message-timeout: 3000 # 消息发送超时时间（毫秒）
    retry-times-when-send-failed: 1 # 发送失败重试次数（入门设1，避免重复发送）
  consumer:
    group: rocketmq-consumer-group # 消费者组（自定义，同一组消费者负载均衡消费）
    message-model: CLUSTERING # 消费模式：CLUSTERING（集群模式，默认）、BROADCASTING（广播模式）
    consume-thread-max: 10 # 最大消费线程数（入门无需修改）
    consume-timeout: 15000 # 消费超时时间（毫秒）

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
/**
 * RocketMQ 消息生产者
 * 发送普通消息到指定主题
 */
@RestController
public class RocketMQProducer {
    // 注入 RocketMQTemplate（SpringBoot 自动配置，直接使用）
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    // 自定义主题名称（可自定义，消费者需订阅相同主题）
    private static final String TOPIC_NAME = "springboot_rocketmq_topic";
    /**
     * 发送普通消息接口（可通过浏览器访问测试）
     * @param msg 要发送的消息内容（可自定义）
     * @return 发送结果
     */
    @GetMapping("/send/rocketmq/{msg}")
    public String sendMessage(@PathVariable String msg) {
        // 消息内容（可自定义，拼接前缀方便区分）
        String message = "【RocketMQ 普通消息】" + msg;
        // 发送消息：参数1=主题名称，参数2=消息内容
        rocketMQTemplate.convertAndSend(TOPIC_NAME, message);
        return "RocketMQ 消息发送成功！发送内容：" + message;
    }
}

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
 * RocketMQ 消息消费者
 * 订阅指定主题，消费消息
 */
@Component // 交给Spring管理，否则无法订阅消息
@RocketMQMessageListener(
        topic = "springboot_rocketmq_topic", // 订阅的主题（必须与生产者的主题一致）
        consumerGroup = "${rocketmq.consumer.group}" // 消费者组（与配置文件中一致）
)
public class RocketMQConsumer implements RocketMQListener<String> {
    /**
     * 接收并处理消息（消息类型与生产者发送的一致，这里是String）
     * @param msg 接收的消息内容
     */
    @Override
    public void onMessage(String msg) {
        // 模拟消息处理（实际项目中替换为业务逻辑，如扣减库存、发送短信等）
        System.out.println("RocketMQ 消费者接收消息：" + msg);
        // 入门阶段暂不添加复杂业务逻辑，后续可扩展异常处理、消息确认等
    }
}

/**
 * 发送同步消息（可靠发送，等待Broker确认）
 */
@GetMapping("/send/rocketmq/sync/{msg}")
public String sendSyncMessage(@PathVariable String msg) {
    String message = "【RocketMQ 同步消息】" + msg;
    // 同步发送，返回发送结果（可判断是否发送成功）
    org.apache.rocketmq.client.producer.SendResult sendResult = rocketMQTemplate.syncSend(TOPIC_NAME, message);
    // 判断发送状态（SUCCESS 表示发送成功）
    if (sendResult.getSendStatus() == org.apache.rocketmq.client.producer.SendStatus.SEND_OK) {
        return "同步消息发送成功！发送内容：" + message + "，消息ID：" + sendResult.getMsgId();
    } else {
        return "同步消息发送失败！";
    }
}

/**
 * 发送异步消息（非阻塞，通过回调处理结果）
 */
@GetMapping("/send/rocketmq/async/{msg}")
public String sendAsyncMessage(@PathVariable String msg) {
    String message = "【RocketMQ 异步消息】" + msg;
    // 异步发送，回调函数处理发送结果
    rocketMQTemplate.asyncSend(TOPIC_NAME, message, new org.apache.rocketmq.spring.support.RocketMQSendCallback() {
        // 发送成功回调
        @Override
        public void onSuccess(org.apache.rocketmq.client.producer.SendResult sendResult) {
            System.out.println("异步消息发送成功，消息ID：" + sendResult.getMsgId());
        }
        // 发送失败回调
        @Override
        public void onException(Throwable e) {
            System.out.println("异步消息发送失败，原因：" + e.getMessage());
        }
    });
    return "异步消息发送中，请查看控制台回调结果！发送内容：" + message;
}