当前位置：首页 > news >正文

（七）Spring Cloud Alibaba 2023.x：RocketMQ 消息队列配置与实现

news 2026/5/12 12:26:27

前言

准备

安装RocketMq服务

下载rocketmq服务

下载rocketmq 控制台

项目集成

引入依赖

生产者服务配置

消费者服务配置

发送队列消息

前言

在微服务架构中，异步消息通信是实现系统解耦、提高性能和增强系统可靠性的重要手段。在 Spring Cloud Alibaba 生态中，RocketMQ 与 Spring Boot 深度集成，提供了开箱即用的消息通信解决方案，极大地简化了开发流程，提升了系统的扩展性和可维护性。

准备

jdk17+
maven3.9.4+
idea2023
spring cloud: 2023.0.1.0
spring cloud alibaba: 2023.0.1

源码获取：GitHub - /spring-cloud-alibaba-base-demo: 基于spring cloud alibaba生态快速构建微服务脚手架

安装RocketMq服务

本地window系统安装rocketmq服务

下载rocketmq服务

当前博文版本使用的是：5.2

官方下载地址：下载 | RocketMQ
百度网盘地址：百度网盘请输入提取码提取码: 92h6

下载完成解压

在window系统中本地启动rocketmq服务需要配置环境变量，步骤和配置与jdk配置相似

Path变量中添加环境变量

配置完成后进入本地rocketmq的bin文件夹中

首先双击mqnamesrv.cmd启动NameServer服务: 它的作用是提供服务发现、路由信息管理、Broker注册和客户端查询等功能

NameServer服务启动成功后，启动Broker服务：它的作用是消息的存储和传递

当前文件夹cmd执行命令：mqbroker -n localhost:9876

下载rocketmq 控制台

当前博文版本使用的是：2.0.1

官方下载地址：Releases · apache/rocketmq-dashboard
百度网盘地址：百度网盘请输入提取码提取码: 92h6

下载完成解压

需要我们自行对项目打包获取jar包

在当前文件夹中cmd执行命令打包：

mvn clean package -Dmaven.test.skip=true

在target文件中找到生成的jar

执行命令启动

java -jar rocketmq-dashboard-2.0.1.jar

浏览器访问地址：http://localhost:8080

注:如果rocketmq的服务地址或者端口进行过调整，那么在打包前自行到项目的application.yml配置文件中更改后再打包

项目集成

根据前面的博文，目前我们已经创建了三个微服务

引入依赖

分别在子模块生产者和消费者服务的pom.xml文件中引入依赖

<!-- 引入消息队列 stream rocketmq --> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-stream-rocketmq</artifactId> </dependency>

生产者服务配置

yml文件配置

spring: application: name: http-cloud-producer cloud: stream: #消息中间件 rocketmq: binder: name-server: localhost:9876 bindings: producer-out-0: producer: group: output_1 bindings: producer-out-0: destination: topic0 producer-out-1: destination: topic1

相关参数/值解释:
output_1 :自定义的消费者组名称
producer-out-0 ：自定义的生产者通道名称
topic0 ：自定义队列名称

生成者发送消息服务类实现

@Service public class MessageProducer { // 自动注入配置文件中绑定的通道 @Autowired private StreamBridge streamBridge; // 消息通道 public void sendMessageToOutput0(String messageContent) { Map<String, Object> headers = new HashMap<>(); headers.put(MessageConst.PROPERTY_TAGS, "tag0"); Message<String> msg = new GenericMessage<>(messageContent, headers); streamBridge.send("producer-out-0", msg); } public void sendMessageToOutput1(String messageContent) { Map<String, Object> headers = new HashMap<>(); headers.put(MessageConst.PROPERTY_TAGS, "tag1"); Message<String> msg = new GenericMessage<>(messageContent, headers); streamBridge.send("producer-out-1", msg); } }

定义控制层接口

@RestController public class DemoController { @Autowired private MessageProducer messageProducer; /** * 发送消息队列 topic0 * @param message * @return */ @GetMapping(value = "/test/mq/topic0") String sendMqTopic0(@RequestParam("message") String message) { messageProducer.sendMessageToOutput0(message); return "topic0消息发送成功了~~~"; } /** * 发送消息队列 topic1 * @param message * @return */ @GetMapping(value = "/test/mq/topic1") String sendMqTopic1(@RequestParam("message") String message) { messageProducer.sendMessageToOutput1(message); return "topic1消息发送成功了~~~"; } }

消费者服务配置

配置yml文件

spring: application: name: http-cloud-consumer cloud: stream: #消息中间件 rocketmq: binder: name-server: localhost:9876 bindings: consumer0-in-0: consumer: messageModel: CLUSTERING consumer1-in-0: consumer: messageModel: CLUSTERING bindings: consumer0-in-0: destination: topic0 group: clustering-consumer consumer1-in-0: destination: topic1 group: clustering-consumer

相关配置解释：
consumer0-in-0：自定义的消费者通道名称
messageModel：消费者消费模式（集群或者是广播）。本文配置的是集群模式。
topic0 ：需要监听的消费者队列名称
clustering-consumer ：自定义的消费者组名称

实现消费者mq消费类

@Component public class MessageConsumer { /** * 通过方法名称自动绑定到符合条件的消费者通道 * （注：当前遇到的难点是一个消费者服务只能实现一个消费方法，实现多个消费方法会使消费功能失效） * @return */ @Bean public Consumer<Message<String>> consumer1() { return msg -> { System.out.println(Thread.currentThread().getName() + " Consumer1 Receive New Messages: " + msg); }; } }