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;
}