高并发:RocketMQ 削峰实战
点击上方“方志朋”,选择“设为星标”
回复”666“获取新整理的面试文章
转自:WilsonHe
链接:juejin.im/post/5ea159e4f265da47f0794da5
MQ的主要特点为解耦、异步、削峰,该文章主要记录与分享个人在实际项目中的RocketMQ削峰用法,用于减少数据库压力的业务场景,其中RocketMQ的核心组件概念如下:
Producer:生产发送消息
Broker:存储Producer发送过来的消息
Consumer:从Broker拉取消息并进行消费
NameServer:为Producer或Consumer路由到Broker
其中消费流程有以下几点是必须注意的:
RocketMQ的Consumer获取消息是通过向Broker发送拉取请求获取的,而不是由Broker发送Consumer接收的方式。
Consumer每次拉取消息时消息都会被均匀分发到消息队列再进行传输,所以RocketMQ中的很多参数都是针对队列而不是Topic的(这个是重点,顺便吐槽下源码的文档讲的真不清晰,很多都需要自己试错,但Dashboard做得很好),其中每个Broker消息队列(ConsumeQueue)的数量都可以通过RocketMQ DashBoard实时更改调整。
rocketmq-spring-boot-starter用法简介
当开发中需要快速集成RocketMQ时可以考虑使用 rocketmq-spring-boot-starter 搭建RocketMQ的集成环境,但该框架并不完全具备RocketMQ所有的配置简化,如需批量消费消息便需要自定义一个DefaultMQPushConsumer bean去消费了。个人在开发中常用的rocketmq-spring-boot-starter
相关类:
RocketMQListener
接口:消费者都需实现该接口的消费方法onMessage(msg)
。RocketMQPushConsumerLifecycleListener
接口:当@RocketMQMessageListener
中的配置不足以满足我们的需求时,可以实现该接口直接更改消费者类DefaultMQPushConsumer
配置@RocketMQMessageListener
:被该注解标注并实现了接口RocketMQListener
的bean为一个消费者并监听指定topic队列中的消息,该注解中包含消费者的一些常用配置(大部分按默认即可),一般只需更改consumerGroup(消费组)与topic。RocketMQMessageListener
中的属性配置是可以使用Placeholder(占位符)从配置文件或配置中心获取的,如下图:业务案例
有一个点赞业务,不限制用户的点赞数只需进行记录(产品需求,开发提议无效),当每个用户都进行x连击享受数量猛增的快感时如果数据库都需要进行x个点赞数据的插入,数据库毫无疑问会塞死导致崩溃。于是想到可以尝试下MQ削峰,比如每秒来了5000消息但数据库只能承受2000,那我消费时每次只拉取消费1600就好了,剩下的放在Broker堆积慢慢消费就好。由于之前的消息中心也在用RocketMQ,于是确认使用RocketMQ来进行削峰。
环境配置
文章例子环境:1NameServer + 2Broker + 1Consumer
添加maven依赖
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>
复制代码
application.yml配置
rocketmq:name-server: 127.0.0.1:9876producer:group: praise-group
server:port: 10000spring:datasource:driver-class-name: com.mysql.cj.jdbc.Driverusername: rootpassword: tigerurl: jdbc:mysql://localhost:3306/wilson
swagger:docket:base-package: io.rocket.consumer.controller
复制代码
点赞接口
PraiseRecord(点赞记录):
@Data
public class PraiseRecord implements Serializable {private Long id;private Long uid;private Long liveId;private LocalDateTime createTime;
}
复制代码
MessageController(简单的测试接口):
RestController
@RequestMapping("/message")
public class MessageController {@Resourceprivate RocketMQTemplate rocketMQTemplate;@PostMapping("/praise")public ServerResponse praise(@RequestBody PraiseRecordVO vo) {rocketMQTemplate.sendOneWay(RocketConstant.Topic.PRAISE_TOPIC, MessageBuilder.withPayload(vo).build());return ServerResponse.success();}// ......}
复制代码
由于用户可以连续点赞,所以考虑可以在点赞消息的处理上宽松一点(容许消息丢失)以追求更高的性能,因此选择使用sendOneyWay()
进行消息发送。
RocketMQ的消息发送方式主要含syncSend()同步发送、asyncSend()异步发送、sendOneWay()三种方式,sendOneWay()也是异步发送,区别在于不需等待Broker返回确认,所以可能会存在信息丢失的状况,但吞吐量更高,具体需根据业务情况选用。
性能:sendOneWay > asyncSend > syncSend RocketMQTemplate的send()方法默认是同步(syncSend)的,更多可看源码实现。
PraiseListener:点赞消息消费者
@Service
@RocketMQMessageListener(topic = RocketConstant.Topic.PRAISE_TOPIC, consumerGroup = RocketConstant.ConsumerGroup.PRAISE_CONSUMER)
@Slf4j
public class PraiseListener implements RocketMQListener<PraiseRecordVO>, RocketMQPushConsumerLifecycleListener {@Resourceprivate PraiseRecordService praiseRecordService;@Overridepublic void onMessage(PraiseRecordVO vo) {praiseRecordService.insert(vo.copyProperties(PraiseRecord::new));}@Overridepublic void prepareStart(DefaultMQPushConsumer consumer) {// 每次拉取的间隔,单位为毫秒consumer.setPullInterval(2000);// 设置每次从队列中拉取的消息数为16consumer.setPullBatchSize(16);}
}
单次pull消息的最大数目受broker存储的MessageStoreConfig.maxTransferCountOnMessageInMemory
(默认为32)值限制,即若想要消费者从队列拉取的消息数大于32有效(pullBatchSize>32)则需更改Broker的启动参数maxTransferCountOnMessageInMemory
值。在MQ削峰的配置参数里,以下几个DefaultMQPushConsumer
的参数是需要注意一下的:
pullInterval:每次从Broker拉取消息的间隔,单位为毫秒
pullBatchSize:每次从Broker队列拉取到的消息数,该参数很容易让人误解,一开始我以为是每次拉取的消息总数,但测试过几次后确认了实质上是从每个队列的拉取数(源码上的注释文档真的很差,跟没有一样),即Consume每次拉取的消息总数如下:
EachPullTotal=所有Broker上的写队列数和(writeQueueNums=readQueueNums) * pullBatchSize
consumeMessageBatchMaxSize:每次消费(即将多条消息合并为List消费)的最大消息数目,默认值为1,rocketmq-spring-boot-starter 目前不支持批量消费(2.1.0版本)
在消费者开始消息消费时会先从各队列中拉取一条消息进行消费,消费成功后再以每次pullBatchSize的数目进行拉取。
PraiseListener中设置了每次拉取的间隔为2s,每次从队列拉取的消息数为16,在搭建了2master broker且broker上writeQueueNums=readQueueNums=4的环境下每次拉取的消息理论数值为16 * 2 * 4 = 128,在第一次从各队列拉取1条消息(即共8条)后消费成功后会每次就会拉取最多128条消息进行消费,想验证下的可以把onMessage()的insert()改为log.info("1")然后统计单位秒内打印的日志数是否为128。
根据以上配置单Conumer情况下每2s理论消费为128,即每2秒数据库新增的点赞数据大概为128条左右,有20%偏差都在个人可接受范围内,然后对点赞接口进行简单压测1s 2000请求校验MQ效果,根据消费配置理论上需要16次拉取即需32s才能消费完,压测后查看数据库校验效果:
由上图可以看出除第一次2s和最后一次2s外数据库每2s的插入数据数和一般都在128附近波动,也用了34s(因第一次拉取数较少所以比理论多花费一次拉取)消费的偏差大小可能会受每次拉取数pullBatchSize、Broker上的消息队列数、网络波动等情况影响,但需要的目的已经达到了,我只想把单位时间内过多的数据库操作交给MQ做分隔成多个单位时间内的小批量操作,消息过多就堆积,当请求峰值过了后直到MQ堆积的消息消费完前数据库的插入数依旧会与峰值期的插入数相差不大,达到了MQ削峰填谷的效果。
上线了但消费效率预估失误如何动态更改消费效率 ?
当把拉取数pullBatchSize设置Broker的默认最大传输值32了,线上又不想重启Broker更改maxTransferCountOnMessageInMemory参数,如有2个Broker且queue都为4,那么拉取消费效率才为32 * 2 * 4 = 256,如果想要动态调整,可以从Broker数或Broker队列数下手,可以将Broker的writeQueueNums、readQueueNums增大,如都改为8,那么效率就成了32 * 2 * 8 = 512。
需要注意的是更改完queues后必须去Dashboard的Topic下的CONSUMER MANAGER查看新增的队列上是否都有Consumer成功注册上去了,因为遇到了在测试与生产上使用rocketmq-spring-boot-starter @RocketMQListener标注消费者不会自动注册到新队列上的情况,但没排除是不是RocketMQ版本的原因(个人本地的版本比环境上的高了一个小版本0.0.1,本地没出现没消费者注册到新队列上的问题),而是使用了自定义DefaultMQPushConsumer bean(原生的方式都是没有问题的)的备用方案。当再启动新的消费者应用时CONSUMER MANAGER(下图)中就会出现 新Consumer数 * 各Broker队列数和的队列行。
如何使用RocketMQ批量消费 ?
虽然点赞业务使用MQ单条插入后TPS已经达到当前业务指标要求了,但考虑到如果后续要求在不添加机器数的情况下增加TPS,且数据量还没到分库分表的程度,个人就打算从批量消费下手,由一次插入一条点赞记录改为一次性插入多条(insertBatch)。当然能满足现有需求能不做肯定不做的,过度优化过分碍事,但想多点方案不会坏事。
rocketmq-spring-boot-starter并没有提供批量消费的功能,所以要批量消费消息需要自定义DefaultMQPushConsumer
并配置其consumeMessageBatchMaxSize
属性。consumeMessageBatchMaxSize
属性默认值为1,即每次只消费一条消息,需要注意的是该属性也会受pullBatchSize
影响,如果consumeMessageBatchMaxSize
为32但pullBatchSize
只为12,那么每次批量消费的最大消息数也就只有12。如下为个人测试批量消费Consumer的测试bean:
@Bean
public DefaultMQPushConsumer userMQPushConsumer() throws MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketConstant.ConsumerGroup.SPRING_BOOT_USER_CONSUMER);consumer.setNamesrvAddr(nameServer);consumer.subscribe(RocketConstant.Topic.SPRING_BOOT_USER_TOPIC, "*");// 设置每次消息拉取的时间间隔,单位毫秒consumer.setPullInterval(1000);// 设置每个队列每次拉取的最大消息数consumer.setPullBatchSize(24);// 设置消费者单次批量消费的消息数目上限consumer.setConsumeMessageBatchMaxSize(12);consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context)-> {List<UserInfo> userInfos = new ArrayList<>(msgs.size());Map<Integer, Integer> queueMsgMap = new HashMap<>(8);msgs.forEach(msg -> {userInfos.add(JSONObject.parseObject(msg.getBody(), UserInfo.class));queueMsgMap.compute(msg.getQueueId(), (key, val) -> val == null ? 1 : ++val);});log.info("userInfo size: {}, content: {}", userInfos.size(), userInfos);/*处理批量消息,如批量插入:userInfoMapper.insertBatch(userInfos);*/return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});consumer.start();return consumer;
}
如果默认配置情况下log打印出的userInfo size恒为1,但由于设置了consumeMessageBatchMaxSize
与pullBatchSize
,且pullBatchSize
较小,所以每次消费的消息数最大值为12,如下图:
附本文相关信息
确保mqnamesrv与mqbroker已启动成功,如该文章环境的启动:
mqnamesrv -n 127.0.0.1:9876 mqbroker -c E:\RocketMQ\rocketmq-all-4.5.2-bin-release\bin\2m-noslave\broker-a.properties mqbroker -c E:\RocketMQ\rocketmq-all-4.5.2-bin-release\bin\2m-noslave\broker-b.properties
RocketMQ DashBoard启动流程可参考官方github文档或到我的资源里下载jar包运行
源码地址(https://github.com/Wilson-He/spring-boot-series/tree/master/spring-rocketmq),2m-noslave目录是该文章中例子中的2master broker配置与启动脚本,spring-boot-consumer-peak目录为包含该文章相关代码的实际例子
热门内容:面试官:我把数据库部署在Docker容器内,你觉得如何?
炫酷,SpringBoot+Echarts实现用户访问地图可视化(附源码)API 面试四连杀:接口如何设计?安全如何保证?签名如何实现?防重如何实现
为什么 SQL 语句不要过多的 join?最近面试BAT,整理一份面试资料《Java面试BAT通关手册》,覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。获取方式:点“在看”,关注公众号并回复 666 领取,更多内容陆续奉上。
明天见(。・ω・。)ノ♡
高并发:RocketMQ 削峰实战相关推荐
- 高并发:RocketMQ 削峰实战!
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试文章 本文来源: https://juejin.im/post/5ea1 ...
- 架构设计 | 高并发流量削峰,共享资源加锁机制
本文源码:GitHub·点这里 || GitEE·点这里 一.高并发简介 在互联网的业务架构中,高并发是最难处理的业务之一,常见的使用场景:秒杀,抢购,订票系统:高并发的流程中需要处理的复杂问题非常多 ...
- 实战:RocketMQ削峰,这一篇就够了
前言 MQ的主要特点为解耦.异步.削峰,该文章主要记录与分享个人在实际项目中的RocketMQ削峰用法,用于减少数据库压力的业务场景,其中RocketMQ的核心组件概念如下: Producer:生产发 ...
- 超级简单的 RocketMQ 流量削峰实战
知道的越多,不知道的就越多,业余的像一棵小草! 编辑:业余草 来源:https://www.xttblog.com/?p=4996 前言 MQ的主要特点为解耦.异步.削峰,该文章主要记录与分享个人在实 ...
- RocketMQ削峰
MQ的主要特点为解耦.异步.削峰,该文章主要记录与分享个人在实际项目中的RocketMQ削峰用法,用于减少数据库压力的业务场景,其中RocketMQ的核心组件概念如下: Producer:生产发送消息 ...
- Java并发编程实战_阿里P9整理分享的亿级流量Java高并发与网络编程实战PDF
前言 为了帮助初级开发者快速掌握高并发.网络编程.微服务.海量数据的处理这些实用技术,本文以"理论+范例"的形式对各个知识点进行了详细的讲解,力争让读者在实践中快速掌握相关知识. ...
- 阿里P9整理分享的亿级流量Java高并发与网络编程实战PDF
前言 有人调侃我们说: 程序员不如送外卖.送外卖是搬运食物,自己是搬运代码,都不产出新的东西-- 透支体力,又消耗健康,可替代性极强,30岁之后就要面临被优化的危险-- 想跳槽,但是更高的平台难进,同 ...
- 《亿级流量JAVA高并发与网络编程实战》笔记--------更新中
<亿级流量JAVA高并发与网络编程实战>笔记 第一章 高并发概述 "高并发技术" 是一个广义的概念,是指一种高效的地实现并发需求的解决方案,是技术领域的名称,可以包含架 ...
- java 高并发第三阶段实战_JAVA多线程编程实战视频-第三阶段(共80节)
高并发编程第三阶段01讲 AtomicInteger多线程下测试讲解 高并发编程第三阶段02讲 AtomicInteger API详解,以及CAS算法详细介绍 高并发编程第三阶段03讲 利用CAS构造 ...
最新文章
- zabbix Server 4.0 监控TCP的12种状态
- 线上比赛投诉:同一赛点两支队伍比赛车模是否相同?
- crc生成多项式怎么算_利用system Verilog生成任意CRC多项式
- python决策树id3算法_决策树ID3算法预测隐形眼睛类型--python实现
- 团队行为心理学读书笔记(2)招聘背后的心理学
- 解决eclipse中jsp没有代码提示问题
- 计算机节电模式不能打开,电脑进入节电模式打不开怎么办
- 第四十期:2019年度十大Web开发趋势
- php红色字体颜色,php生成文字颜色渐变 高级用法
- php mysql长连接聊天室_PHP之探索MySQL 长连接、连接池
- jenkins 手动执行_想知道如何用Jenkins自动执行Python脚本输出测试报告?
- 新装 Asterisk-1.6 没有 Meetme
- 什么是Python解释器?
- EndNote X7集成到office word2016(原有EndNote选项卡,却突然消失)
- .net core快速开发平台,learun自主工作流引擎设计规范
- 渗透第一个靶机 shop1.0 (还是简单渗透版)
- 微信小程序引入UI组件库
- Deepfacelab 新手教程【AI智能换脸】
- 高数 | 一点可导和邻域内可导能推出来什么?
- ZYNQ裸板:LHB155304-RT篇
热门文章
- Python将MySQL表数据写入excel
- Qt 在Label上面绘制罗盘
- 1016. Phone Bills (25)
- 保证你现在和未来不失业的10种关键技【转载】
- mnist数据集保存为图片
- 在linux中只将“桌面”修改成“Desktop”而系统仍然使用中文
- 【Codeforces】 2A - Winner (map)
- 【MATLAB】find 函数 总结
- android app文档,android App项目需求描述文档.docx
- 如何用 Slack 和 Kubernetes 构建一个聊天机器人?| 附代码