如何在 Spring Boot 中利用 RocketMQ 实现批量消息消费

news2024/11/13 3:57:04

文章目录

- - 准备工作
  - 项目依赖
  - 配置 RocketMQ
  - 生产批量消息
  - 消费批量消息
  - 测试批量消息发送和消费
  - 总结
  - 推荐阅读文章

RocketMQ 是一款分布式消息队列，支持高吞吐、低延迟的消息传递。对于需要一次处理多条消息的场景，RocketMQ 提供了批量消费的机制，这篇文章将展示如何在 Spring Boot 中实现这一功能。

准备工作

在开始之前，请确保你已经安装和配置好 RocketMQ。如果还没安装，请参考 RocketMQ 官网获取安装指南。

项目依赖

首先，我们需要在 Spring Boot 项目中添加 RocketMQ 的依赖。打开 pom.xml 文件，添加以下内容：

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

这个依赖包包含了与 RocketMQ 集成所需的所有内容。

配置 RocketMQ

在 application.yml 文件中添加 RocketMQ 的相关配置：

rocketmq:
  name-server: 127.0.0.1:9876
  consumer:
    group: batchConsumerGroup
  producer:
    group: batchProducerGroup

name-server：RocketMQ 服务的地址
consumer.group：消息消费的分组
producer.group：消息生产的分组

确保 name-server 地址是正确的，指向你的 RocketMQ 服务。

生产批量消息

创建一个消息生产者，用于发送批量消息。以下是 BatchProducer.java 的示例代码：

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

import java.util.ArrayList;
import java.util.List;

@Service
public class BatchProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendBatchMessages() {
        List<Message<String>> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> message = MessageBuilder.withPayload("Hello RocketMQ " + i).build();
            messages.add(message);
        }
        
        rocketMQTemplate.syncSend("BatchTopic", messages, 10000);
        System.out.println("批量消息发送成功！");
    }
}

这里，我们创建了 10 条消息并将它们添加到列表 messages 中。
调用 rocketMQTemplate.syncSend 方法将消息批量发送到主题 BatchTopic。

消费批量消息

接下来，我们创建一个消息消费者，用于批量消费消息。以下是 BatchConsumer.java 的示例代码：

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
@RocketMQMessageListener(topic = "BatchTopic", consumerGroup = "batchConsumerGroup", selectorExpression = "*", consumeMessageBatchMaxSize = 10)
public class BatchConsumer implements RocketMQListener<List<String>> {

    @Override
    public void onMessage(List<String> messages) {
        System.out.println("批量接收到消息：");
        messages.forEach(message -> System.out.println("消息内容：" + message));
    }
}

在这段代码中：

@RocketMQMessageListener 注解用于标识这是一个 RocketMQ 的消息监听器，指定了监听的主题 BatchTopic 和消费分组 batchConsumerGroup。
consumeMessageBatchMaxSize = 10 表示每次批量消费最多 10 条消息。
onMessage 方法会处理接收到的消息列表，并逐条打印出消息内容。

测试批量消息发送和消费

创建一个简单的 Spring Boot 控制器，用于触发批量消息发送。以下是 MessageController.java 的代码：

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

@RestController
public class MessageController {

    @Autowired
    private BatchProducer batchProducer;

    @GetMapping("/sendBatchMessages")
    public String sendBatchMessages() {
        batchProducer.sendBatchMessages();
        return "批量消息已发送";
    }
}