全网最细RocketMQ源码四：消息存储

2024-01-16 02:36
文章标签 源码 全网 存储 消息 rocketmq 最细

本文主要是介绍全网最细RocketMQ源码四：消息存储，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

看完上一章之后，有没有很好奇，生产者发送完消息之后，server是如何存储，这一章节就来学习

入口

SendMessageProcessor.processRequest
在这里插入图片描述
在这里插入图片描述

  private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, RemotingCommand request,SendMessageContext mqtraceContext,SendMessageRequestHeader requestHeader) {final RemotingCommand response = preSend(ctx, request, requestHeader);final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();if (response.getCode() != -1) {return CompletableFuture.completedFuture(response);}final byte[] body = request.getBody();int queueIdInt = requestHeader.getQueueId();TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());if (queueIdInt < 0) {queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());}MessageExtBrokerInner msgInner = new MessageExtBrokerInner();msgInner.setTopic(requestHeader.getTopic());msgInner.setQueueId(queueIdInt);if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig)) {return CompletableFuture.completedFuture(response);}msgInner.setBody(body);msgInner.setFlag(requestHeader.getFlag());MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));msgInner.setPropertiesString(requestHeader.getProperties());msgInner.setBornTimestamp(requestHeader.getBornTimestamp());msgInner.setBornHost(ctx.channel().remoteAddress());msgInner.setStoreHost(this.getStoreHost());msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName);msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));CompletableFuture<PutMessageResult> putMessageResult = null;Map<String, String> origProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);if (transFlag != null && Boolean.parseBoolean(transFlag)) {if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {response.setCode(ResponseCode.NO_PERMISSION);response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()+ "] sending transaction message is forbidden");return CompletableFuture.completedFuture(response);}putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);} else {// 使用defaultMessageStore.aysncPutMessage存储putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);}return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);}

实际真正的负责存储就是DefaultMessageStore, 不过在讲述DefaultMessageStore的时候，我们是自底往上学，因为DefaultMessageStore比较复杂，从顶往下学容易学乱。先从地基开始，然后再看高楼大厦

MappedFile

public class MappedFile extends ReferenceResource {// 内存页大小：4kpublic static final int OS_PAGE_SIZE = 1024 * 4;protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);// 当前进程下 所有的 mappedFile占用的总虚拟内存大小private static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0);// 当前进程下 所有的 mappedFile个数private static final AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0);// 当前mappedFile数据写入点protected final AtomicInteger wrotePosition = new AtomicInteger(0);protected final AtomicInteger committedPosition = new AtomicInteger(0);// 当前mappedFIle数据落盘位点（flushedPosition 之前的数据 都是安全数据，flushedPosition~wrotePosition之间的数据 属于脏页）private final AtomicInteger flushedPosition = new AtomicInteger(0);// 文件大小protected int fileSize;// 文件通道protected FileChannel fileChannel;/*** Message will put to here first, and then reput to FileChannel if writeBuffer is not null.*/protected ByteBuffer writeBuffer = null;protected TransientStorePool transientStorePool = null;// 文件名称（commitLog ConsumeQueue：文件名就是 第一条消息的 物理偏移量   索引文件： 年月日小时分钟秒.. ）private String fileName;// 文件名转longprivate long fileFromOffset;// 文件对象private File file;// 内存映射缓冲区，访问虚拟内存private MappedByteBuffer mappedByteBuffer;// 该文件下 保存的第一条 msg 的存储时间private volatile long storeTimestamp = 0;// 当前文件如果是 目录内 有效文件的 首文件的话，该值为trueprivate boolean firstCreateInQueue = false;

  • 构造方法
    在这里插入图片描述
    在这里插入图片描述

  • appendMessage方法
    在这里插入图片描述
    在这里插入图片描述

  • appendMessage(byte[] data)
    在这里插入图片描述

  • flush
    在这里插入图片描述

MappedFileQueue

public class MappedFileQueue {private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);private static final InternalLogger LOG_ERROR = InternalLoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);private static final int DELETE_FILES_BATCH_MAX = 10;// mfq 管理的目录（CommitLog： ../store/commitlog  或者  consumeQueue： ../store/xxx_topic/0）private final String storePath;// 目录下每个文件大小（commitLog文件 默认 1g     consumeQueue 文件 默认 600w字节）private final int mappedFileSize;// list，目录下的每个 mappedFile 都加入该listprivate final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();// 创建mappedFile的服务，内部有自己的线程，咱们通过向他提交 request ，内部线程处理完后 会返回给我们结果  结果 就是 mappedFile对象。private final AllocateMappedFileService allocateMappedFileService;// 目录的刷盘位点（它的值： curMappedFile.fileName + curMappedFile.wrotePosition）private long flushedWhere = 0;private long committedWhere = 0;// 当前目录下最后一条msg存储时间private volatile long storeTimestamp = 0;

  • load方法
    在这里插入图片描述

  • getLastMappedFile

 /*** 参数1：startOffset ，文件起始偏移量* 参数2：needCreate ，当 list 为空时，是否创建mappedFile*/public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {// 该值控制是否创建 mappedFile，当需要创建mappedFile时，它充当文件名的结尾// 两种情况会创建：// 1. list内没有mappedFile// 2. list最后一个 mappedFile (当前顺序写的mappedFile) 它写满了..long createOffset = -1;MappedFile mappedFileLast = getLastMappedFile();if (mappedFileLast == null) {// 情况1  list内没有mappedFile// createOffset 取值 必须是 mappedFileSize 的倍数 或者 0createOffset = startOffset - (startOffset % this.mappedFileSize);}if (mappedFileLast != null && mappedFileLast.isFull()) { // 情况2  list最后一个 mappedFile (当前顺序写的mappedFile) 它写满了..// 上一个文件名 转long + mappedFileSizecreateOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;}if (createOffset != -1 && needCreate) {// 这里面是创建 新的 mappedFile 的逻辑。// 获取待创建文件的 绝对路径(下次即将要创建的文件名)String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);// 获取 下下次 要创建的文件的 绝对路径String nextNextFilePath = this.storePath + File.separator+ UtilAll.offset2FileName(createOffset + this.mappedFileSize);MappedFile mappedFile = null;if (this.allocateMappedFileService != null) {// 创建mappedFile的服务，内部有自己的线程，咱们通过向他提交 request ，// 内部线程处理完后 会返回给我们结果  结果 就是 mappedFile对象。// 当mappedFileSize >= 1g 的话，这里创建的mappedFile 会执行它的 预热方法。mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,nextNextFilePath, this.mappedFileSize);} else {try {mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);} catch (IOException e) {log.error("create mappedFile exception", e);}}if (mappedFile != null) {if (this.mappedFiles.isEmpty()) {mappedFile.setFirstCreateInQueue(true);}this.mappedFiles.add(mappedFile);}return mappedFile;}return mappedFileLast;}
  • flush
 /*** @param flushLeastPages (0 表示强制刷新， > 0 脏页数据必须达到 flushLeastPages 才刷新)* @return boolean true 表示本次刷盘无数据落盘   false 表示本次刷盘有数据落盘*/public boolean flush(final int flushLeastPages) {boolean result = true;// 获取当前正在刷盘的文件 （正在顺序写的mappedFile）MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, this.flushedWhere == 0);if (mappedFile != null) {// 获取mappedFile 最后一条消息的存储时间long tmpTimeStamp = mappedFile.getStoreTimestamp();// 调用mf 刷盘 ,返回mf的最新的落盘位点int offset = mappedFile.flush(flushLeastPages);// mf起始偏移量 + mf最新的落盘位点long where = mappedFile.getFileFromOffset() + offset;// true 表示本次刷盘无数据落盘   false 表示本次刷盘有数据落盘result = where == this.flushedWhere;// 将最新的目录刷盘位点 赋值给 flushedWherethis.flushedWhere = where;if (0 == flushLeastPages) {this.storeTimestamp = tmpTimeStamp;}}return result;}

CommitLog

ConsumeQueue

DefaultMessageStore

这篇关于全网最细RocketMQ源码四：消息存储的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/611042。

相关文章

异构存储（冷热数据分离）

异构存储（冷热数据分离）

异构存储主要解决不同的数据，存储在不同类型的硬盘中，达到最佳性能的问题。 异构存储Shell操作 （1）查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies （2）为指定路径（数据存储目录）设置指定的存储策略 hdfs storagepolicies -setStoragePo
阅读更多...
HDFS—存储优化（纠删码）

HDFS—存储优化（纠删码）

纠删码原理 HDFS 默认情况下，一个文件有3个副本，这样提高了数据的可靠性，但也带来了2倍的冗余开销。 Hadoop3.x 引入了纠删码，采用计算的方式，可以节省约50％左右的存储空间。 此种方式节约了空间，但是会增加 cpu 的计算。 纠删码策略是给具体一个路径设置。所有往此路径下存储的文件，都会执行此策略。 默认只开启对 RS-6-3-1024k
阅读更多...
字节面试 | 如何测试RocketMQ、RocketMQ？

字节面试 | 如何测试RocketMQ、RocketMQ？

字节面试：RocketMQ是怎么测试的呢？ 答： 首先保证消息的消费正确、设计逆向用例，在验证消息内容为空等情况时的消费正确性； 推送大批量MQ，通过Admin控制台查看MQ消费的情况，是否出现消费假死、TPS是否正常等等问题。（上述都是临场发挥，但是RocketMQ真正的测试点，还真的需要探讨） 01 先了解RocketMQ 作为测试也是要简单了解RocketMQ。简单来说，就是一个分
阅读更多...
JAVA智听未来一站式有声阅读平台听书系统小程序源码

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来，一站式有声阅读平台听书系统 🌟&nbsp;开篇：遇见未来，从“智听”开始 在这个快节奏的时代，你是否渴望在忙碌的间隙，找到一片属于自己的宁静角落？是否梦想着能随时随地，沉浸在知识的海洋，或是故事的奇幻世界里？今天，就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统，它正悄悄改变着我们的阅读方式，让未来触手可及！ 📚&nbsp;第一站：海量资源，应有尽有 走进“智听
阅读更多...
Java ArrayList扩容机制 （源码解读）

Java ArrayList扩容机制 （源码解读）

结论：初始长度为10，若所需长度小于1.5倍原长度，则按照1.5倍扩容。若不够用则按照所需长度扩容。 一. 明确类内部重要变量含义         1：数组默认长度         2:这是一个共享的空数组实例，用于明确创建长度为0时的ArrayList ，比如通过 new ArrayList<>(0)，ArrayList 内部的数组 elementData 会指向这个 EMPTY_EL
阅读更多...
如何在Visual Studio中调试.NET源码

如何在Visual Studio中调试.NET源码

今天偶然在看别人代码时，发现在他的代码里使用了Any判断List<T>是否为空。 我一般的做法是先判断是否为null，再判断Count。 看了一下Count的源码如下： 1 [__DynamicallyInvokable]2 public int Count3 {4 [__DynamicallyInvokable]5 get
阅读更多...
工厂ERP管理系统实现源码（JAVA）

工厂ERP管理系统实现源码（JAVA）

工厂进销存管理系统是一个集采购管理、仓库管理、生产管理和销售管理于一体的综合解决方案。该系统旨在帮助企业优化流程、提高效率、降低成本，并实时掌握各环节的运营状况。 在采购管理方面，系统能够处理采购订单、供应商管理和采购入库等流程，确保采购过程的透明和高效。仓库管理方面，实现库存的精准管理，包括入库、出库、盘点等操作，确保库存数据的准确性和实时性。 生产管理模块则涵盖了生产计划制定、物料需求计划、
阅读更多...
Spring 源码解读：自定义实现Bean定义的注册与解析

Spring 源码解读：自定义实现Bean定义的注册与解析

引言 在Spring框架中，Bean的注册与解析是整个依赖注入流程的核心步骤。通过Bean定义，Spring容器知道如何创建、配置和管理每个Bean实例。本篇文章将通过实现一个简化版的Bean定义注册与解析机制，帮助你理解Spring框架背后的设计逻辑。我们还将对比Spring中的BeanDefinition和BeanDefinitionRegistry，以全面掌握Bean注册和解析的核心原理。
阅读更多...
音视频入门基础：WAV专题（10）——FFmpeg源码中计算WAV音频文件每个packet的pts、dts的实现

音视频入门基础：WAV专题（10）——FFmpeg源码中计算WAV音频文件每个packet的pts、dts的实现

一、引言 从文章《音视频入门基础：WAV专题（6）——通过FFprobe显示WAV音频文件每个数据包的信息》中我们可以知道，通过FFprobe命令可以打印WAV音频文件每个packet（也称为数据包或多媒体包）的信息，这些信息包含该packet的pts、dts： 打印出来的“pts”实际是AVPacket结构体中的成员变量pts，是以AVStream->time_base为单位的显
阅读更多...
速了解MySQL 数据库不同存储引擎

速了解MySQL 数据库不同存储引擎

快速了解MySQL 数据库不同存储引擎 MySQL 提供了多种存储引擎，每种存储引擎都有其特定的特性和适用场景。了解这些存储引擎的特性，有助于在设计数据库时做出合理的选择。以下是 MySQL 中几种常用存储引擎的详细介绍。 1. InnoDB 特点： 事务支持：InnoDB 是一个支持 ACID（原子性、一致性、隔离性、持久性）事务的存储引擎。行级锁：使用行级锁来提高并发性，减少锁竞争
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2