RocketMQ整合Spring Boot实现普通消息功能

一、引言

Apache RocketMQ 是一款开源的分布式消息中间件，广泛应用于企业的异步消息传递场景。Spring Boot 是一个基于 Spring 的快速开发框架，极大地简化了 Java 应用的开发过程。将 RocketMQ 与 Spring Boot 整合，可以更方便地实现消息的发送和接收。本文将详细介绍如何在 Spring Boot 应用中整合 RocketMQ，并实现普通消息的发送与接收功能。

二、环境准备

1. 安装 RocketMQ

首先，确保你的环境中安装了 Java 1.8 或更高版本。接着，可以通过以下命令下载并启动 RocketMQ：

# 下载 RocketMQ
wget https://archive.apache.org/dist/rocketmq/4.9.2/rocketmq-all-4.9.2-bin-release.zip
unzip rocketmq-all-4.9.2-bin-release.zip
cd rocketmq-all-4.9.2/bin

# 启动 NameServer
sh mqnamesrv &

# 启动 Broker
sh mqbroker -n localhost:9876 &

2. 创建 Spring Boot 项目

可以使用 Spring Initializr 创建一个新的 Spring Boot 项目，选择以下依赖：

  • Spring Web
  • Spring Boot DevTools
  • RocketMQ

也可以使用 Maven 或 Gradle 手动添加依赖。

Maven 依赖

在 pom.xml 中添加以下依赖：

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

Gradle 依赖

在 build.gradle 中添加以下依赖：

implementation 'org.apache.rocketmq:rocketmq-spring-boot-starter:2.3.0'

3. 配置 application.yml

在 src/main/resources/application.yml 文件中，配置 RocketMQ 的连接信息：

rocketmq:
  name-server: 127.0.0.1:9876

三、普通消息的生产者

1. 创建消息生产者

在 Spring Boot 应用中创建一个消息生产者类，负责发送普通消息。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String message) {
        // 发送普通消息
        rocketMQTemplate.convertAndSend(topic, message);
        System.out.println("消息发送成功: " + message);
    }
}

2. 创建控制器

创建一个控制器，暴露一个 API 接口，用于接收发送消息的请求。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {

    @Autowired
    private MessageProducer messageProducer;

    @PostMapping("/send")
    public String sendMessage(@RequestParam String topic, @RequestParam String message) {
        messageProducer.sendMessage(topic, message);
        return "消息发送请求已接收";
    }
}

四、普通消息的消费者

1. 创建消息消费者

在 Spring Boot 应用中创建一个消息消费者类，负责接收普通消息。

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageMapping;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "${rocketmq.topic}", consumerGroup = "my-consumer-group", consumeMode = ConsumeMode.CONCURRENTLY)
public class MessageConsumer {

    @MessageMapping
    public void onMessage(String message) {
        System.out.println("收到消息: " + message);
    }
}

2. 配置消费者的 Topic

在 application.yml 中配置消费者监听的 Topic：

rocketmq:
  topic: your-topic-name

五、测试功能

1. 启动应用

确保 RocketMQ 正在运行，接着启动 Spring Boot 应用。

2. 发送消息

使用 Postman 或 cURL 测试消息发送接口：

curl -X POST "http://localhost:8080/send?topic=your-topic-name&message=Hello RocketMQ"

3. 查看消息接收

在控制台中，你应该能看到消费者成功接收到消息的输出。

六、总结

通过上述步骤，我们成功地将 RocketMQ 整合到 Spring Boot 应用中，实现了普通消息的发送与接收功能。这种整合方式不仅提高了开发效率，还让我们能够轻松地实现异步处理和高并发场景下的消息传递。在实际应用中，我们可以根据业务需求，进一步扩展消息的处理逻辑，提升系统的性能和可用性。