RocketMQ组件Broker目录结构和存储机制 |
您所在的位置：网站首页 › rocketmq组建 › RocketMQ组件Broker目录结构和存储机制 |
RocketMQ组件Broker目录结构和存储机制
|
携手创作，共同成长！这是我参与「掘金日新计划 · 8 月更文挑战」的第12天，点击查看活动详情 1 Broker 目录结构Broker在安装启动之后会自动生成如下的存储文件： 包含具体的Commitlog文件，是真正存储消息的地方。文件名长度20个字符，由该文件保存消息的最大物理offset值最高位补0组成。文件大小最大为1G，可以通过配置mapedFileSizeCommitLog进行配置。 `00000000000000000000` 第一个文件，物理偏移量为0 `00000000001073741824` 第二个文件，物理偏移量为1G `00000000002147483648` 第三个文件，物理偏移量为2G `00000000003221225472` 第四个文件，物理偏移量为3G 1.2 consumequeue包含该Broker上所有的Topic对应的消息队列文件，每个消费队列其实是Commitlog的一个索引，提供给消费者做拉取消息，更新位点使用。 1.3 Index索引文件，文件是按照消息的key创建的Hash索引。 1.4 Config这个目录保存了当前Broker的所有Topic，订阅关系和消费进度，并会持久化到磁盘，以供宕机之后恢复工作。 1.5 abortBroker正常关闭的时候会删除此文件，异常关闭则不会，所以当Broker重新启动时根据此文件决定是否需要重建Index等操作。 1.6 cheekpointBroker最近一次正常运行时的状态，最后一次刷盘的时间，最后一次正确索引的时间等。 2 Broker存储机制存储机制是RockerMQ的核心，也是亮点设计，这将决定写入和查询的效率。 存储消息的步骤如下 2.1 Broker接收客户端发送消息的请求并作预处理org.apache.rocketmq.broker.processor.SendMessageProcessor#processRequest会将把接收到的请求转化为消息实例。这个方法又分为四步 解析请求参数 执行发送消息前的Hook 调用保存方法存储消息 执行发送后的Hook 2.2 Broker存储前与处理消息org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage做预处理操作。 2.2.1 设置请求的处理的返回对象标志 final RemotingCommand response = preSend(ctx, request, requestHeader); final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class); response.setOpaque(request.getOpaque()); 2.2.2 校验存储前会做一系列的校验 校验Broker是否配置可写 校验Topic名字是否默认值 校验Topic配置是否存在 校验queueId与读写队列是否匹配 校验Broker是都支持事务消息 2.3 进行消息校验和存储模块检查保存消息之前，还会对消息本身进行校验 校验存储模块是否关闭 校验Broker是否Slave 校验存储模块运行标记 校验Topic长度 校验扩展信息的长度 校验操作系统Page Cache是否繁忙 2.4 存储消息核心过程如下 设置消息保存时间为当前时间戳 延迟消息特殊处理，保存原来的topic和queueid，并将topic改为SCHEDULE_TOPIC_XXXX. 获取最后一个CommitLog文件 校验最后一个MappedFile，如果为空或已经写满，则新建一个MappedFile 调用appendMessage()方法，将消息写入MappedFile。写入的方法具体步骤如下： 2.4.1 查找Offset查找即将写入的物理Offset 2.4.2 事务消息处理针对事务消息单独处理，将消息的queueOffset设置为0 2.4.3 序列化消息这里会将消息序列化并保存到ByteBuffer中。 2.4.4 更新消息位点最后会更新消息所在Queue的位点 在消息存储完成后，会处理刷盘逻辑和主从同步逻辑。 |
【本文地址】
今日新闻 |
推荐新闻 |