• Kafka的消息存储机制


    前面咱们简单讲了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方法。

    alt
    • 消息消费

      消费者有消费者族群的概念,当生产者将数据发布到topic时,消费者通过pull的方式,定期从服务器拉取数据,当然在pull数据的时候,,服务器会告诉consumer可消费的消息offset。

    alt

    2、源码解析与技术细节:

    2.1 日志文件格式:

    Kafka使用一种特殊的文件格式来存储消息日志,该格式称为“分段的日志(segmented log)”。

    alt 根据你的需求,这里给出一个简单的针对Kafka日志文件源码解析的示例:

    Kafka日志文件的源码实现位于Kafka项目的core模块中,主要包括以下几个关键类和接口:

    1. Log类:代表一个分区的日志文件,它负责对消息的追加、读取和索引等操作。在Log类中,核心的数据结构是Segment,它表示一个日志分段。

    2. Segment类:代表日志文件中的一个分段,是Kafka用于存储消息的基本单元。每个分段都有一个起始偏移量和一个结束偏移量,用于定位消息的位置。分段由多个消息组成,按照消息的追加顺序顺序存储。

    3. OffsetIndex类:用于支持高效的偏移量查找。Kafka在每个分段中维护一个偏移量索引,使得可以通过偏移量快速定位到消息的物理位置。

    4. OffsetPosition类:表示一个偏移量在日志文件中的位置信息,包括分段文件名、消息在文件中的位置和消息的大小等信息。

    2.2 消息追加与索引:

    Kafka使用追加写的方式将消息写入日志文件,并使用索引结构来提供高效地消息检索。我们将通过源码解析,详细讲解消息追加和索引的实现原理及相关技术细节。

    下面是一个简化的示例,包括消息的追加、读取和索引等操作:

    // Log类的部分关键源码
    class Log(dir: File, config: LogConfig{
      // 初始化日志目录
      private val logDir = CoreUtils.createDirectory(dir)

      // 初始化日志片段
      private val segments: mutable.Map[LongLogSegment] = 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[LongLogSegment] = {
        // ...
        // 遍历日志目录下的所有分段文件,加载到内存中
        val segments = mutable.Map[LongLogSegment]()
        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
      }
    }
    • 1

    2.3 日志压缩:

    在Kafka中,可以通过启用日志压缩来减小存储空间的占用和网络传输的开销。Kafka支持多种压缩算法,包括Gzip、Snappy和LZ4等。下面是一个简单的步骤,说明如何在Kafka中启用日志压缩:

    1. 在Kafka服务器配置文件中,找到以下配置项:

      compression.type = producer
      • 1

      将该配置项的值设置为所需的压缩算法,例如:

      compression.type = gzip
      • 1
    2. 如果你的Kafka集群有多个副本(replica),你还需要在Kafka服务器配置文件中为每个副本设置以下配置项:

      min.insync.replicas = 2
      • 1

      该配置项指定了进行压缩的最小副本数,确保至少有指定数量的副本处于同步状态。这是为了防止数据丢失,在进行日志压缩时仍然能够保持高可靠性。

    3. 重启Kafka服务器,以使配置生效。

    在启用日志压缩后,Kafka将会自动对生产者发送的消息进行压缩,并在消费者读取消息时自动解压缩。这样可以显著减小消息的存储空间和网络传输开销,提高系统的性能和效率。

    需要注意的是,在启用日志压缩后,读写数据的性能会受到一些影响,因为压缩和解压缩需要一定的计算资源。因此,在选择压缩算法和配置压缩参数时,需要权衡存储空间的节省和性能的需求。

    此外,还有一种在Kafka中压缩日志的方法是使用外部工具(如Hadoop的hadoop-archive-logs命令),先将日志文件打包成压缩文件(如tar.gz),然后再进行存储。这种方式需要额外的步骤和工具,并且不支持实时的压缩和解压缩。因此,如果需要实时的压缩和解压缩功能,建议使用Kafka内置的日志压缩功能。

    3、存储性能优化:

    优化Kafka存储性能可以提高消息的写入和读取速度,以及减少存储占用。下面是一些常见的Kafka存储性能优化策略建议:

    1. 批量发送:通过将多条消息合并为一个批次进行发送,可以减少网络传输开销和降低磁盘IO。在生产者端,可以设置batch.size参数来调整批次大小。较大的批次大小可以提高吞吐量,但可能会增加延迟。在消费者端,可以使用fetch.min.bytes参数来配置批量拉取的最小字节数,默认为1字节。

    2. 合理的副本因子:Kafka的消息是以副本的形式存储在不同的节点上。通过合理配置副本因子,可以在保证消息的可靠性的同时,提高写入性能。较小的副本因子可以减少副本间的同步开销。如有必要,可以将min.insync.replicas参数设置为小于副本因子的值。但同时要注意,较小的副本因子可能会增加消息的丢失风险。

    3. 启用压缩:Kafka支持对消息进行压缩,在减小存储占用和网络传输开销方面具有很大优势。可以配置生产者的compression.type参数来启用压缩功能,并选择合适的压缩算法(如Gzip、Snappy或LZ4)。压缩会增加一些额外的CPU开销,但通常能在存储和传输方面带来明显的性能收益。

    4. SSD存储:Kafka使用大量的磁盘IO,因此使用固态硬盘(SSD)可以显著提高性能。SSD具有更低的读写延迟和更高的吞吐量,适合处理大量的随机读写操作。

    5. 分区和副本的平衡:合理设置分区和副本的数量,可以提高负载均衡和并行处理能力。如果某个分区或副本的读写速度较慢,可以考虑增加其数量。

    6. 优化日志清理:Kafka会定期清理日志段文件来释放磁盘空间。通过调整log.retention.hourslog.retention.bytes等参数,可以控制日志的保留时间和大小。合理设置这些参数可以避免过早的数据清理和降低磁盘压力。

    7. 确保足够的磁盘带宽:Kafka的存储性能受限于磁盘带宽。确保磁盘子系统具有足够的带宽和IO吞吐量,可以避免磁盘成为性能瓶颈。

    以上是一些常见的Kafka存储性能优化策略,根据实际情况和需求,可以选择适合的优化方法,并进行配置和调整。同时,定期监控系统性能并进行性能测试,可以帮助发现潜在的性能问题并进行进一步优化。

    顶尖架构师栈

    关注回复关键字

    【C01】超10G后端学习面试资源

    【IDEA】最新IDEA激活工具和码及教程

    【JetBrains软件名】 最新软件激活工具和码及教程

    工具&码&教程

    本文由 mdnice 多平台发布

  • 相关阅读:
    QT配合CSS隐藏按钮
    Enhancing Quality for HEVC Compressed Videos
    Netty网络框架学习笔记-19(实现一个简单RPC-1)
    RVMedia VCL 8.0 for Delphi 10.1-10.4
    React进阶之路(一)-- JSX基础、组件基础
    小米 Civi 2 (ziyi) 机型解锁bl 获取root权限教程 +其他机型参数对比+救砖
    学习【Cesium】第三篇,从Cesium.Viewer查看器开始学习(学不会揍我)
    C++核心编程 day09 类型转换、异常、输入输出流
    STM8与汇编语言(9)--EEPROM应用
    leetcode 942. 增减字符串匹配
  • 原文地址:https://blog.csdn.net/Dc253061007/article/details/133249619