消息拉取介绍|学习笔记

本文涉及的产品
应用型负载均衡 ALB,每月750个小时 15LCU
网络型负载均衡 NLB,每月750个小时 15LCU
传统型负载均衡 CLB,每月750个小时 15LCU
简介: 快速学习消息拉取介绍

开发者学堂课程【RocketMQ 知识精讲与项目实战(第三阶段)消息拉取介绍】学习笔记,与课程紧密联系,让用户快速学习知识。

课程地址:https://developer.aliyun.com/learning/course/704/detail/12495


消息拉取介绍

 

消息拉取介绍

消息消费拉取的过程,消息消费有两种模式,一种是广播模式,一种是集群模式。广播模式比较简单不存在什么负载均衡,就是每一个消费者需要拉取订阅主题下面的所有队列的消息,那么重点讲解的是集群模式下的一个负载均衡的一个过程。那集群模式之下,同一个消费者组内有多个消费者,然后同一个主题又有多个消息队列,那么消费者是通过负载均衡的方式去消费这个消息,它的做法是一个消息队列只能同一时间被一个消费者去消费,而一个消费者可以去消费多个消息队列,这是它的负载均衡的一个基本思想。

以 PullMessageSerivce来看消息消费实现的机制,为什么要去研究 PullMessageSerivce ?就是刚才所讲的消费者在启动的时候,会看到 mQclientFactory cline 的客户端,它在启动的时候有PullMessageSerivce 拉取消息的服务的启动,服务的启动是做了什么事情?启动之后会调用它的 run 方法,通过PullMessageSerivce 拉取消息的方式去把消息拉下来,那怎么去拉?

while (!this.isStopped())[

try (

PullRequestpul Request = this.pullRequestQueue.take();

this.pullMessage(pullRequest);

] catch(InterruptedException ignored)

]catch (Exception e) (

log.error("PullMessage Service Run Method exception", e);

首先是从拉取队列当中去请求对象,拉取的 request 请求对象之后处理拉取的请求,那么怎么去处理?

private void pullMessage(final PullRequest pullRequest){

FinalMQConsumerInner consumer = this.moclientFactory.selectConsumer(pullRequest.getConsumerGroup());

if (consumer !m null) {

DefaultMQPushConsumerImpl impl =(DefaultMQPushConsumerImp1) consumer;

impl.pullMessage(pullRequest);

)else (

log.warn("No matchedconsumer for the PullRequest (), drop it",pullRequest);

发现它首先从请求对象当中去找到当前的 consumer ,把消费者先去找到,找到之后把消费者发现变成了推送的消费者,然后再去调用 pullMessage 的方法。

image.png

整个过程总结,PullMessageSerivce  随着 MQClientlnstance  的启动而启动起来的,启动起来之后它就会去源源不断的去取出队列当中拉取数据的请求。然后开始处理请求,在处理请求的时候先去拿到消费者,通过消费者的pullMessage  把消息请求进行处理,去拿到消息并且推给消费方。

private String consumerGroup; //消费者组

private MessageQueue messageQueue; //待拉取消息队列

private ProcessQueue processQueue; //消息处理队列

private long nextoffset; //待拉取的Messagebueue偏移量

private boolean lockedFirst = false; //是否被锁定

Pull Request其中包括当前的消费者组、待拉取消息队列、消息处理队列、待拉取的 MessageQueue偏移量以及是否被锁定。

重点研究process queen 从名字上能够看到这是处理的队列,那它和消息队列到底有什么关系?其实processqueen 就是 MessageQueue 在消费端的一个重现或者是一个快照,PullMessageSerivce 从消息服务器去拉取32条消息,默认每次拉取32条消息,按照消息队列的偏移量按顺序存放到process queen当中,然后PullMessageSerivce  就会将消息提交到消费池当中,让它去进行对应的处理,它就可以再一次去拉取新的消息了,它这么做的好处就是能实现数据的转换,并且清空之后再可以从远端拉取新的消息。

所以这里有几个东西需要额外注意,其中PullMessageSerivce 是一个拉取消息的服务。通过这个是从服务端去拉取这个消息,拉取消息拉下来之后把这个消息放到process queen当中,然后让消息消费方去从这里去取出消息进行一个处理。这是拉取消息了之后PullMessageSerivce 和process queen这两个重要的一个作用。

相关实践学习
每个IT人都想学的“Web应用上云经典架构”实战
本实验从Web应用上云这个最基本的、最普遍的需求出发,帮助IT从业者们通过“阿里云Web应用上云解决方案”,了解一个企业级Web应用上云的常见架构,了解如何构建一个高可用、可扩展的企业级应用架构。
相关文章
|
11月前
|
监控 数据可视化 测试技术
从工作分解到产品分解:如何灵活应用项目管理结构?
在项目管理中,工作分解结构(WBS)和产品分解结构(PBS)是将大型任务分解为可管理步骤的关键方法。本文详细介绍了WBS和PBS的概念、应用场景和实践步骤,并推荐了板栗看板、Asana和Microsoft Project等高效项目管理工具,帮助提升项目管理效率。
600 4
|
7月前
|
数据可视化 数据挖掘 BI
Quick BI评测报告
Quick BI评测报告
235 2
|
测试技术 API 数据库
gRPC Status 状态码枚举类型 介绍文档 (更新 gRPC Status 状态码 实操 代码技巧介绍)
gRPC Status 状态码枚举类型 介绍文档 (更新 gRPC Status 状态码 实操 代码技巧介绍)
308 5
|
10月前
|
消息中间件 存储 负载均衡
2024消息队列“四大天王”:Rabbit、Rocket、Kafka、Pulsar巅峰对决
本文对比了 RabbitMQ、RocketMQ、Kafka 和 Pulsar 四种消息队列系统,涵盖架构、性能、可用性和适用场景。RabbitMQ 以灵活路由和可靠性著称;RocketMQ 支持高可用和顺序消息;Kafka 专为高吞吐量和低延迟设计;Pulsar 提供多租户支持和高可扩展性。性能方面,吞吐量从高到低依次为
3075 1
|
开发者 Python
模块化编程:高效编程之道
模块化编程:高效编程之道
|
JavaScript 前端开发 API
介绍Three
【8月更文挑战第21天】介绍Three
426 2
|
自然语言处理 搜索推荐 机器人
阿里巴巴的通义千问大模型
阿里巴巴通义千问是基于Transformer的大型语言模型,预训练于多样化数据集,支持18亿至720亿参数规模。在多模态英文任务中表现出色,且具备多语言对话及图片文本识别能力。可应用于搜索引擎、问答系统和对话交互,提供智能体验。然而,模型在逻辑题和指令理解上存在不足,需在特定领域进行优化。
4223 1
|
消息中间件 JSON 监控
Kafka 的消息格式:了解消息结构与序列化
Kafka 作为一款高性能的消息中间件系统,其消息格式对于消息的生产、传输和消费起着至关重要的作用。本篇博客将深入讨论 Kafka 的消息格式,包括消息的结构、序列化与反序列化,以及一些常用的消息格式选项。通过更丰富的示例代码和深入的解析,希望能够帮助大家更好地理解 Kafka 消息的内部机制。
|
JSON Rust API
【Rust 实战】抖音短视频解析工具
【Rust 实战】抖音短视频解析工具
3042 0
【Rust 实战】抖音短视频解析工具
|
机器学习/深度学习 搜索推荐 算法
天池 DeepRec CTR 模型性能优化大赛 - 夺冠技术分享
这篇文章复盘一下我们本次的参赛经验, 希望对大家有所启发。今天我们会从5大角度来浅谈我们夺冠的优势。