Kafka和RocketMQ底层存储之那些你不知道的事

大家好,我是yes。

我们都知道 RocketMQ 和 Kafka 消息都是存在磁盘中的,那为什么消息存磁盘读写还可以这么快?有没有做了什么优化?都是存磁盘它们两者的实现之间有什么区别么?各自有什么优缺点?

今天我们就来一探究竟。

先说下快的主要原因就是顺序读写、mmap、sendfile。我们先来看看这几点,然后再盘一下 RocketMQ 和 Kafka是如何应用的。

存储介质-磁盘

一般而言消息中间件的消息都存储在本地文件中,因为从效率来看直接放本地文件是最快的,并且稳定性最高。毕竟要是放类似数据库等第三方存储中的话,就多一个依赖少一份安全,并且还有网络的开销。

那对于将消息存入磁盘文件来说一个流程的瓶颈就是磁盘的写入和读取。我们知道磁盘相对而言读写速度较慢,那通过磁盘作为存储介质如何实现高吞吐呢?

顺序读写

答案就是顺序读写

首先了解一下页缓存,页缓存是操作系统用来作为磁盘的一种缓存,减少磁盘的I/O操作。

在写入磁盘的时候其实是写入页缓存中,使得对磁盘的写入变成对内存的写入。写入的页变成脏页,然后操作系统会在合适的时候将脏页写入磁盘中。

在读取的时候如果页缓存命中则直接返回,如果页缓存 miss 则产生缺页中断,从磁盘加载数据至页缓存中,然后返回数据。

并且在读的时候会预读,根据局部性原理当读取的时候会把相邻的磁盘块读入页缓存中。在写入的时候会后写,写入的也是页缓存,这样存着可以将一些小的写入操作合并成大的写入,然后再刷盘。

而且根据磁盘的构造,顺序 I/O 的时候,磁头几乎不用换道,或者换道的时间很短。

根据网上的一些测试结果,顺序写盘的速度比随机写内存还要快。

当然这样的写入存在数据丢失的风险,例如机器突然断电,那些还未刷盘的脏页就丢失了。不过可以调用 fsync 强制刷盘,但是这样对于性能的损耗较大。

因此一般建议通过多副本机制来保证消息的可靠,而不是同步刷盘

可以看到顺序 I/O 适应磁盘的构造,并且还有预读和后写。 RocketMQ 和 Kafka 都是顺序写入和近似顺序读取。它们都采用文件追加的方式来写入消息,只能在日志文件尾部写入新的消息,老的消息无法更改。

mmap-文件内存映射

从上面可知访问磁盘文件会将数据加载到页缓存中,但是页缓存属于内核空间,用户空间访问不了,因此数据还需要拷贝到用户空间缓冲区。

可以看到数据需要从页缓存再经过一次拷贝程序才能访问的到,因此还可以通过mmap来做一波优化,利用内存映射文件来避免拷贝。

简单的说文件映射就是将程序虚拟页面直接映射到页缓存上,这样就无需有内核态再往用户态的拷贝,而且也避免了重复数据的产生。并且也不必再通过调用readwrite方法对文件进行读写,可以通过映射地址加偏移量的方式直接操作

sendfile-零拷贝

既然消息是存在磁盘中的,那消费者来拉消息的时候就得从磁盘拿。我们先来看看一般发送文件的流程是如何的。

简单说下DMA是什么,全称 Direct Memory Access ,它可以独立地直接读写系统内存,不需要 CPU 介入,像显卡、网卡之类都会用DMA

可以看到数据其实是冗余的,那我们来看看mmap之后的发送文件流程是怎样的。

可以看到上下文切换的次数没有变化,但是数据少拷贝一份,这和我们上文提到的mmap能达到的效果是一样的。

但是数据还是冗余了一份,这不是可以直接把数据从页缓存拷贝到网卡不就好了嘛?sendfile就有这个功效。我们先来看看Linux2.1版本中的sendfile

因为就一个系统调用就满足了发送的需求,相比 read + write 或者 mmap + write 上下文切换肯定是少了的,但是好像数据还是有冗余啊。是的,因此 Linux2.4 版本的 sendfile + 带 「分散-收集(Scatter-gather)」的DMA。实现了真正的无冗余。

这就是我们常说的零拷贝,在 Java 中FileChannal.transferTo() 底层用的就是sendfile

接下来我们看看以上说的几点在 RocketMQ 和 Kafka中是如何应用的。

RocketMQ 和 Kafka 的应用

RocketMQ

采用Topic混合追加方式,即一个 CommitLog 文件中会包含分给此 Broker 的所有消息,不论消息属于哪个 Topic 的哪个 Queue 。

所以所有的消息过来都是顺序追加写入到 CommitLog 中,并且建立消息对应的 CosumerQueue ,然后消费者是通过 CosumerQueue 得到消息的真实物理地址再去 CommitLog 获取消息的。可以将 CosumerQueue 理解为消息的索引。

在 RocketMQ 中不论是 CommitLog 还是 CosumerQueue 都采用了 mmap。

在发消息的时候默认用的是将数据拷贝到堆内存中,然后再发送。我们来看下代码。

可以看到这个配置 transferMsgByHeap 默认是 true ,那我们再看消费者拉消息时候的代码。

可以看到 RocketMQ 默认把消息拷贝到堆内 Buffer 中,再塞到响应体里面发送。但是可以通过参数配置不经过堆,不过也并没有用到真正的零拷贝,而是通过mapedBuffer 发送到 SocketBuffer 。

所以 RocketMQ 用了顺序写盘、mmap。并没有用到 sendfile ,还有一步页缓存到 SocketBuffer 的拷贝。

然后拉消息的时候严格的说对于 CommitLog 来说读取是随机的,因为 CommitLog 的消息是混合的存储的,但是从整体上看,消息还是从 CommitLog 顺序读的,都是从旧数据到新数据有序的读取。并且一般而言消息存进去马上就会被消费,因此消息这时候应该还在页缓存中,所以不需要读盘。

而且我们在上面提到,页缓存会定时刷盘,这刷盘不可控,并且内存是有限的,会有swap等情况,而且mmap其实只是做了映射,当真正读取页面的时候产生缺页中断,才会将数据真正加载到内存中,这对于消息队列来说可能会产生监控上的毛刺。

因此 RocketMQ 做了一些优化,有:文件预分配和文件预热

文件预分配

CommitLog 的大小默认是1G,当超过大小限制的时候需要准备新的文件,而 RocketMQ 就起了一个后台线程 AllocateMappedFileService,不断的处理 AllocateRequest,AllocateRequest其实就是预分配的请求,会提前准备好下一个文件的分配,防止在消息写入的过程中分配文件,产生抖动。

文件预热

有一个warmMappedFile方法,它会把当前映射的文件,每一页遍历多去,写入一个0字节,然后再调用mlockmadvise(MADV_WILLNEED)

我们再来看下this.mlock,内部其实就是调用了mlockmadvise(MADV_WILLNEED)

mlock:可以将进程使用的部分或者全部的地址空间锁定在物理内存中,防止其被交换到swap空间。

madvise:给操作系统建议,说这文件在不久的将来要访问的,因此,提前读几页可能是个好主意。

RocketMQ 小结

顺序写盘,整体来看是顺序读盘,并且使用了 mmap,不是真正的零拷贝。又因为页缓存的不确定性和 mmap 惰性加载(访问时缺页中断才会真正加载数据),用了文件预先分配和文件预热即每页写入一个0字节,然后再调用mlockmadvise(MADV_WILLNEED)

Kafka

Kafka 的日志存储和 RocketMQ 不一样,它是一个分区一个文件。

Kafka 的消息写入对于单分区来说也是顺序写,如果分区不多的话从整体上看也算顺序写,它的日志文件并没有用到 mmap,而索引文件用了 mmap。但发消息 Kafka 用到了零拷贝。

对于消息的写入来说 mmap 其实没什么用,因为消息是从网络中来。而对于发消息来说 sendfile 对比 mmap+write 我觉得效率更高,因为少了一次页缓存到 SocketBuffer 中的拷贝。

来看下Kafka发消息的源码,最终调用的是 FileChannel.transferTo,底层就是 sendfile。

从 Kafka 源码中我没看到有类似于 RocketMQ的 mlock 等操作,我觉得原因是首先日志也没用到 mmap,然后 swap 其实可以通过 Linux 系统参数 vm.swappiness 来调节,这里建议设置为1,而不是0。

假设内存真的不足,设置为 0 的话,在内存耗尽的情况下,又不能 swap,则会突然中止某些进程。设置个 1,起码还能拖一下,如果有良好的监控手段,还能给个机会发现一下,不至于突然中止。

RocketMQ & Kafka 对比

首先都是顺序写入,不过 RocketMQ 是把消息都存一个文件中,而 Kafka 是一个分区一个文件

每个分区一个文件在迁移或者数据复制层面上来说更加得灵活

但是分区多了的话,写入需要频繁的在多个文件之间来回切换,对于每个文件来说是顺序写入的,但是从全局看其实算随机写入,并且读取的时候也是一样,算随机读。而就一个文件的 RocketMQ 就没这个问题。

从发送消息来说 RocketMQ 用到了 mmap + write 的方式,并且通过预热来减少大文件 mmap 因为缺页中断产生的性能问题。而 Kafka 则用了 sendfile,相对而言我觉得 kafka 发送的效率更高,因为少了一次页缓存到 SocketBuffer 中的拷贝。

并且 swap 问题也可以通过系统参数来设置。

最后

这篇文章中间写 RocketMQ 卡壳了,源码还是不太熟,有点绕。 多亏丁威大佬的点拨。不然我就陷入了死胡同出不来了。

最后再推荐下丁威大佬和周继锋大佬的《RocketMQ技术内幕:RocketMQ架构设计与实现原理》。对 RocketMQ 有兴趣的同学可以看看。

文章如果哪里有纰漏请抓紧联系我,感谢!

我是 yes,从一点点到亿点点,我们下篇见

往期推荐:

消息队列面试热点一锅端

图解+代码|常见限流算法以及限流在单机分布式场景下的思考

表弟面试被虐我教他缓存连招

面试官:说说Kafka处理请求的全流程

Kafka索引设计的亮点

Kafka日志段读写分析

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


文章浏览阅读4.1k次。kafka认证_kafka认证
文章浏览阅读4.8k次,点赞4次,收藏11次。kafka常用参数_kafka配置
文章浏览阅读1.4k次,点赞25次,收藏10次。Kafka 生产者发送消息的流程涉及多个步骤,从消息的创建到成功存储在 Kafka 集群中。_kafka发送消息流程
文章浏览阅读854次,点赞22次,收藏24次。点对点模型:适用于一对一的消息传递,具有高可靠性。发布/订阅模型:适用于广播消息给多个消费者,实现消息的广播。主题模型:适用于根据消息的主题进行灵活的过滤和匹配,处理复杂的消息路由需求。
文章浏览阅读1.5k次,点赞2次,收藏3次。kafka 自动配置在KafkaAutoConfiguration
文章浏览阅读1.3w次,点赞6次,收藏33次。Offset Explorer(以前称为Kafka Tool)是一个用于管理和使Apache Kafka ®集群的GUI应用程序。它提供了一个直观的UI,允许人们快速查看Kafka集群中的对象以及存储在集群主题中的消息。它包含面向开发人员和管理员的功能。二、环境信息系统环境:windows 10版本:2.2Kafka版本:Kafka2.0.0三、安装和使用3.1 下载Offset Explorer 和安装下载到本地的 .exe文件Next安装路径 ,Next。_offset explorer
文章浏览阅读1.3k次,点赞12次,收藏19次。kafka broker 在启动的时候,会根据你配置的listeners 初始化它的网络组件,用来接收外界的请求,这个listeners你可能没配置过,它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议,监听哪个端口,如果我们没有特殊的要求的话,使用它默认的配置就可以了,顶多是修改下端口这块。
文章浏览阅读1.3k次,点赞2次,收藏2次。Kafka 是一个强大的分布式流处理平台,用于实时数据传输和处理。通过本文详细的介绍、使用教程和示例,你可以了解 Kafka 的核心概念、安装、创建 Topic、使用生产者和消费者,从而为构建现代分布式应用打下坚实的基础。无论是构建实时数据流平台、日志收集系统还是事件驱动架构,Kafka 都是一个可靠、高效的解决方案。_博客系统怎么使用kafka
文章浏览阅读3.5k次,点赞42次,收藏56次。对于Java开发者而言,关于 Spring ,我们一般当做黑盒来进行使用,不需要去打开这个黑盒。但随着目前程序员行业的发展,我们有必要打开这个黑盒,去探索其中的奥妙。本期 Spring 源码解析系列文章,将带你领略 Spring 源码的奥秘。本期源码文章吸收了之前 Kafka 源码文章的错误,将不再一行一行的带大家分析源码,我们将一些不重要的分当做黑盒处理,以便我们更快、更有效的阅读源码。废话不多说,发车!
文章浏览阅读1.1k次,点赞14次,收藏16次。一、自动提交offset1、概念Kafka中默认是自动提交offset。消费者在poll到消息后默认情况下,会自动向Broker的_consumer_offsets主题提交当前主题-分区消费的偏移量2、自动提交offset和手动提交offset流程图3、在Java中实现配置4、自动提交offset问题自动提交会丢消息。因为如果消费者还没有消费完poll下来的消息就自动提交了偏移量,那么此时消费者挂了,于是下一个消费者会从已经提交的offset的下一个位置开始消费消息。_kafka中自动提交offsets
文章浏览阅读1.6k次。如果生产者发送消息的速度超过发送到服务器的速度,则会导致生产者空间不足,这个时候KafkaProducer的send()方法调用要么被阻塞,要么抛出异常,这个取决于参数max.block.ms的配置,此参数的默认值为60000,即60秒。在默认情况下,生产者发送的消息是未经压缩的。如果应用程序调用send()方法的速度超过生产者将消息发送给服务器的速度,那么生产者的缓冲空间可能会被耗尽,后续的send()方法调用会等待内存空间被释放,如果在max.block.ms之后还没有可用空间,就抛出异常。_kafka producer 参数
文章浏览阅读2.9k次,点赞3次,收藏10次。kafka解决通信问题_kafka3.6
文章浏览阅读1.5k次,点赞9次,收藏11次。上面都配置完了之后可以先验证下,保证数据最终到ck,如果有问题,需要再每个节点调试,比如先调试nginx->rsyslog ,可以先不配置kafka 输出,配置为console或者文件输出都可以,具体这里就不写了。这里做了一个类型转换,因为nginx,request-time 单位是s,我想最终呈现在grafana 中是ms,所以这里做了转换,当然grafana中也可以做。kafka 相关部署这里不做赘述,只要创建一个topic 就可以。
文章浏览阅读1.4k次,点赞22次,收藏16次。Kafka中的enable-auto-commit和auto-commit-interval配置_auto-commit-interval
文章浏览阅读742次。thingsboard规则链调用外部 kafka_thingsboard kafka
文章浏览阅读1.3k次,点赞18次,收藏22次。Kafka_简介
文章浏览阅读1.1k次,点赞16次,收藏14次。在数据库系统中有个概念叫事务,事务的作用是为了保证数据的一致性,意思是要么数据成功,要么数据失败,不存在数据操作了一半的情况,这就是数据的一致性。在很多系统或者组件中,很多场景都需要保证数据的一致性,有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性,比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。
文章浏览阅读1.4k次。概述介绍架构发展架构原理类型系统介绍类型hive_table类型介绍DataSet类型定义Asset类型定义Referenceable类型定义Process类型定义Entities(实体)Attributes(属性)安装安装环境准备安装Solr-7.7.3安装Atlas2.1.0Atlas配置Atlas集成HbaseAtlas集成SolrAtlas集成KafkaAtlas Server配置Kerberos相关配置Atlas集成HiveAtlas启动Atlas使用Hive元数据初次导入Hive元数据增量同步。_atlas元数据管理
文章浏览阅读659次。Zookeeper是一个开源的分布式服务管理框架。存储业务服务节点元数据及状态信息,并负责通知再 ZooKeeper 上注册的服务几点状态给客户端。
文章浏览阅读1.4k次。Kafka-Kraft 模式架构部署_kafka kraft部署