如何在 Spring Boot 中利用 RocketMQ 实现批量消息消费
文章目录
- 准备工作
- 项目依赖
- 配置 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 枚举的几个常用技巧,你可以试着用用
相关文章:
如何在 Spring Boot 中利用 RocketMQ 实现批量消息消费
文章目录 准备工作项目依赖配置 RocketMQ生产批量消息消费批量消息测试批量消息发送和消费总结推荐阅读文章 RocketMQ 是一款分布式消息队列,支持高吞吐、低延迟的消息传递。对于需要一次处理多条消息的场景,RocketMQ 提供了批量消费的机制,这…...
推荐一个Star超过2K的.Net轻量级的CMS开源项目
推荐一个具有模块化和可扩展的架构的CMS开源项目。 01 项目简介 Piranha CMS是一个轻量级且跨平台的CMS库,专为.NET 8设计。 该项目提供多种模板,具备CMS基本功能,也有空模板方便从头开始构建新网站,甚至可以作为移动应用的后端…...
基于驾驶员面部特征的疲劳检测系统
大家好,本文是对基于驾驶员面部特征的疲劳检测系统源码的介绍与说明。 项目下载:基于驾驶员面部特征的疲劳检测系统 1.关于项目 疲劳驾驶检测系统通过监测驾驶人的眼睛状态,头部状态,嘴部状态等指标,识别出疲劳迹象…...
前端知识点---字符串的8种拼接方法(Javascript)
文章目录 01使用 运算符(改变了原始字符串)02使用 运算符(改变了原本的字符串)03 使用 concat() 方法(不改变原本的字符串)04使用模板字面量(不改变原本的字符串)05使用 join() 方法(不改变原本的字符串)①指定分隔符 ②没有指定…...
用 Python 从零开始创建神经网络(一):编码我们的第一个神经元
编码我们的第一个神经元 引言1. A Single Neuron:Example 1Example 2 2. A Layer of Neurons:Example 1 引言 本教程专为那些对神经网络已有基础了解、但尚未动手实践过的读者而设计。尽管网上充斥着各种教程,但很多内容要么过于简略&#x…...
低代码开发
低代码(Low Code)是一种软件开发方法,它通过可视化界面和少量的编码来快速构建应用程序。低代码平台的核心理念是通过抽象和最小化手工编码的方式,加速软件开发和部署的过程。 定义 低代码是一种软件开发方法,它允许…...
sql server 文件和文件组介绍
sql server 文件和文件组介绍 数据库文件和文件组 - SQL Server | Microsoft Learn...
caozha-CEPCS(新冠肺炎疫情防控系统)
caozha-CEPCS,是一个基于PHP开发的新冠肺炎疫情防控系统,CEPCS(全称:COVID-19 Epidemic Prevention and Control System),可以应用于单位、企业、学校、工业园区、村落等等。小小系统,希望能为大…...
1Panel修改PostgreSQL时区
需求 1Panel安装的PostgreSQL默认是UTC时区,需要将它修改为上海时间 步骤 进入PostgreSQL的安装目录 /opt/1panel/apps/postgresql/postgresql/data打开postgresql.conf文件 修改: log_timezone Asia/Shanghai timezone Asia/Shanghai保存后重启…...
开发一个CRM系统难吗?CRM系统的实现步骤
越来越多企业意识到了,客户关系管理(CRM)系统已成为企业提升客户体验、推动销售增长的必备工具。一个高效的CRM系统不仅能够帮助企业优化客户数据管理,还能提升客户满意度,增强客户忠诚度,从而推动业务的持…...
kafka常见面试题总结
Kafka 核心知识解析 一、Kafka 消息发送流程 Kafka 发送消息涉及两个线程:main 线程和 sender 线程。在 main 线程中,会创建一个双端队列 RecordAccumulator,main 线程负责将消息发送给 RecordAccumulator,而 sender 线程则从 R…...
前端知识点---Javascript中检测数据类型函数总结
文章目录 01 typeof 运算符02 instanceof 运算符03 Array.isArray()04 Object.prototype.toString.call()05 constructor 属性06 isNaN() 和 Number.isNaN() (常用)07 isFinite() 和 Number.isFinite()08 typeof null 是 "object" 的问题 01 typeof 运算符 返回值是…...
aspose如何获取PPT放映页“切换”的“持续时间”值
aspose如何获取PPT放映页“切换”的“持续时间”值 项目场景问题描述问题1:从官方文档和资料查阅发现并没有对切换的持续时间进行处理的方法问题2:aspose的依赖包中,所有的关键对象都进行了混淆处理 解决方案1、找到ppt切换的持续时间对应的混…...
【MQTT】代理服务比较RabbitMQ、Mosquitto 和 EMQX
前言 目前要处理大量设备同时频繁发送数据的情况,MQTT协议确实是一个更优的选择,因为它特别适合需要低带宽和高效能的物联网应用,下面是对目前主流协议的对比 数据截止日期:2024年11月10日 基础设施 后端: springclo…...
【C#/C++】C++/CL中String^的含义和举例,C++层需要调用C#层对象时...
示例: String^ IDataServer::GetParam(String^ aParamName){ /// }在 C/CLI 中,String^ 和 IDataServer::GetParam(String^ aParamName) 这种写法是一种混合了 C 和 .NET 的语法,用于在 C 中操作 .NET 对象。C/CLI 是微软扩展的 C 语言&…...
Python学习从0到1 day26 第三阶段 Spark ② 数据计算Ⅰ
人总是会执着于失去的,而又不珍惜现在所拥有的 —— 24.11.9 一、map方法 PySpark的数据计算,都是基于RDD对象来进行的,采用依赖进行,RDD对象内置丰富的成员方法(算子) map算子 功能:map算子…...
【详细】如何优雅地删除 Docker 容器与镜像
内容预览 ≧∀≦ゞ 镜像与容器的区别删除容器和镜像的具体步骤1. 删除容器步骤 1:查看当前运行的容器步骤 2:停止容器步骤 3:删除容器 2. 删除镜像步骤 1:查看镜像列表步骤 2:删除镜像 3. 删除所有容器和镜像 使用 1Pa…...
Spring Spring Boot 常用注解总结
在 Java 开发中,Spring 和 Spring Boot 框架广泛应用于企业级应用开发。这两个框架提供了丰富的注解,使得开发更加高效和便捷。本文将对 Spring 和 Spring Boot 中常用的注解进行总结。 一、Spring 常用注解 1. Component 作用:用于将普通的…...
Flink独立集群+Flink整合yarn
Flink独立集群的搭建: 1、上传解压配置环境变量 # 1、解压 tar -xvf flink-1.15.4-bin-scala_2.12.tgz # 2、修改环境变量 export FLINK_HOME/usr/local/soft/flink-1.15.4 export PATH$PATH:$FLINK_HOME/bin 2、修改配置文件 cd /usr/local/soft/flink-1.15.4/…...
动态规划 之 简单多状态 dp 问题 算法专题
一. 按摩师 按摩师 状态表示 根据经验 题目要求 dp[i] 表示: 选择到i位置时, 此时的最长预约时长 但是根据题目又分成两种情况: f[i] : 选择到 i 位置的时候, nums[i] 必选, 此时的最长预约时长 g[i] : 选择到 i 位置的时候, nums[i] 不选, 此时的最长预约时长状态转移方程 …...
线程同步:确保多线程程序的安全与高效!
全文目录: 开篇语前序前言第一部分:线程同步的概念与问题1.1 线程同步的概念1.2 线程同步的问题1.3 线程同步的解决方案 第二部分:synchronized关键字的使用2.1 使用 synchronized修饰方法2.2 使用 synchronized修饰代码块 第三部分ÿ…...
JVM垃圾回收机制全解析
Java虚拟机(JVM)中的垃圾收集器(Garbage Collector,简称GC)是用于自动管理内存的机制。它负责识别和清除不再被程序使用的对象,从而释放内存空间,避免内存泄漏和内存溢出等问题。垃圾收集器在Ja…...
Axios请求超时重发机制
Axios 超时重新请求实现方案 在 Axios 中实现超时重新请求可以通过以下几种方式: 1. 使用拦截器实现自动重试 import axios from axios;// 创建axios实例 const instance axios.create();// 设置超时时间 instance.defaults.timeout 5000;// 最大重试次数 cons…...
【HTTP三个基础问题】
面试官您好!HTTP是超文本传输协议,是互联网上客户端和服务器之间传输超文本数据(比如文字、图片、音频、视频等)的核心协议,当前互联网应用最广泛的版本是HTTP1.1,它基于经典的C/S模型,也就是客…...
JVM暂停(Stop-The-World,STW)的原因分类及对应排查方案
JVM暂停(Stop-The-World,STW)的完整原因分类及对应排查方案,结合JVM运行机制和常见故障场景整理而成: 一、GC相关暂停 1. 安全点(Safepoint)阻塞 现象:JVM暂停但无GC日志,日志显示No GCs detected。原因:JVM等待所有线程进入安全点(如…...
学习STC51单片机32(芯片为STC89C52RCRC)OLED显示屏2
每日一言 今天的每一份坚持,都是在为未来积攒底气。 案例:OLED显示一个A 这边观察到一个点,怎么雪花了就是都是乱七八糟的占满了屏幕。。 解释 : 如果代码里信号切换太快(比如 SDA 刚变,SCL 立刻变&#…...
接口自动化测试:HttpRunner基础
相关文档 HttpRunner V3.x中文文档 HttpRunner 用户指南 使用HttpRunner 3.x实现接口自动化测试 HttpRunner介绍 HttpRunner 是一个开源的 API 测试工具,支持 HTTP(S)/HTTP2/WebSocket/RPC 等网络协议,涵盖接口测试、性能测试、数字体验监测等测试类型…...
C# 表达式和运算符(求值顺序)
求值顺序 表达式可以由许多嵌套的子表达式构成。子表达式的求值顺序可以使表达式的最终值发生 变化。 例如,已知表达式3*52,依照子表达式的求值顺序,有两种可能的结果,如图9-3所示。 如果乘法先执行,结果是17。如果5…...
打手机检测算法AI智能分析网关V4守护公共/工业/医疗等多场景安全应用
一、方案背景 在现代生产与生活场景中,如工厂高危作业区、医院手术室、公共场景等,人员违规打手机的行为潜藏着巨大风险。传统依靠人工巡查的监管方式,存在效率低、覆盖面不足、判断主观性强等问题,难以满足对人员打手机行为精…...
【p2p、分布式,区块链笔记 MESH】Bluetooth蓝牙通信 BLE Mesh协议的拓扑结构 定向转发机制
目录 节点的功能承载层(GATT/Adv)局限性: 拓扑关系定向转发机制定向转发意义 CG 节点的功能 节点的功能由节点支持的特性和功能决定。所有节点都能够发送和接收网格消息。节点还可以选择支持一个或多个附加功能,如 Configuration …...
