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

前言

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

​

‍

一、环境准备

在开始前，请确保：

1. JDK 17、Maven 3.6+、Spring Boot 2.7+。
2. 安装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 {
        //创建一个生产者，指定生产者组为jihaiProducer
        DefaultMQProducer producer = new DefaultMQProducer("jihaiProducer");

        // 指定NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 第一次发送可能会超时，设置的比较大
        producer.setSendMsgTimeout(60000);

        // 启动生产者
        producer.start();

        // 创建一条消息
        // topic为 jihaishibei
        // 消息内容为 技海拾贝的消息1
        // tags 为 TagA
        Message 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() {

            @Override
            public 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-demo

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: rocket-mq-demo-producer
    send-message-timeout: 10000
  comsumer:
    group: rocket-mq-demo-comsumer
    send-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> {

    @Override
    public 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 {
    @Autowired
    private 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() {

            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("异步发送成功: %s\n", sendResult);
            }

            @Override
            public 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 {
    @Autowired
    private 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无缝协作，为分布式系统提供高效的消息通信能力。

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

​