李锋镝的博客

  • 首页
  • 时间轴
  • 说说
  • 左邻右舍
  • 博友圈
  • 关于我
    • 关于我
    • 另一个网站
    • 我的导航站
    • 网站地图
    • 赞助
  • 留言
  • 走心评论
  • 系列文章
  • Now
  • 每日心情
  • 🚇开往
Destiny
自是人生长恨水长东
  1. 首页
  2. 转载
  3. 技术
  4. 正文

RocketMQ的push消费方式实现详解

2022年8月9日 约 3,007 字11 分钟 34点热度 0人点赞 3条评论
本文最后更新于 2022年8月9日,距今已 1459 天,其中的信息可能已经发生变化,请注意甄别。

MQ消费方式

消费方式就是指消费者如何从MQ中获取到消息,分为两种方式,push(推方式)和pull(拉方式)。

1、push(推方式)

push,顾名思义,就是推的意思。就是当MQ收到生产者产生的消息的时候,会主动将消息推送到消费者进行消费,这种模式就叫push,也就是MQ将消息推给到消费者的意思。

自动草稿push模式

push这种模式的好处就是响应快,消息的实时性比较高,一旦消息MQ收到消息,那么就能立马将消息推送给消费者,消费者也就能立马收到消息进行消费。

但是这种push的模式,有个缺点就是一旦消息量比较大时,对消费者性能要求比较高,因为是消费者无法控制MQ消息的推送速度,一旦消息量大,那么消费者消费消息的压力就比较大。

2、pull(拉方式)

push是MQ主动给消费者推消息,那么pull呢?刚好跟push相反,就是消费者主动去MQ中拉取消息。

自动草稿pull模式

那么pull的优缺点自然也就跟push刚好相反。因为是消费者主动去MQ中拉取消息,那么消费者可根据自身消费的情况,决定何时去拉取消息,主动权在自己手上,这样消费者的压力就会相对小点;但是缺点也很明显,那么就会实时性相对于push方式会低一些,因为你得决定拉的时间间隔。

其实想想,消费方式就跟拿快递一样,快递就是一个消息,我自己就是消费者,快递要么快递小哥主动送(push)到家,要么我自己去快递站拿(pull)。

RocketMQ对于消费方式的实现

上一节说了消费消息的两种方式push和pull,或者说算一种理念。尚大的周阳老师有一句经常说的话我比较赞同,那就是“天上飞的理念,必然有落地的实现”。所以push或者pull到底如何落地,得看具体的MQ的产品了。

而RocketMQ作为阿里开源的一款高性能、功能丰富的MQ,自然同时实现了push和pull的两种消费方式,用户可以选择在项目中使用push还是pull。

自动草稿push模式的实现自动草稿pull模式的实现

但是一般情况下,项目中都是使用push的方式来消费,因为pull除了时实性差外,pull方式还得让开发人员主动去维护消息消费进度,增加额外的操作。

所以接下来就着重讲一下RocketMQ是如何实现push的逻辑。

RocketMQ聪明地实现push的原因

上文说到push模式的优点是时实性好,但是缺点就是消费者压力会比较大,所以,难道实现push模式,只能舍弃压力的控制么?

就在这时,RocketMQ大喊了一声

自动草稿

是的,RocketMQ对于push模式做到了实时和压力的平衡,这主要是因为RocketMQ的push模式其实算是一个“伪push”模式,真正底层的实现还是基于pull。

到这里可能有的小伙伴比较迷糊,怎么push变成“伪push”了,还是用pull实现的,到底是push还是pull?

自动草稿

前面我说过,push和pull只是一种理论,具体的实现看MQ。

所以RocketMQ为了兼顾两者,就选择通过消费者主动拉消息来实现push的效果,这也是为什么我称为“伪push”的原因,RocketMQ都给封装好了,让你用起来感觉是MQ主动push消息给你的。

既然底层是pull,那么RokcetMQ在实现消费者的逻辑的时候,就可以很容易实现控制压力的效果,毕竟这是“拉”方式天然自带的buff;但是如何通过pull实现push的时实的优点呢?毕竟鱼和熊掌我RokcetMQ偏要兼得。

这时这就不得不提到一种叫“长轮询”的机制。

轮询与长轮询

轮询与长轮询都属于pull的实现,都是由客户端主动给服务端发送请求,拉取数据。套到MQ中,就是都是消费者主动去MQ拉消息。

轮询

轮询是指不管服务端数据有无更新,客户端每隔定长时间请求拉取一次数据,可能有更新数据返回,也可能什么都没有。

自动草稿

再拿快递举例子,轮询就好比,小明买的iphone 13 pro max快递到了,显示正在派送中,但是小明等不及了,于是就去快递站拿,但是快递还没放到快递站,但是小明的心里急啊,他忍受不了相思之苦,于是小明每隔5分钟就往快递站跑一次,问一下快递到了没,到了就拿回来。这就是轮询的意思,也就是不论有没有数据,客户端都会每隔一定时间去请求一次服务端。

来分析一下拿快递的例子的问题:

  • 每隔5分钟就往快递站跑,那不是累死个小明么。
  • 还有一个问题,假设刚跑到快递站,快递没到,就回去了,但是刚到家的时候,快递到了,于是又等了5分钟,再去快递站终于拿到快递了,但是其实快递都到了几分钟了,你还是没有第一时间拿到快递,这就造成了延迟。

从而对应到程序中,就是会产生如下问题

  • 对于消息而言,会一直产生,这就要求消费者不停地间隔一定时间去拉取消息,即使没有消息也需要去请求,就会造成大量无用的请求,白白浪费大量耗费服务器内存和宽带资源。
  • 可能造成数据的延迟

长轮询

说长轮询概念之前,先来救救小明吧,毕竟小明可不想狗带。

既然原先小明每隔5分钟跑一次,那么是不是可以换种思路,当快递还没到的时候,让小明不要回来,直接在快递站待着,当快递到的时候,才让小明拿着快递回家。这下小明就喜死了,既可以有时间刷刷某音,逛逛某东,还可以在第一时间拿到13 pro max。

自动草稿

所以这种可以在快递站等待的机制,就叫长轮询。

长轮询也是客户端请求服务端,如果服务端有数据,那么就立马返回,客户端再次请求;当服务端不存在数据的时候,服务端并不会给客户端响应,而是将请求给hold住,当服务端有数据的时候才会给客户端响应,返回数据。

所以长轮询可以解决如下问题

  • 解决轮询带来的频繁请求服务端但是没有的问题
  • 一旦新的数据到了,那么消费者能立马就可以获取到新的数据,所以从效果上,有点像是push的感觉。

但是长轮询也会带来服务端代码实现逻辑复杂的问题,当然相比于优点来说,都不太重要。

push消费方式源码探究

理论都讲完了,接下来就到了show me the code的时间了,来看看RocketMQ的是如何通过长轮询机制来实现压力和时实的平衡。

这里我画了一张push模式下消费者消费流程图。

自动草稿消费者拉取消息的逻辑

  • ①消费者有一个后台线程,会去处理拉取消息(PullRequest)
  • ②先去判断有没有过多消息没有消费,如果有的话,那么就间隔一定时间再次从①开始执行拉取消息的逻辑
  • ③消费者没有过多消息没有消费,那么就会直接向MQ发送拉取消息的请求,有消息就返回,没有消息就hold住请求,等有新的消息到的时候才返回
  • ④消费者获取到消息之后,会去找用户自定义的消息处理逻辑的实现(MessageListener的实现)去消费消息,同时会再次拉取消息,继续从①开始执行逻辑

1、消费者拉取消息控制压力源码

当消费者准备去拉消息的时候,会先去判断当前消费者消费的压力再决定是否去拉取消息。

RocketMQ提供了两种判断消费压力逻辑,一种是基于还未消费的消息的数量的大小,还有一种是基于还未消费的消息所占内存的大小。

自动草稿控制压力源码

  • 判断还未消费消息的数量,数量太多就等会再执行重新执行拉取消息的逻辑
  • 判断还未消费消息的大小,如果还未消息的消息占用的内存过大,就等会再执行重新执行拉取消息的逻辑

总的一句话就是,当消费者消费的压力过大时,就不会去拉取消息,而是等待一定的时间再去执行拉取消息的逻辑,如果压力还是很大,就还继续等,如此循环,直到消费者的消费压力小于阈值的时候,才会真正的发送请求到MQ中拉取消息。

2、MQ将请求hold住源码

当服务端未找到消息时,就将请求进行挂起,存起来

自动草稿请求hold住源码

拉取不到消息时,会调用PullRequestHoldService的suspendPullRequest方法讲请求存储起来。PullRequestHoldService是用来存储拉取请求的类。

自动草稿PullRequestHoldService

suspendPullRequest方法会将请求分类,放到ManyPullRequest里,然后用一个ConcurrentHashMap进行存储

3、MQ收到消息响应给消费者的源码

自动草稿NotifyMessageArrivingListener

当生产者发送的消息达到MQ的时候,MQ会回调NotifyMessageArrivingListener的arriving方法,之后就会调用PullRequestHoldService的notifyMessageArriving方法,MQ会重新处理拉取消息的逻辑,此时就能找到最新来的那条消息,从而将最新的消息通过网络返回给消费者。

自动草稿notifyMessageArriving和返回消息逻辑

最后

所以从以上的分析可以看出,RocketMQ对于push的消费方式的实现是基于长轮询机制来实现的,同时平衡了时实和压力,这其实就很nice了。

最后我想说一句,其实不论是pull还是push,又或是轮询和长轮询,其实都是一种理论或者说是一种思想,不单单是MQ的东西,就比如在Nacos中,也使用了push和长轮询机制。但是这些理论在不同产品的具体实现,实现方式可能不太一样,但都是大同小异,所以当你懂了这些思想,再看其它框架的源码,其实就很容易了。

 

原文链接:https://www.cnblogs.com/zzyang/p/16565620.html

除非注明,否则均为李锋镝的博客原创文章,转载必须以链接形式标明本文链接

本文链接:https://www.lifengdi.com/transport/technology/3937

推荐阅读

  • RocketMQ入门级教程
  • 使用RocketMQ时,服务启动过程中,Consumer在服务未启动时消费消息问题处理
  • MQ消费端遇到瓶颈该怎么办?
  • ThreadLocal如何解决内存泄漏问题
  • Redisson分布式锁的watch dog自动续期机制
本作品采用 知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议 进行许可
标签: JAVA MQ RocketMQ
最后更新:2022年8月9日

岁月同一天 8 月 8 日

回望过去的今天,你在写什么

  • 4 年前 2022年8月8日
    减肥四个月~

    今年四月初开始每天跳绳减肥,到了今天刚好四个月了。从一开始的每天跳绳十分钟到半个小时,再从半个小时到一个小时,最后又稳定…

相关文章
  • 以面试官视角万字解读线程池10大经典面试题2025年5月14日
  • mybatis-plus-join-boot-starter介绍及用法2025年4月24日
  • CPU飙高,系统性能问题如何排查?2020年10月9日
  • JAVA技术点总结一2019年11月18日
  • Eclipse反编译插件Jad安装2019年9月27日

李锋镝

既然选择了远方,便只顾风雨兼程。

打赏 点赞
< 上一篇
下一篇 >
1234567891112131415161718192021222324252627282930313233343536373839404142434446474849505152535455575859606162636465666769727476777879808182858687909293949596979899
取消回复
…

文章评论

  • 城南旧事Lv 1

    不瞒你说,我看不懂哈哈哈

    macOSFirefox 102.0 中国-郑州市
    2022年8月9日
    00 回复
    • 李锋镝管理

      @城南旧事 多看几次就可以了 :douyin.74:

      macOSChrome 104.0.0.0 中国-北京市
      2022年8月10日
      00 回复
      • 城南旧事Lv 1

        @李锋镝 好吧

        macOSEdge 107.0.1418.28 中国-郑州市
        2022年11月7日
        00 回复
  • 九年没联系的女友突然问我借钱,我开口道:四十万行么?不够了再说话。”前女友很是高兴地回复说:“够用,还是我男朋友有办法啊,你放心,我日后一定还给你,九年多没见了你现在干啥呢,这么有钱?”我呵呵一笑:“没事,不够用您尽管说话,俺现在改做抵押贷款了。”

    听点儿音乐吧 朋友~
    文章目录
    最新 热点 随机
    最新 热点 随机
    Kratos+ v1.1.14版本更新说明 Spring Boot 指定外部配置文件的方式 Spring Boot 配置加载优先级总结 Claude Fable 5(claude-fable-5)深度详解 如何通过命令查看Java应用内存中对象数量 关于服务的探活端口和业务端口不一致有什么问题
    给主题增加了Now、每日心情、年度回顾、岁月同一天、随机漫步等功能AI时代,个人技术博客的出路在哪里?增加了两套复古皮肤-牛皮纸、千禧网页这个域名注册整整十年了,十年时间,真快啊WordPress实现用户评论等级排行榜插件WordPress网站换了个字体,差点儿把样式换崩了
    MybatisCodeHelperPro激活 解决kubectl exec -it xxxx-service-bfbd45bb9-ktvzj bash -n bit error: exec [POD] [COMMAND] is not supported anymore. Use exec [POD] -- [COMMAND] instead See 'kubectl exec -h' for help and examples Apollo配置中心中的protalDB的作用是什么 k8s + docker + Jenkins使用Pipeline部署SpringBoot项目时Jenkins错误集锦 彻底搞懂mysql日志系统binlog,redolog,undolog Spring事件驱动深度指南:从单机异步到亿级流量,比MQ更轻的架构神器
    最近评论
    Huo 发布于 8 小时前(08月07日) 挺有特色的主题,还是感觉 WP 的确是强大
    李锋镝 发布于 21 小时前(08月07日) 没理解你想说啥
    aboss 发布于 21 小时前(08月07日) 你的后台web-login?
    李锋镝 发布于 22 小时前(08月07日) 这个专门的插件实现的功能更好更全,还能对接支付之类的
    李锋镝 发布于 22 小时前(08月07日) 自定义登录地址是为了防止大部分机器人通过WP固定登录页面暴力破解用户账号密码
    标签聚合
    K8s 日常 Claude docker JVM WordPress Redis AI编程 多线程 SQL 架构 MySQL IDEA JAVA SpringBoot Spring AI 数据库 ElasticSearch 分布式
    友情链接
    • Blogs·CN
    • 懋和道人
    • 旧时繁华
    • Honesty
    • 拾趣博客导航
    • 搬砖日记
    • 知向前端
    • 临窗旋墨
    • 皮皮社
    • 瓦匠个人小站
    • 志文工作室
    • 韩小韩博客
    • 风渡言
    • 彬红茶日记
    • 老张博客
    • Mr.Sun的博客
    • 韩情脉脉
    • 哥斯拉
    • 林羽凡

    COPYRIGHT © 2026 lifengdi.com. ALL RIGHTS RESERVED.

    正在博友圈履约中

    域名年龄

    Theme Kratos+ By Dylan Li

    津ICP备2024022503号-3

    京公网安备11011502039375号