前面咱们简单讲了K啊开发入门相关的概念、架构、特点以及安装启动。 今天咱们来说一下它的消息存储机制。
前言:
Kafka通过将消息持久化到磁盘上的日志文件来实现高吞吐量的消息传递。
这种存储机制使得Kafka能够处理大量的消息,并保证消息的可靠性。
1、消息存储机制概述:
1.1 分区与副本:
Kafka将每个主题划分为一个或多个分区,每个分区可以有多个副本。分区和副本的概念为Kafka提供了水平扩展和故障容错的能力。
1.2 消息日志:
Kafka的消息存储机制基于消息日志的概念。消息被追加到一个或多个分区的日志文件中,每个分区都有一个单独的日志文件,其中的消息按顺序存储。
1.2.1 消息发送
每当往某个Topic发送数据时,数据会被hash到不同的partition,这些partition位于不同的集群节点上,所以每个消息都会被记录一个offset消息号,随着消息的增加逐渐增加,这个offset也会递增,同时,每个消息会有一个编号,就是offset号。消费者通过这个offset号去查询读取这个消息。
-
发送消息流程
1. 首先获取topic的所有Patition
2. 如果客户端不指定Patition,也没有指定Key的话,使用自增长的数字取余数的方式实现指定的Partition。这样Kafka将平均的向Partition中生产数据。
3. 如果想要控制发送的partition,则有两种方式,一种是指定partition,另一种就是根据Key自己写算法。继承Partitioner接口,实现其partition方法。
![kafka 消息存储机制,后端](https://imgs.yssmx.com/Uploads/2023/10/731333-1.png)
-
消息消费
消费者有消费者族群的概念,当生产者将数据发布到topic时,消费者通过pull的方式,定期从服务器拉取数据,当然在pull数据的时候,,服务器会告诉consumer可消费的消息offset。
![kafka 消息存储机制,后端](https://imgs.yssmx.com/Uploads/2023/10/731333-2.png)
2、源码解析与技术细节:
2.1 日志文件格式:
Kafka使用一种特殊的文件格式来存储消息日志,该格式称为“分段的日志(segmented log)”。
根据你的需求,这里给出一个简单的针对Kafka日志文件源码解析的示例:
Kafka日志文件的源码实现位于Kafka项目的core
模块中,主要包括以下几个关键类和接口:
-
Log
类:代表一个分区的日志文件,它负责对消息的追加、读取和索引等操作。在Log
类中,核心的数据结构是Segment
,它表示一个日志分段。 -
Segment
类:代表日志文件中的一个分段,是Kafka用于存储消息的基本单元。每个分段都有一个起始偏移量和一个结束偏移量,用于定位消息的位置。分段由多个消息组成,按照消息的追加顺序顺序存储。 -
OffsetIndex
类:用于支持高效的偏移量查找。Kafka在每个分段中维护一个偏移量索引,使得可以通过偏移量快速定位到消息的物理位置。 -
OffsetPosition
类:表示一个偏移量在日志文件中的位置信息,包括分段文件名、消息在文件中的位置和消息的大小等信息。
2.2 消息追加与索引:
Kafka使用追加写的方式将消息写入日志文件,并使用索引结构来提供高效地消息检索。我们将通过源码解析,详细讲解消息追加和索引的实现原理及相关技术细节。
下面是一个简化的示例,包括消息的追加、读取和索引等操作:
// Log类的部分关键源码
class Log(dir: File, config: LogConfig) {
// 初始化日志目录
private val logDir = CoreUtils.createDirectory(dir)
// 初始化日志片段
private val segments: mutable.Map[Long, LogSegment] = loadSegments()
// 向日志中追加消息
def append(messages: Seq[Message]): LogAppendInfo = {
// ...
// 将消息追加到当前活跃的日志片段中
val currentSegment = segments(activeSegmentIndex)
currentSegment.append(messages)
// ...
}
// 从日志中读取消息
def read(offset: Long, maxLength: Int): FetchDataInfo = {
// ...
// 根据偏移量找到对应的分段文件
val segment = segments.floorEntry(offset).getValue
segment.read(offset, maxLength)
// ...
}
// 根据配置删除老旧的日志片段
def deleteOldSegments(): Unit = {
// ...
// 删除所有小于日志保留大小的分段
val deletableSegments = segments.filter(segment => segment.getSize <= config.retentionSize)
deletableSegments.foreach {
case (_, segment) => segment.delete()
}
// ...
}
// 加载已存在的日志片段
private def loadSegments(): mutable.Map[Long, LogSegment] = {
// ...
// 遍历日志目录下的所有分段文件,加载到内存中
val segments = mutable.Map[Long, LogSegment]()
val segmentFiles = logDir.listFiles.filter(_.isFile).sortBy(_.getName)
segmentFiles.foreach { file =>
// 解析文件名中的偏移量
val offset = parseOffset(file)
// 创建并加载分段
val segment = new LogSegment(file, offset)
segment.load()
// 添加到分段列表中
segments.put(offset, segment)
}
// ...
segments
}
}
2.3 日志压缩:
在Kafka中,可以通过启用日志压缩来减小存储空间的占用和网络传输的开销。Kafka支持多种压缩算法,包括Gzip、Snappy和LZ4等。下面是一个简单的步骤,说明如何在Kafka中启用日志压缩:
-
在Kafka服务器配置文件中,找到以下配置项:
compression.type = producer
将该配置项的值设置为所需的压缩算法,例如:
compression.type = gzip
-
如果你的Kafka集群有多个副本(replica),你还需要在Kafka服务器配置文件中为每个副本设置以下配置项:
min.insync.replicas = 2
该配置项指定了进行压缩的最小副本数,确保至少有指定数量的副本处于同步状态。这是为了防止数据丢失,在进行日志压缩时仍然能够保持高可靠性。
-
重启Kafka服务器,以使配置生效。
在启用日志压缩后,Kafka将会自动对生产者发送的消息进行压缩,并在消费者读取消息时自动解压缩。这样可以显著减小消息的存储空间和网络传输开销,提高系统的性能和效率。
需要注意的是,在启用日志压缩后,读写数据的性能会受到一些影响,因为压缩和解压缩需要一定的计算资源。因此,在选择压缩算法和配置压缩参数时,需要权衡存储空间的节省和性能的需求。
此外,还有一种在Kafka中压缩日志的方法是使用外部工具(如Hadoop的hadoop-archive-logs
命令),先将日志文件打包成压缩文件(如tar.gz),然后再进行存储。这种方式需要额外的步骤和工具,并且不支持实时的压缩和解压缩。因此,如果需要实时的压缩和解压缩功能,建议使用Kafka内置的日志压缩功能。
3、存储性能优化:
优化Kafka存储性能可以提高消息的写入和读取速度,以及减少存储占用。下面是一些常见的Kafka存储性能优化策略建议:
-
批量发送:通过将多条消息合并为一个批次进行发送,可以减少网络传输开销和降低磁盘IO。在生产者端,可以设置
batch.size
参数来调整批次大小。较大的批次大小可以提高吞吐量,但可能会增加延迟。在消费者端,可以使用fetch.min.bytes
参数来配置批量拉取的最小字节数,默认为1字节。 -
合理的副本因子:Kafka的消息是以副本的形式存储在不同的节点上。通过合理配置副本因子,可以在保证消息的可靠性的同时,提高写入性能。较小的副本因子可以减少副本间的同步开销。如有必要,可以将
min.insync.replicas
参数设置为小于副本因子的值。但同时要注意,较小的副本因子可能会增加消息的丢失风险。 -
启用压缩:Kafka支持对消息进行压缩,在减小存储占用和网络传输开销方面具有很大优势。可以配置生产者的
compression.type
参数来启用压缩功能,并选择合适的压缩算法(如Gzip、Snappy或LZ4)。压缩会增加一些额外的CPU开销,但通常能在存储和传输方面带来明显的性能收益。 -
SSD存储:Kafka使用大量的磁盘IO,因此使用固态硬盘(SSD)可以显著提高性能。SSD具有更低的读写延迟和更高的吞吐量,适合处理大量的随机读写操作。
-
分区和副本的平衡:合理设置分区和副本的数量,可以提高负载均衡和并行处理能力。如果某个分区或副本的读写速度较慢,可以考虑增加其数量。
-
优化日志清理:Kafka会定期清理日志段文件来释放磁盘空间。通过调整
log.retention.hours
和log.retention.bytes
等参数,可以控制日志的保留时间和大小。合理设置这些参数可以避免过早的数据清理和降低磁盘压力。 -
确保足够的磁盘带宽:Kafka的存储性能受限于磁盘带宽。确保磁盘子系统具有足够的带宽和IO吞吐量,可以避免磁盘成为性能瓶颈。
以上是一些常见的Kafka存储性能优化策略,根据实际情况和需求,可以选择适合的优化方法,并进行配置和调整。同时,定期监控系统性能并进行性能测试,可以帮助发现潜在的性能问题并进行进一步优化。
顶尖架构师栈
关注回复关键字
【C01】超10G后端学习面试资源
【IDEA】最新IDEA激活工具和码及教程
【JetBrains软件名】 最新软件激活工具和码及教程
工具&码&教程文章来源:https://www.toymoban.com/news/detail-731333.html
本文由 mdnice 多平台发布文章来源地址https://www.toymoban.com/news/detail-731333.html
到了这里,关于Kafka的消息存储机制的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!