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

Spring Boot 集成 RocketMQ 全流程指南：从依赖引入到消息收发

news 来源：原创 2025/9/5 22:24:04

前言

在分布式系统中，消息中间件是解耦服务、实现异步通信的核心组件。RocketMQ 作为阿里巴巴开源的高性能分布式消息中间件，凭借其高吞吐、低延迟、高可靠等特性，成为企业级应用的首选。而 Spring Boot 通过其“约定优于配置”的设计理念，极大简化了项目开发的复杂度。本文将通过 手动连接 和 配置连接 两种方式，详细讲解如何在 Spring Boot 中集成 RocketMQ，实现消息的同步与异步发送，并提供完整示例代码。

微信图片_20250414010059

‍

一、环境准备

在开始前，请确保：

JDK 17、Maven 3.6+、Spring Boot 2.7+。
安装RocketMQ服务（本地或远程），推荐使用RocketMQ Docker镜像快速搭建(可参考之前文章)。

‍

二、示例—Springboot集成mq（手动连接）

通过编码方式初始化生产者，适用于需要动态控制资源的场景。

2.1 新建项目

‍

2.2 引入依赖

       <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.4</version></dependency>

‍

2.3 生产者发送消息

构建一个消息生产者DefaultMQProducer实例，然后指定生产者组为jihaiProducer；
指定NameServer的地址：服务器的ip:9876，因为需要从NameServer拉取Broker的信息
producer.start() 启动生产者
构建一个内容为：技海拾贝的消息1，然后指定这个消息往jihaishibei这个topic发送
producer.send(msg)：发送消息，打印结果
关闭生产者

public class Producer {public static void main(String[] args) throws Exception {//创建一个生产者，指定生产者组为jihaiProducerDefaultMQProducer producer = new DefaultMQProducer("jihaiProducer");// 指定NameServer的地址producer.setNamesrvAddr("localhost:9876");// 第一次发送可能会超时，设置的比较大producer.setSendMsgTimeout(60000);// 启动生产者producer.start();// 创建一条消息// topic为 jihaishibei// 消息内容为 技海拾贝的消息1// tags 为 TagAMessage msg = new Message("jihaishibei", "TagA", "技海拾贝的消息1 ".getBytes(RemotingHelper.DEFAULT_CHARSET));// 发送消息并得到消息的发送结果，然后打印SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);// 关闭生产者producer.shutdown();}}

‍

启动，发送消息

‍

在控制台可以看到这条消息

‍

这里就能看到发送消息的详细信息。

左下角消息的消费的消费，因为我们还没有消费者订阅这个topic，所以左下角没数据。

‍

2.4 消费者消费消息

创建一个消费者实例对象，指定消费者组为jihaiConsumer
指定NameServer的地址：服务器的ip:9876
订阅 jihaishibei这个topic的所有信息
consumer.registerMessageListener ，这个很重要，是注册一个监听器，这个监听器是当有消息的时候就会回调这个监听器，处理消息，所以需要用户实现这个接口，然后处理消息。
启动消费者

‍

public class Consumer {public static void main(String[] args) throws InterruptedException, MQClientException {// 通过push模式消费消息，指定消费者组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("jihaiConsumer");// 指定NameServer的地址consumer.setNamesrvAddr("localhost:9876");// 订阅这个topic下的所有的消息consumer.subscribe("jihaishibei", "*");// 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {System.out.printf("消费消息:%s", new String(msg.getBody()) + "\n");}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者consumer.start();System.out.printf("Consumer Started.%n");}
}

‍

启动服务，进行消费

‍

在控制台，发现被jihaiConsumer这个消费者组给消费了。

‍

三、示例2—Springboot集成mq（配置连接）

在 Spring Boot 中，可以通过配置文件简化 RocketMQ 的连接配置。以下是在 application.yml 文件中进行的配置：

3.1 配置文件修改

‍

spring:application:name: rocket-mq-demorocketmq:name-server: 127.0.0.1:9876producer:group: rocket-mq-demo-producersend-message-timeout: 10000comsumer:group: rocket-mq-demo-comsumersend-message-timeout: 10000

‍

3.2 添加依赖

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.3.0</version>  
</dependency>

根据需要选择最新版本，从中央仓库可以查看https://central.sonatype.com/artifact/org.apache.rocketmq/rocketmq-spring-boot-starter

‍

备注：如果添加rocketmq-client依赖，先注释这个依赖

‍

3.3 消费者service类

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.Service;/*** messageModel=MessageModel.CLUSTERING* 监听模式，有消息就会消费*/
@Service
@RocketMQMessageListener(topic = "jihaishibei-topic", consumerGroup = "rocket-mq-demo-comsumer", messageModel = MessageModel.CLUSTERING)
public class RocketMQConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String s) {System.out.printf("收到消息: %s\n", s);}
}

‍

3.4 生产者service类

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;@Service
public class RocketMQProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;private final String topic = "jihaishibei-topic";// 1.同步发送消息// 同步发送是指发送方发送一条消息后，会等待服务器返回确认信息后再进行后续操作。这种方式适用于需要可靠性保证的场景。public void createAndSend(String message){rocketMQTemplate.convertAndSend(topic, message);System.out.printf("同步发送结果: %s\n", message);}// 1.同步发送消息// 同步发送是指发送方发送一条消息后，会等待服务器返回确认信息后再进行后续操作。这种方式适用于需要可靠性保证的场景。public void sendSyncMessage(String message){SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build());System.out.println(sendResult.getMsgId());System.out.printf("同步发送结果: %s\n", message);}// 2.异步发送消息// 异步发送是指发送方发送消息后，不等待服务器返回确认信息，而是通过回调接口处理返回结果。这种方式适用于对响应时间要求较高的场景。public void sendAsyncMessage(String message){rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(), new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.printf("异步发送成功: %s\n", sendResult);}@Overridepublic void onException(Throwable throwable) {System.out.printf("异步发送失败: %s\n", throwable.getMessage());}});}// 3.单向发送消息// 单向发送是指发送方只负责发送消息，不关心服务器的响应。该方式适用于对可靠性要求不高的场景，如日志收集。public void sendOneWayMessage(String message){rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(message).build());System.out.println("单向消息发送成功");}
}

‍

3.5 测试controller类

@RequestMapping("api")
@RestController
public class RocketController {@Autowiredprivate RocketMQProducer rocketMQProducer;@GetMapping("/createAndSend")public String createAndSend(@RequestParam String message) {rocketMQProducer.createAndSend(message);return "同步消息发送成功";}@GetMapping("/sendSync")public String sendSync(@RequestParam String message) {rocketMQProducer.sendSyncMessage(message);return "同步消息发送成功";}@GetMapping("/sendAsync")public String sendAsync(@RequestParam String message) {rocketMQProducer.sendAsyncMessage(message);return "异步消息发送中";}@GetMapping("/sendOneWay")public String sendOneWay(@RequestParam String message) {rocketMQProducer.sendOneWayMessage(message);return "单向消息发送成功";}
}

‍

3.6 启动服务

‍

3.7 测试

同步消息1

‍

同步消息2

‍

异步消息

‍

单向发送消息

‍

四、结束语

本文通过手动连接与配置连接两种方式，展示了Spring Boot与RocketMQ的集成实践。手动连接帮助开发者理解底层API逻辑，而Spring Boot的配置化集成则极大简化了开发流程。无论是同步消息的可靠性保障，还是异步消息的性能优化，RocketMQ均能与Spring Boot无缝协作，为分布式系统提供高效的消息通信能力。

未来可进一步探索集群部署、消息重试机制及监控告警，以实现更健壮的消息服务。希望本文能为开发者快速构建高可用的消息系统提供参考！

求点关注-gif动图 138_爱给网_aigei_com