ip池源码

❶ 学python最想要提升的是哪些地方

1.学习 Python 包并实现基本的爬虫过程
大部分爬虫都是按“发送请求——获得页面——解析页面——抽取并储存内容”这样的流程来进行，这其实也是模拟了我们使用浏览器获取网页信息的过程。Python中爬虫相关的包很多：urllib、requests、bs4、scrapy、pyspider 等，建议从requests+Xpath 开始，requests 负责连接网站，返回网页，Xpath 用于解析网页，便于抽取数据。
如果你用过 BeautifulSoup，会发现 Xpath 要省事不少，一层一层检查元素代码的工作，全都省略了。这样下来基本套路都差不多，一般的静态网站根本不在话下。当然如果你需要爬取异步加载的网站，可以学习浏览器抓包分析真实请求或者学习Selenium来实现自动化。
2.了解非结构化数据的存储
爬回来的数据可以直接用文档形式存在本地，也可以存入数据库中。开始数据量不大的时候，你可以直接通过 Python 的语法或 pandas 的方法将数据存为csv这样的文件。当然你可能发现爬回来的数据并不是干净的，可能会有缺失、错误等等，你还需要对数据进行清洗，可以学习 pandas 包的基本用法来做数据的预处理，得到更干净的数据。
3.学习scrapy，搭建工程化爬虫
掌握前面的技术一般量级的数据和代码基本没有问题了，但是在遇到非常复杂的情况，可能仍然会力不从心，这个时候，强大的 scrapy 框架就非常有用了。scrapy 是一个功能非常强大的爬虫框架，它不仅能便捷地构建request，还有强大的 selector 能够方便地解析 response，然而它最让人惊喜的还是它超高的性能，让你可以将爬虫工程化、模块化。学会 scrapy，你可以自己去搭建一些爬虫框架，你就基本具备Python爬虫工程师的思维了。
4.学习数据库知识，应对大规模数据存储与提取
Python客栈送红包、纸质书
爬回来的数据量小的时候，你可以用文档的形式来存储，一旦数据量大了，这就有点行不通了。所以掌握一种数据库是必须的，学习目前比较主流的 MongoDB 就OK。MongoDB 可以方便你去存储一些非结构化的数据，比如各种评论的文本，图片的链接等等。你也可以利用PyMongo，更方便地在Python中操作MongoDB。因为这里要用到的数据库知识其实非常简单，主要是数据如何入库、如何进行提取，在需要的时候再学习就行。
5.掌握各种技巧，应对特殊网站的反爬措施
当然，爬虫过程中也会经历一些绝望啊，比如被网站封IP、比如各种奇怪的验证码、userAgent访问限制、各种动态加载等等。遇到这些反爬虫的手段，当然还需要一些高级的技巧来应对，常规的比如访问频率控制、使用代理IP池、抓包、验证码的OCR处理等等。往往网站在高效开发和反爬虫之间会偏向前者，这也为爬虫提供了空间，掌握这些应对反爬虫的技巧，绝大部分的网站已经难不到你了。
6.分布式爬虫，实现大规模并发采集，提升效率
爬取基本数据已经不是问题了，你的瓶颈会集中到爬取海量数据的效率。这个时候，相信你会很自然地接触到一个很厉害的名字：分布式爬虫。分布式这个东西，听起来很恐怖，但其实就是利用多线程的原理让多个爬虫同时工作，需要你掌握Scrapy+ MongoDB + Redis 这三种工具。Scrapy 前面我们说过了，用于做基本的页面爬取，MongoDB 用于存储爬取的数据，Redis 则用来存储要爬取的网页队列，也就是任务队列。所以有些东西看起来很吓人，但其实分解开来，也不过如此。当你能够写分布式的爬虫的时候，那么你可以去尝试打造一些基本的爬虫架构了，实现一些更加自动化的数据获取。
只要按照以上的Python爬虫学习路线，一步步完成，即使是新手小白也能成为老司机，而且学下来会非常轻松顺畅。所以新手在一开始的时候，尽量不要系统地去啃一些东西，找一个实际的项目，直接开始操作。
其实学Python编程和练武功其实很相似，入门大致这样几步:找本靠谱的书,找个靠谱的师傅，找一个地方开始练习。
学语言也是这样的：选一本通俗易懂的书，找一个好的视频资料，然后自己装一个IDE工具开始边学边写。
7.给初学Python编程者的建议：
①信心。可能你看了视频也没在屏幕上做出点啥，都没能把程序运行起来。但是要有自信，所有人都是这样过来的。
②选择适合自己的教程。有很早的书籍很经典，但是不是很适合你，很多书籍是我们学过一遍Python之后才会发挥很大作用。
③写代码，就是不断地写，练。这不用多说，学习什么语言都是这样。总看视频，编不出东西。可以从书上的小案例开始写，之后再写完整的项目。
④除了学Python，计算机的基础也要懂得很多，补一些英语知识也行。
⑤不但会写，而且会看，看源码是一个本领，调试代码更是一个本领，就是解决问题的能力，挑错。理解你自己的报错信息，自己去解决。
⑥当你到达了一个水平，就多去看官方的文档，在CSDN上面找下有关Python的博文或者群多去交流。
希望想学习Python的利用好现在的时间，管理好自己的学习时间，有效率地学习Python，Python这门语言可以做很多事情。

❷ lwip源码解析

lwip源码解析

lwip 是 TCP/IP 协议栈的轻量化实现，它在嵌入式平台上广泛应用，尤其在资源有限的 MCU 设备上。lwip 的体积小巧，运行内存需求仅几十 KB，支持裸机移植和操作系统移植。

lwip 提供了三种接口类型：raw api、netconn api 和 socket api。raw api 是基于事件驱动，以回调函数形式实现，适用于裸机环境。netconn api 是顺序 API，需要线程机制支持，适用于操作系统环境。socket api 同样是顺序 API，基于 netconn api 重新封装，以兼容 POSIX 标准。

在 lwip 的内存管理方面，它有自己的动态内存管理机制，包括简单的内存管理和基于内存池的内存管理。简单内存管理通过 mem 模块实现，基于全局数组划分内存区域，用户通过 mem_init 初始化堆区，并在分配和释放内存时，通过查找和合并空闲内存块进行操作。基于内存池的内存管理使用 memp 模块，通过内存池分配和释放内存，内存池由多个固定大小的内存块构成，适用于特定类型的数据存储。

lwip 使用 pbuf 结构体对数据进行封装和协议栈内部传递，pbuf 的重要性与 linux 内核协议栈中的 skb_buff 类似。pbuf 可以是 PBUF_RAM、PBUF_POOL、PBUF_REF 或 PBUF_ROM 类型，分别用于不同内存管理和数据引用场景。

在数据收发过程中，以 UDP 为例，lwip 通过注册回调函数实现 UDP 接收，通过创建 pbuf 和调用相关函数实现 UDP 发送。发送时，lwip 会根据目的 IP 地址查找路由、绑定本地 IP 和端口、添加 UDP 和 IP 头部信息，并最终通过网卡驱动发送数据。接收时，数据通过注册的回调函数以 pbuf 形式传递给协议栈，解析以太网帧头部后，根据协议类型调用不同接口进行进一步处理。

在 TCP 协议的实现中，lwip 通过监听和接收回调函数管理连接状态，处理 SYN 报文时，会生成初始化序列号，进行序列号与确认机制，并通过计算 MSS 和发送窗口管理数据收发过程中的数据管理。在接收过程，lwip 会接收 TCP 报文并处理，计算接收窗口裁剪数据，并通过接收数据缓冲区进行应用层的数据处理。当出现超时重传时，lwip 会更新重传定时器、计算 RTT 和 RTO，并在快重传和快恢复模式下调整拥塞窗口。

❸ MySQL与Redis数据库连接池介绍（图示+源码+代码演示）

数据库连接池（Connection pooling）是程序启动时建立足够的数据库连接，并将这些连接组成一个连接池，由程序动态地对池中的连接进行申请，使用，释放。

简单的说：创建数据库连接是一个很耗时的操作，也容易对数据库造成安全隐患。所以，在程序初始化的时候，集中创建多个数据库连接，并把他们集中管理，供程序使用，可以保证较快的数据库读写速度，还更加安全可靠。

不使用数据库连接池

如果不使用数据库连接池，对于每一次SQL操作，都要走一遍下面完整的流程：

1.TCP建立连接的三次握手（客户端与 MySQL服务器的连接基于TCP协议）

2.MySQL认证的三次我收

3.真正的SQL执行

4.MySQL的关闭

5.TCP的四次握手关闭

可以看出来，为了执行一条SQL，需要进行大量的初始化与关闭操作

使用数据库连接池

如果使用数据库连接池，那么会 事先申请（初始化）好 相关的数据库连接，然后在之后的SQL操作中会复用这些数据库连接，操作结束之后数据库也不会断开连接，而是将数据库对象放回到数据库连接池中

资源重用：由于数据库连接得到重用，避免了频繁的创建、释放连接引起的性能开销，在减少系统消耗的基础上，另一方面也增进了系统运行环境的平稳性（减少内存碎片以及数据库临时进程/线程的数量）。

更快的系统响应速度：数据库连接池在初始化过程中，往往已经创建了若干数据库连接置于池中备用。 此时连接的初始化工作均已完成。对于业务请求处理而言，直接利用现有可用连接，避免了从数据库连接初始化和释放过程的开销，从而缩减了系统整体响应时间。

统一的连接管理，避免数据库连接泄露：在较为完备的数据库连接池实现中，可根据预先的连接占用超时设定，强制收回被占用连接。从而避免了常规数据库连接操作中可能出现的资源泄露。

如果说你的服务器CPU是4核i7的，连接池大小应该为((4*2)+1)=9

相关视频推荐

90分钟搞懂数据库连接池技术|linux后台开发

《tcp/ip详解卷一》： 150行代码拉开协议栈实现的篇章

学习地址：C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂

需要C/C++ Linux服务器架构师学习资料加qun 812855908 获取（资料包括 C/C++，Linux，golang技术，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg 等），免费分享

源码下载

下载方式：https://github.com/dongyusheng/csdn-code/tree/master/db_pool（Github中下载）

db_pool目录下有两个目录，mysql_pool目录为MySQL连接池代码，redis_pool为redis连接池代码

下面介绍mysql_pool

CDBConn解析

概念： 代表一个数据连接对象实例

相关成员：

m_pDBPool：该数据库连接对象所属的数据库连接池

构造函数： 绑定自己所属于哪个数据库连接池

Init()函数： 创建数据库连接句柄

CDBPool解析

概念：代表一个数据库连接池

相关成员：

Init()函数：常见指定数量的数据库实例句柄，然后添加到m_free_list中，供后面使用

GetDBConn()函数： 用于从空闲队列中返回可以使用的数据库连接句柄

RelDBConn()函数： 程序使用完该数据库句柄之后，将句柄放回到空闲队列中

测试之前，将代码中的数据库地址、端口、账号密码等改为自己的（代码中有好几处）

进入MySQL， 创建mysql_pool_test数据库

进入到mysql_pool目录下， 创建一个build目录并进入 ：

然后输入如下的命令进行编译

之后就会在目录下生成如下的可执行文件

输入如下两条命令进行测试： 可以看到不使用数据库连接池，整个操作耗时4秒左右；使用连接池之后，整个操作耗时2秒左右，提升了一倍

源码下载

下面介绍redis_pool

测试

进入到redis_pool目录下， 创建一个build目录并进入 ：

然后输入如下的命令进行编译

之后就会在目录下生成如下的可执行文件

输入如下的命令进行测试： 可以看到不使用数据库连接池，整个操作耗时182ms；使用连接池之后，整个操作耗时21ms，提升了很多

进入redis，可以看到我们新建的key：

❹ RocketMQ源码分析4：Broker处理消息流程

基于RocketMQ-4.9.0 版本分析rocketmq

1.接收和处理请求

其实这里和NameServer处理请求的过程是一样的。 在前面Broker启动过程文章中,我们知道Broker启动时,最终会启动一个netty服务，可以接收读写请求。那么这篇文章我们就来看看他是如何接收和处理请求的。

在前面分析Broker启动的过程中，我们通过源码看到，netty服务端启动类会绑定很多ChannelHandler，有负责处理握手的，有负责处理心跳的，有负责处理连接的，也有负责读写的，其中NettyServerHandler就是负责读写的。 我们在简单看下代码：

//TODO: ServerBootstrap 是netty的启动类ServerBootstrap childHandler =this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector).channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024).option(ChannelOption.SO_REUSEADDR, true).option(ChannelOption.SO_KEEPALIVE, false).childOption(ChannelOption.TCP_NODELAY, true).childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize()).childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())//TODO:绑定ip和端口.localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort())).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler).addLast(defaultEventExecutorGroup,encoder,new NettyDecoder(),//TODO:处理心跳的ChannelHandlernew IdleStateHandler(0, 0, nettyServerConfig.()),//TODO:处理连接的 ,//TODO: 处理读写的 ChannelHandler (NettyServerHandler)serverHandler);}});

broker开放10911端口用来与procer和consumer进行通信。 所以，NettyServerHandler?类就是我们分析的入口: 它是?NettyRemotingServer?类的内部类

@Sharableclass NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> { @Overrideprotected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {processMessageReceived(ctx, msg);}

}

> 看到它是不是很熟悉？没错，`NameServer`处理请求也是它（不过他们是分别new的不同对象)然后我们继续看它的内部实现：```javapublic void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {//TODO:从处理器表中获取Pair对象final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;final int opaque = cmd.getOpaque();if (pair != null) {//TODO:构建一个线程，当submit到线程池中时会执行//TODO: 先不要展开看这里，继续往后看，会将该run 提交到线程池中Runnable run = new Runnable() {@Overridepublic void run() {try {//TODO:...省略部分代码...... //TODO:核心if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();processor.asyncProcessRequest(ctx, cmd, callback);} else {NettyRequestProcessor processor = pair.getObject1();RemotingCommand response = processor.processRequest(ctx, cmd);callback.callback(response);}} catch (Throwable e) { //TODO:....省略catch......}}};//TODO: ....省略部分代码.......try {//TODO:将上面创建的Runnable放入线程池中然后执行final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);pair.getObject2().submit(requestTask);} catch (RejectedExecutionException e) {//TODO:省略catch代码.......}} else {//TODO:...省略else......}}

总结一下就主要做两件事，获取处理器，然后调用处理器的处理方法。

获取处理器

final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;

从处理器表processorTable中根据code获取处理器,在上一篇文章Broker启动-注册处理器中,它会注册处理器到这个处理器表中，这个code 就是用来区分业务场景的。比如生产者发送消息的code=RequestCode.SEND_MESSAGE,消费者拉取消息的code=RequestCode.PULL_MESSAGE。

我们这里先不关注code,总之就是从处理器表processorTable中获取一个处理器。这里提一嘴，在分析NameServer处理请求的时候，它也会注册处理器，不过它不会向这个处理器表中注册处理器，而是使用一个默认的处理器。

调用处理器的处理方法

if (pair.getObject1() instanceof AsyncNettyRequestProcessor) { AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1(); processor.asyncProcessRequest(ctx, cmd, callback);} else { NettyRequestProcessor processor = pair.getObject1(); RemotingCommand response = processor.processRequest(ctx, cmd); callback.callback(response);}

在调用处理方法之前，我们要知道这个Pair对象是什么? 在前面Broker启动-注册处理器一文中，我们发现，它会new 一个 Pair对象放入到处理器表中，作为Map的value,而Pair对象有两个泛型T1,T2,分别对应着处理器和线程池。

public class Pair<T1, T2> { private T1 object1; private T2 object2; public Pair(T1 object1, T2 object2) { this.object1 = object1; this.object2 = object2; }}

object1=业务处理器，object2=线程池 那么接下来就是进入处理器开始处理业务逻辑了，那么我们就以接收procer发送消息的处理器SendMessageProcessor为例来往下看 从类的继承结构上看，它继承了 AsyncNettyRequestProcessor, 处理器基本上都继承了它，所以走的是异步逻辑(也就是if逻辑）

SendMessageProcessor#asyncProcessRequest(..)

public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {final SendMessageContext mqtraceContext;switch (request.getCode()) {//TODO: 消费失败，发起重试，则会来到这里case RequestCode.CONSUMER_SEND_MSG_BACK:return this.asyncConsumerSendMsgBack(ctx, request);default:SendMessageRequestHeader requestHeader = parseRequestHeader(request);if (requestHeader == null) {return CompletableFuture.completedFuture(null);}mqtraceContext = buildMsgContext(ctx, requestHeader);this.executeSendMessageHookBefore(ctx, request, mqtraceContext);if (requestHeader.isBatch()) {//TODO: 批量消息return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);} else {//TODO: 普通消息发送就会走这里return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);}}}

接下来我们就具体分析Broker是如何一步一步将消息存储起来的。

2.消息存储

处理procer发送消息的处理器是 SendMessageProcessor。

生产者可以发送普通消息，顺序消息，延迟消息，事务消息，批量消息，我们就以普通消息为例来分析。

private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, RemotingCommand request,SendMessageContext mqtraceContext,SendMessageRequestHeader requestHeader) {//TODO:预发送final RemotingCommand response = preSend(ctx, request, requestHeader);final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();//TODO:....省略部分代码......//TODO:如果没有指定queueid,则随机指定一个if (queueIdInt < 0) {queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());}//TODO:构建消息体对象，保存topic,queueid,msg等内容MessageExtBrokerInner msgInner = new MessageExtBrokerInner();msgInner.setTopic(requestHeader.getTopic());msgInner.setQueueId(queueIdInt);msgInner.setBody(body);//TODO:.......other info......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)) {//TODO: 事务消息putMessageResult = this.brokerController.().asyncPrepareMessage(msgInner);} else {//TODO: 普通消息putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);}return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);

}

@Sharableclass NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {0

那么我们继续点进去CommitLog的内部实现：

代码是非常的多，还是只保留关注的部分

@Sharableclass NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {1

代码比较多，我们简单总结下：

判断是否为延迟消息，如果是延迟消息，则将topic替换为SCHEDULE_TOPIC_XXXX,将queueid替换为(延迟级别-1),将原始的topic和queueid保存到消息对象的Properties属性中。

获取最新的MappedFile文件对象。这个MappedFile在这里可以理解为是消息文件的逻辑映射，然后调用MappedFile对象的方法put消息.

提交刷盘请求

同步刷盘：只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Procer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多

异步刷盘：能够充分利用OS的PageCache的优势，只要消息写入PageCache即可将成功的ACK返回给Procer端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了MQ的性能和吞吐量。

所以我们继续看第2点，MappedFile它是如何Put消息的：

@Sharableclass NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {2

然后我们继续看cb.doAppend(....)方法，看它如何存储消息：

@Sharableclass NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {3

这个方法的内容也比较多，我们还是简单总结下：

设置消息的msgId(说明msgId是broker端生成的)

计算consumequeue的offset(在broker启动时，会加载配置文件中的内容到这个Map中)

在消息分发时，会根据这个offset值按顺序写入到索引文件中。

判断消息是否超过4M,如果超过4M则返回一个MESSAGE_SIZE_EXCEEDED错误码。

单个消息大小限制是4M,procer端会check, broker端会check,这个值可以修改，但注意要修改两端。

判断当前文件是否还写的下当前消息，如果写不下，则新建一个MappedFile,再次写入

初始化存储空间，写入一个消息体的全部内容，然后放入缓冲区。消息体内容大致如图：

消息写入到commitlog后，这条消息的处理流程就结束了。然后接下来就是消息分发（构建消息索引）

关于消息存储，我们看下官网的文章阐述：

消息存储整体架构

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

(1) CommitLog：消息主体以及元数据的存储主体，存储Procer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G, 文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；

(2) ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；

(3) IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$home/store/index${fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

在上面的RocketMQ的消息存储整体架构图中可以看出，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Procer和Consumer分别采用了数据和索引部分相分离的存储结构，Procer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Procer发送的消息就不会丢失。正因为如