Broker如何检索Message

曾经终败给现在 2024-03-22 09:05 73阅读 0赞

**Consumer发起请求到Broker**

有一个请求，那 Broker 必然会有一个processor初始化后的 Handler 来专门处理它，拉取消息对应的 RequestCode 是 **PULL\_MESSAGE**，顾名思义就是拉取 Message

**Broker进行校验**

**权限验证**

**参数验证**

**核心逻辑处理**

**通过MessageStore获取数据**

MessageStore，即 Broker 对消息存储的封装。

传参:

![0abc579d502251828947f6f70f1a4cb4.png][]

consumergroup、topic、messagequeue、

offset、maxMsgNums、filter

简单翻译一下：

从指定 Topic MessageQueue 中查询 Message，

查询条数由 maxMsgNums 控制，

查询的起始偏移量由 offset 控制，

最后结果将通过 MessageFilter 进一步筛选，

这里其实就是根据 Message 的 Tag 对消息进行过滤。而像 maxMsgNums 和 offset 这些控制变量全部都由 Consumer 传给 Broker。

**offset逻辑偏移量**

queueOffset **就是逻辑偏移量**。大家可以简单地理解为某个 MessageQueue 当中所存储的 Message 数量。

由Consumer发起请求时已经确定，原理是远程请求Broker查询

#### 从远端 Broker 获取 Offset ####

此请求对应的 RequestCode 为 QUERY\_CONSUMER\_OFFSET

![a3c7e4368b8991d82129cacdb64a9242.png][]

请求中会包含 3 个参数，分别是：ConsumerGroup、topic、QueueId。

在 ConsumerOffsetManager 的内部有一个叫 offsetTable 的 Map 来实际存储数据，Map 的详情如下。

Key：$\{Topic\}@$\{ConsumerGroup\} 的组合，即某个 Topic 下的某个消费者组。

Value：还是个 Map，也就是上图中右侧的部分。里面分别记录了 Consumer 在每个 MessageQueue 当中的消费进度。

**最后一步**

### 从 MessageStore 获取 Message ###

这里会通过传入的 Topic 和 MessageQueueID，找到其对应的 ConsumeQueue 索引

举个例子，假设我们要消费某个 MessageQueue 中的**第 2 条消息**，那么此时 RequestHeader 中的 Offset 值就是 1，那么 Broker 会根据 Offset 计算出其在 ConsumeQueue 文件中起始偏移量，即 1 \* 20 = 20 ，每条 ConsumeQueue 大小固定为 **20 字节**，所以通过**偏移量 + 消息长度**就能够快速地定位到 ConsumeQueue。

定位到了 ConsumeQueue 就能够通过其中存储的**物理偏移量** \+ **消息体长度**快速定位到存储在磁盘上的 Message，还能通过 Tag 来进行消息过滤。

[0abc579d502251828947f6f70f1a4cb4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/b099c63015b8488dae6ccd5ab4d2c8c1.png
[a3c7e4368b8991d82129cacdb64a9242.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/0c19b786e55d4ca68d26fd38007929b1.png

发表评论取消回复

表情：

评论列表（有 0 条评论，73人围观）

还没有评论，来说两句吧...

相关阅读

相关 Kafka：broker如何处理请求

前言 ![format_png][] 生产者生产消息，调用Kafka Client提供的API通过send方法发送到broker之后，broker会根据消息请求类型有对

桃扇骨/ 2024年03月24日 14:18/ 0 赞/ 101 阅读

相关 RocketMQ源码12-broker 消息投递流程(处理PULL_MESSAGE请求)

`RocketMq`消息处理整个流程如下： ![format_png][] 1. 消息接收：消息接收是指接收`producer`的消息，处理类是`SendMessageP

谁践踏了优雅/ 2024年03月24日 06:16/ 0 赞/ 81 阅读

相关 Broker如何检索Message

Consumer发起请求到Broker 有一个请求，那 Broker 必然会有一个processor初始化后的 Handler 来专门处理它，拉取消息对应的 RequestC

曾经终败给现在/ 2024年03月22日 09:05/ 0 赞/ 74 阅读

相关 kafka架构生产、消费、broker如何分工

分工合作我们将kafka的分成三个部分：生产者存储Broker 消费者然后看看这三个部分都分别做什么事情。 kafka生产者都干了什么事

末蓝、/ 2022年11月03日 11:09/ 0 赞/ 94 阅读

相关 broker

topic配置，订阅（消费）组配置，调度（延迟）消息配置，消费位点，消费过滤信息启动加载 MessageStore初始化加载等 netty服务端：启动8888和8886端

「爱情、让人受尽委屈。」/ 2022年11月01日 05:56/ 0 赞/ 157 阅读

相关 Git如何规范填写Commit Message

0.规范commit message的好处可以清晰地知道每个 commit 的变更内容可以基于这些 Commit Message 进行过滤查找，比如只查找

喜欢ヅ旅行/ 2022年10月08日 05:47/ 0 赞/ 311 阅读

相关 Elasticsearch如何检索数据

我们都知道Elasticsearch是一个全文检索引擎，那么它是如何实现快速的检索呢？传统的数据库给每个字段都存储成一个单个值，对于全文检索而言，这样的存储是低效的。举个例

布满荆棘的人生/ 2022年05月29日 01:52/ 0 赞/ 313 阅读

相关 IBM WebSphere Message Broker Toolkit中使用数据源

本地数据源的创建：通常本地测试的时候会用到数据源，在本地创建一个操作系统数据源来使用：不同的数据库需要配置不同的数据源，这里需要说明两点： 1.如果本地没有装数据库，

╰半夏微凉°/ 2022年05月23日 10:45/ 0 赞/ 80 阅读

相关 Elasticsearch如何检索数据

我们都知道Elasticsearch是一个全文检索引擎，那么它是如何实现快速的检索呢？传统的数据库给每个字段都存储成一个单个值，对于全文检索而言，这样的存储是低效的

布满荆棘的人生/ 2022年04月06日 12:30/ 0 赞/ 352 阅读

相关如何规范 Git Message

为什么要关注提交信息加快 Reviewing Code 的过程帮助我们写好 release note 5年后帮你快速想起来某个分支，tag 或者 c

小灰灰/ 2022年01月12日 03:57/ 0 赞/ 256 阅读