文章目录
- 准备工作
- 项目依赖
- 配置 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:9876consumer:group: batchConsumerGroupproducer: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 {@Autowiredprivate 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>> {@Overridepublic 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 {@Autowiredprivate BatchProducer batchProducer;@GetMapping("/sendBatchMessages")public String sendBatchMessages() {batchProducer.sendBatchMessages();return "批量消息已发送";}
}
通过访问 http://localhost:8080/sendBatchMessages
触发消息发送。
- 调用这个接口会将批量消息发送到 RocketMQ 主题
BatchTopic
。 BatchConsumer
会自动接收并批量处理这些消息。
总结
我们成功在 Spring Boot 中实现了 RocketMQ 的批量消息发送与消费:
- 使用
BatchProducer
类批量发送消息。 - 使用
BatchConsumer
类批量消费消息,并设置最大批量大小。 - 通过简单的 REST API 控制消息发送,确保一切顺利。
批量消息处理可以提高消息传递的效率,适合高并发场景。这种方式可以减少网络开销,并有效利用系统资源。
推荐阅读文章
-
由 Spring 静态注入引发的一个线上T0级别事故(真的以后得避坑)
-
如何理解 HTTP 是无状态的,以及它与 Cookie 和 Session 之间的联系
-
HTTP、HTTPS、Cookie 和 Session 之间的关系
-
什么是 Cookie?简单介绍与使用方法
-
什么是 Session?如何应用?
-
使用 Spring 框架构建 MVC 应用程序:初学者教程
-
有缺陷的 Java 代码:Java 开发人员最常犯的 10 大错误
-
如何理解应用 Java 多线程与并发编程?
-
把握Java泛型的艺术:协变、逆变与不可变性一网打尽
-
Java Spring 中常用的 @PostConstruct 注解使用总结
-
如何理解线程安全这个概念?
-
理解 Java 桥接方法
-
Spring 整合嵌入式 Tomcat 容器
-
Tomcat 如何加载 SpringMVC 组件
-
“在什么情况下类需要实现 Serializable,什么情况下又不需要(一)?”
-
“避免序列化灾难:掌握实现 Serializable 的真相!(二)”
-
如何自定义一个自己的 Spring Boot Starter 组件(从入门到实践)
-
解密 Redis:如何通过 IO 多路复用征服高并发挑战!
-
线程 vs 虚拟线程:深入理解及区别
-
深度解读 JDK 8、JDK 11、JDK 17 和 JDK 21 的区别
-
10大程序员提升代码优雅度的必杀技,瞬间让你成为团队宠儿!
-
“打破重复代码的魔咒:使用 Function 接口在 Java 8 中实现优雅重构!”
-
Java 中消除 If-else 技巧总结
-
线程池的核心参数配置(仅供参考)
-
【人工智能】聊聊Transformer,深度学习的一股清流(13)
-
Java 枚举的几个常用技巧,你可以试着用用