Kafka的日志存储

Kafka 消息是以主题为单位进行归类，各个主题之间是彼此独立的，互不影响。

每个主题⼜可以分为⼀个或多个分区。

每个分区各⾃存在⼀个记录消息数据的日志文件。

图中，创建了⼀个 tp_demo_01 主题，其存在6个 Parition，对应的每个Parition下存在⼀个 [Topic-Parition] 命名的消息⽇志⽂件。在理想情况下，数据流量分摊到各个 Parition 中，实现了负载均衡的效果。在分区日志文件中，你会发现很多类型的⽂件，比如： index、timestamp、log、snapshot 等。

其中，文件名⼀致的⽂件集合就称为 LogSement。

当满⾜如下⼏个条件中的其中之⼀，就会触发文件的切分：

偏移量索引文件用于记录消息偏移量与物理地址之间的映射关系。时间戳索引文件则根据时间戳查找对应的偏移量。

文件：

查看⼀个topic分区目录下的内容，发现有log、index和timeindex三个⽂件：

创建主题：

创建消息⽂件：

将⽂本消息⽣产到主题中：

查看存储⽂件：

如果想查看这些文件，可以使⽤kafka提供的shell来完成，几个关键信息如下：

（1）offset是逐渐增加的整数，每个offset对应⼀个消息的偏移量。

（2）position：消息批字节数，用于计算物理地址。

（3）CreateTime：时间戳。

（4）magic：2代表这个消息类型是V2，如果是0则代表是V0类型，1代表V1类型。

（5）compresscodec：None说明没有指定压缩类型，kafka目前提供了4种可选择，0-None、1-GZIP、2-snappy、3-lz4。

（6）crc：对所有字段进行校验后的crc值。

在偏移量索引文件中，索引数据都是顺序记录 offset ，但时间戳索引文件中每个追加的索引时间戳必须大于之前追加的索引项，否则不予追加。在 Kafka 01100 以后，消息元数据中存在若⼲的时间戳信息。如果 broker 端参数logmessagetimestamptype 设置为 LogAppendTIme ，那么时间戳必定能保持单调增⻓。反之如果是CreateTime 则⽆法保证顺序。

注意：timestamp文件中的 offset 与 index ⽂件中的 relativeOffset 不是⼀⼀对应的。因为数据的写⼊是各自追加。

思考：如何查看偏移量为23的消息？

Kafka 中存在⼀个 ConcurrentSkipListMap 来保存在每个日志分段，通过跳跃表方式，定位到在00000000000000000000index ，通过二分法在偏移量索引文件中找到不⼤于 23 的最⼤索引项，即 offset 20 那栏，然后从⽇志分段⽂件中的物理位置为320 开始顺序查找偏移量为 23 的消息。

在偏移量索引文件中，索引数据都是顺序记录 offset ，但时间戳索引⽂件中每个追加的索引时间戳必须大于之前追加的索引项，否则不予追加。在 Kafka 01100 以后，消息信息中存在若⼲的时间戳信息。

如果 broker 端参数logmessagetimestamptype 设置为 LogAppendTIme ，那么时间戳必定能保持单调增长。反之如果是CreateTime 则无法保证顺序。

通过时间戳方式进行查找消息，需要通过查找时间戳索引和偏移量索引两个文件。

时间戳索引索引格式：前⼋个字节表示时间戳，后四个字节表示偏移量。

思考：查找时间戳为 1557554753430 开始的消息？

Kafka 提供两种⽇志清理策略：

⽇志删除：按照⼀定的删除策略，将不满⾜条件的数据进⾏数据删除

⽇志压缩：针对每个消息的 Key 进⾏整合，对于有相同 Key 的不同 Value 值，只保留最后⼀个版本。

Kafka 提供 logcleanuppolicy 参数进⾏相应配置，默认值： delete ，还可以选择 compact 。

主题级别的配置项是 cleanuppolicy 。

基于时间

⽇志删除任务会根据 logretentionhours/logretentionminutes/logretentionms 设定⽇志保留的

时间节点。如果超过该设定值，就需要进⾏删除。默认是 7 天， logretentionms 优先级最⾼。

Kafka 依据⽇志分段中最⼤的时间戳进⾏定位。

⾸先要查询该⽇志分段所对应的时间戳索引⽂件，查找时间戳索引⽂件中最后⼀条索引项，若最后⼀条索引项的时间戳字段值⼤于 0，则取该值，否则取最近修改时间。

为什么不直接选最近修改时间呢？

因为日志文件可以有意⽆意的被修改，并不能真实的反应日志分段的最⼤时间信息。

删除过程

⽇志压缩是Kafka的⼀种机制，可以提供较为细粒度的记录保留，⽽不是基于粗粒度的基于时间的保留。

对于具有相同的Key，⽽数据不同，只保留最后⼀条数据，前⾯的数据在合适的情况下删除。

⽇志压缩特性，就实时计算来说，可以在异常容灾⽅⾯有很好的应⽤途径。⽐如，我们在Spark、Flink中做实时

计算时，需要⻓期在内存⾥⾯维护⼀些数据，这些数据可能是通过聚合了⼀天或者⼀周的⽇志得到的，这些数据⼀旦

由于异常因素（内存、⽹络、磁盘等）崩溃了，从头开始计算需要很⻓的时间。⼀个⽐较有效可⾏的⽅式就是定时将

内存⾥的数据备份到外部存储介质中，当崩溃出现时，再从外部存储介质中恢复并继续计算。

使⽤⽇志压缩来替代这些外部存储有哪些优势及好处呢？这⾥为⼤家列举并总结了⼏点：

Kafka即是数据源⼜是存储⼯具，可以简化技术栈，降低维护成本

使⽤外部存储介质的话，需要将存储的Key记录下来，恢复的时候再使⽤这些Key将数据取回，实现起来有⼀定的⼯程难度和复杂度。使⽤Kafka的⽇志压缩特性，只需要把数据写进Kafka，等异常出现恢复任务时再读

回到内存就可以了

Kafka对于磁盘的读写做了⼤量的优化⼯作，⽐如磁盘顺序读写。相对于外部存储介质没有索引查询等⼯作

量的负担，可以实现⾼性能。同时，Kafka的⽇志压缩机制可以充分利⽤廉价的磁盘，不⽤依赖昂贵的内存

来处理，在性能相似的情况下，实现⾮常⾼的性价⽐（这个观点仅仅针对于异常处理和容灾的场景来说）

主题的 cleanuppolicy 需要设置为compact。

Kafka的后台线程会定时将Topic遍历两次：

Kafka的offset是如何设置的？

答：是生产者设置的，生产者在发送消息的时候，为每条消息生成一个唯一的offset。

Kafka消息的格式？

答：

Kafka最新版本的消息集叫做RecordBatch，而不是先前的MessageSet。RecordBatch内部存储了一条或多条消息。

RecordBatch的结构包含以下部分：

first offset，起始位移，占位8B

length，消息总长度，占位4B

partition leader epoch，分区leader纪元，可以看做分区leader的版本号或者更新次数，占位4B。

magic，消息格式的版本号，对于V2版本而言，magic的值为2。

attributes，消息属性，占位2B，低三位表示压缩格式，第4位表示时间戳类型，第五位表示当前RecordBatch是否处于事务中第6位表示是否控制消息。

last offset delta，占位4B，RecordBatch中最后一个Record的offset与first offset的差值，主要被broker用来确保RecordBatch中Record组装的正确性。

first timestamp，占位8B，RecordBatch中第一条Record的时间戳。

max timestamp，占位8B，RecordBatch中最大的时间戳，一般情况下是最后一个Record的时间戳。和last offset delta功能一样，主要被broker用来确保RecordBatch中Record组装的正确性。

producer id，即PID，占位8B，用来支持幂等和事务。

producer epoch，占位2B，用来支持幂等和事务。

first sequence，占位4B，用来支持幂等和事务。

records count，占位4B，RecordBatch中Record的总数。

records，存放消息的容器。

Records的数据结构又是什么样的呢？Record包含以下属性：

length，消息总长度。

attributes，目前已弃用，但是还是会占用1B的空间，以备未来的格式扩展。

timestamp delta，时间戳增量，通常一个timestamp占用8B，这里时间戳增量保存的是当前时间戳与RecordBatch中first timestamp的差值，这样可以节省占用空间。

offset delta，位移增量，这个是当前消息的位移与RecordBatch中first offset的差值，这样可以节省占用空间。

key length，消息的key的长度。

key，消息key。

value length，消息的value的长度。

value，消息的值。

headers count，headers的总数。

headers，这个字段是用来支持应用级别的扩展，而不需要将一些应用级别的属性嵌入到消息体中。

consumer group是kafka提供的可扩展且具有容错性的消费者机制。组内可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。

consumer group下可以有一个或多个consumer instance，consumer instance可以是一个进程，也可以是一个线程

groupid是一个字符串，唯一标识一个consumer group

consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)

Coordinator一般指的是运行在broker上的group Coordinator，用于管理Consumer Group中各个成员，每个KafkaServer都有一个GroupCoordinator实例，管理多个消费者组，主要用于offset位移管理和Consumer Rebalance。

对于每个Consumer Group，Coordinator会存储以下信息：

consumer group如何确定自己的coordinator是谁呢？ 简单来说分为两步：

消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在Kafka中这个位置信息有个专门的术语：位移(offset)。

(1)、很多消息引擎都把这部分信息保存在服务器端(broker端)。这样做的好处当然是实现简单，但会有三个主要的问题：

1 broker从此变成有状态的，会影响伸缩性；

2 需要引入应答机制(acknowledgement)来确认消费成功。

3 由于要保存很多consumer的offset信息，必然引入复杂的数据结构，造成资源浪费。

而Kafka选择了不同的方式：每个consumer group管理自己的位移信息，那么只需要简单的一个整数表示位置就够了；同时可以引入checkpoint机制定期持久化，简化了应答机制的实现。

(2)、Kafka默认是定期帮你自动提交位移的(enableautocommit = true)，你当然可以选择手动提交位移实现自己控制。

(3)、另外kafka会定期把group消费情况保存起来，做成一个offset map，如下图所示：

上图中表明了test-group这个组当前的消费情况。

老版本的位移是提交到zookeeper中的，目录结构是：/consumers/<groupid>/offsets/<topic>/<partitionId>，但是zookeeper其实并不适合进行大批量的读写 *** 作，尤其是写 *** 作。

__consumers_offsets topic配置了compact策略，使得它总是能够保存最新的位移信息，既控制了该topic总体的日志容量，也能实现保存最新offset的目的。compact的具体原理请参见： Log Compaction

至于每个group保存到__consumers_offsets的哪个分区，如何查看的问题请参见这篇文章： Kafka 如何读取offset topic内容 (__consumer_offsets)

offset提交消息会根据消费组的key(消费组名称)进行分区 对于一个给定的消费组,它的所有消息都会发送到唯一的broker(即Coordinator)

Coordinator上负责管理offset的组件是 Offset manager 。负责存储,抓取,和维护消费者的offsets 每个broker都有一个offset manager实例 有两种具体的实现:

ZookeeperOffsetManager: 调用zookeeper来存储和接收offset（老版本的位移管理）。

DefaultOffsetManager: 提供消费者offsets内置的offset管理。

通过在config/serverproperties中的offsetstorage参数选择。

DefaultOffsetManager

除了将offset作为logs保存到磁盘上,DefaultOffsetManager维护了一张能快速服务于offset抓取请求的 consumer offsets表 。这个表作为缓存，包含的含仅仅是”offsets topic”的partitions中属于leader partition对应的条目(存储的是offset)。

对于DefaultOffsetManager还有两个其他属性: “offsetstopicreplicationfactor和”offsetstopicnumpartitions”,默认值都是1。这两个属性会用来自动地创建”offsets topic”。

offset manager接口的概要：

什么是rebalance？

rebalance本质上是一种协议，规定了一个consumer group下的所有consumer如何达成一致来分配订阅topic的每个分区。比如某个group下有20个consumer，它订阅了一个具有100个分区的topic。正常情况下，Kafka平均会为每个consumer分配5个分区。这个分配的过程就叫rebalance。Kafka新版本consumer默认提供了两种分配策略：range和round-robin。

rebalance的触发条件有三种：

组成员发生变更(新consumer加入组、已有consumer主动离开组或已有consumer崩溃了——这两者的区别后面会谈到)

订阅主题数发生变更——这当然是可能的，如果你使用了正则表达式的方式进行订阅，那么新建匹配正则表达式的topic就会触发rebalance

订阅主题的分区数发生变更

refer

>

Kafka __consumer_offsets是一个特殊的存储元数据的Topic

数据格式可以想象成一个 KV 格式的消息，key 就是一个三元组：groupid+topic+分区号，而 value 就是 offset 的值。

查看方式:使用kafka自带的读取类

/bin/kafka-console-consumersh --topic __consumer_offsets --partition 01 --bootstrap-server xxx:9092 --formatter "kafkacoordinatorgroupGroupMetadataManager$OffsetsMessageFormatter" --from-beginning --max-messages 30

一般情况下， 使用 OffsetsMessageFormatter 打印的格式可以概括为：

"[%s,%s,%d]::[OffsetMetadata[%d,%s],CommitTime %d,ExpirationTime %d]"format(group, topic, partition, offset, metadata, commitTimestamp, expireTimestamp)

数据内容:

[flink-payment-alert_query_time_1576066085229,payment-result-count,4]::NULL

[flink-payment-alert_query_time_1576066085229,payment-result-count,3]::NULL

[flink-payment-alert_query_time_1576066085229,payment-result-count,9]::NULL

另外一种是

[work_default_ywintspringcxywblackgoldkafkaorderdomaincoresub,work_default_ywintspringcxywblackgoldkafkaorderdomaintopic,0]::OffsetAndMetadata(offset=19, leaderEpoch=Optionalempty, metadata=, commitTimestamp=1636939024066, expireTimestamp=None)

[work_default_ywintspringcxywblackgoldkafkaorderdomaincoresub,work_default_ywintspringcxywblackgoldkafkaorderdomaintopic,0]::OffsetAndMetadata(offset=19, leaderEpoch=Optionalempty, metadata=, commitTimestamp=1636939028621, expireTimestamp=None)

[work_default_ywintspringcxywblackgoldkafkaorderdomaincoresub,work_default_ywintspringcxywblackgoldkafkaorderdomaintopic,0]::OffsetAndMetadata(offset=19, leaderEpoch=Optionalempty, metadata=, commitTimestamp=1636939033680, expireTimestamp=None)

还有一种是

[ProcessEngineBusinessProcess,CasBusinessTopic,1]::[OffsetMetadata[99649027,NO_METADATA],CommitTime 1636930671854,ExpirationTime 1637017071854]

[ProcessEngineBusinessProcess,CasBusinessTopic,0]::[OffsetMetadata[99650360,NO_METADATA],CommitTime 1636930671854,ExpirationTime 1637017071854]

[ProcessEngineBusinessProcess,CasBusinessTopic,3]::[OffsetMetadata[99640798,NO_METADATA],CommitTime 1636930672471,ExpirationTime 1637017072471]

分别解释一下:

在 Kafka 中有一个名为“delete-expired-group-metadata”的定时任务来负责清理过期的消费位移，这个定时任务的执行周期由参数 offsetsretentioncheckintervalms 控制，默认值为600000，即10分钟。这和普通的topic的不太一样

还有 metadata，一般情况下它的值要么为 null 要么为空字符串，OffsetsMessageFormatter 会把它展示为 NO_METADATA，否则就按实际值进行展示。

看一下源码里这些类的结构

case class OffsetAndMetadata(offsetMetadata: OffsetMetadata,

commitTimestamp: Long = orgapachekafkacommonrequestsOffsetCommitRequestDEFAULT_TIMESTAMP,

expireTimestamp: Long = orgapachekafkacommonrequestsOffsetCommitRequestDEFAULT_TIMESTAMP) {

case class OffsetMetadata(offset: Long, metadata: String = OffsetMetadataNoMetadata) {

override def toString = "OffsetMetadata[%d,%s]"

format(offset,

if (metadata != null && metadatalength > 0) metadata else "NO_METADATA")

}

@Deprecated

public static final long DEFAULT_TIMESTAMP = -1L; // for V0, V1

另外0110之后对应的数据格式版本是V2,这个版本的消息相比于v0和v1的版本而言改动很大，同时还参考了Protocol Buffer而引入了变长整型（Varints）和ZigZag编码。

另外:

offset为什么会有墓碑消息

因为offset本身也会过期清理受offsetsretentionminutes 这个配置的影响

看下官网介绍

After a consumer group loses all its consumers (ie becomes empty) its offsets will be kept for this retention period before getting discarded For standalone consumers (using manual assignment), offsets will be expired after the time of last commit plus this retention period

当group里的consumer全部下线后过offsetsretentionminutes 时间后offset就会被删除

val OffsetsRetentionMinutes: Int = 7 24 60 // 默认7天

默认20之前是1天,20及以后是7天 这个官方真是要么就改为2天,结果直接改为7天,改动不可谓不大,而且active的group不会过期

附: >

作为一款典型的消息中间件产品，kafka系统仍然由producer、broker、consumer三部分组成。kafka涉及的几个常用概念和组件简单介绍如下：

当consumer group的状态发生变化（如有consumer故障、增减consumer成员等）或consumer group消费的topic状态发生变化（如增加了partition，消费的topic发生变化），kafka集群会自动调整和重新分配consumer消费的partition，这个过程就叫做rebalance（再平衡）。

__consumer_offsets是kafka集群自己维护的一个特殊的topic，它里面存储的是每个consumer group已经消费了每个topic partition的offset。__consumer_offsets中offset消息的key由group id，topic name，partition id组成，格式为 {topic name}-${partition id}，value值就是consumer提交的已消费的topic partition offset值。__consumer_offsets的分区数和副本数分别由offsetstopicnumpartitions（默认值为50）和offsetstopicreplicationfactor（默认值为1）参数配置。我们通过公式 hash(group id) % offsetstopicnumpartitions 就可以计算出指定consumer group的已提交offset存储的partition。由于consumer group提交的offset消息只有最后一条消息有意义，所以__consumer_offsets是一个compact topic，kafka集群会周期性的对__consumer_offsets执行compact *** 作，只保留最新的一次提交offset。

group coordinator运行在kafka某个broker上，负责consumer group内所有的consumer成员管理、所有的消费的topic的partition的消费关系分配、offset管理、触发rebalance等功能。group coordinator管理partition分配时，会指定consumer group内某个consumer作为group leader执行具体的partition分配任务。存储某个consumer group已提交offset的__consumer_offsets partition leader副本所在的broker就是该consumer group的协调器运行的broker。

跟大多数分布式系统一样，集群有一个master角色管理整个集群，协调集群中各个成员的行为。kafka集群中的controller就相当于其它分布式系统的master，用来负责集群topic的分区分配，分区leader选举以及维护集群的所有partition的ISR等集群协调功能。集群中哪个borker是controller也是通过一致性协议选举产生的，28版本之前通过zookeeper进行选主，28版本后通过kafka raft协议进行选举。如果controller崩溃，集群会重新选举一个broker作为新的controller，并增加controller epoch值（相当于zookeeper ZAB协议的epoch，raft协议的term值）

当kafka集群新建了topic或为一个topic新增了partition，controller需要为这些新增加的partition分配到具体的broker上，并把分配结果记录下来，供producer和consumer查询获取。

因为只有partition的leader副本才会处理producer和consumer的读写请求，而partition的其他follower副本需要从相应的leader副本同步消息，为了尽量保证集群中所有broker的负载是均衡的，controller在进行集群全局partition副本分配时需要使partition的分布情况是如下这样的：

在默认情况下，kafka采用轮询（round-robin）的方式分配partition副本。由于partition leader副本承担的流量比follower副本大，kafka会先分配所有topic的partition leader副本，使所有partition leader副本全局尽量平衡，然后再分配各个partition的follower副本。partition第一个follower副本的位置是相应leader副本的下一个可用broker，后面的副本位置依此类推。

举例来说，假设我们有两个topic，每个topic有两个partition，每个partition有两个副本，这些副本分别标记为1-1-1，1-1-2，1-2-1，1-2-2，2-1-1，2-1-2，2-2-1，2-2-2（编码格式为topic-partition-replia，编号均从1开始，第一个replica是leader replica，其他的是follower replica）。共有四个broker，编号是1-4。我们先对broker按broker id进行排序，然后分配leader副本，最后分配foller副本。

1）没有配置brokerrack的情况

现将副本1-1-1分配到broker 1，然后1-2-1分配到broker 2，依此类推，2-2-1会分配到broker 4。partition 1-1的leader副本分配在broker 1上，那么下一个可用节点是broker 2，所以将副本1-1-2分配到broker 2上。同理，partition 1-2的leader副本分配在broker 2上，那么下一个可用节点是broker 3，所以将副本1-1-2分配到broker 3上。依此类推分配其他的副本分片。最后分配的结果如下图所示：

2）配置了brokerrack的情况

假设配置了两个rack，broker 1和broker 2属于Rack 1，broker 3和broker 4属于Rack 2。我们对rack和rack内的broker分别排序。然后先将副本1-1-1分配到Rack 1的broker 1，然后将副本1-2-1分配到下一个Rack的第一个broker，即Rack 2的broker 3。其他的parttition leader副本依此类推。然后分配follower副本，partition 1-1的leader副本1-1-1分配在Rack 1的broker上，下一个可用的broker是Rack 2的broker 3，所以分配到broker 3上，其他依此类推。最后分配的结果如下图所示：

kafka除了按照集群情况自动分配副本，也提供了reassign工具人工分配和迁移副本到指定broker，这样用户可以根据集群实际的状态和各partition的流量情况分配副本

kafka集群controller的一项功能是在partition的副本中选择一个副本作为leader副本。在topic的partition创建时，controller首先分配的副本就是leader副本，这个副本又叫做preference leader副本。

当leader副本所在broker失效时（宕机或网络分区等），controller需要为在该broker上的有leader副本的所有partition重新选择一个leader，选择方法就是在该partition的ISR中选择第一个副本作为新的leader副本。但是，如果ISR成员只有一个，就是失效的leader自身，其余的副本都落后于leader怎么办？kafka提供了一个uncleanleaderelection配置参数，它的默认值为true。当uncleanleaderelection值为true时，controller还是会在非ISR副本中选择一个作为leader，但是这时候使用者需要承担数据丢失和数据不一致的风险。当uncleanleaderelection值为false时，则不会选择新的leader，该partition处于不可用状态，只能恢复失效的leader使partition重新变为可用。

当preference leader失效后，controller重新选择一个新的leader，但是preference leader又恢复了，而且同步上了新的leader，是ISR的成员，这时候preference leader仍然会成为实际的leader，原先的新leader变为follower。因为在partition leader初始分配时，使按照集群副本均衡规则进行分配的，这样做可以让集群尽量保持平衡。

为了保证topic的高可用，topic的partition往往有多个副本，所有的follower副本像普通的consumer一样不断地从相应的leader副本pull消息。每个partition的leader副本会维护一个ISR列表存储到集群信息库里，follower副本成为ISR成员或者说与leader是同步的，需要满足以下条件：

1）follower副本处于活跃状态，与zookeeper（28之前版本）或kafka raft master之间的心跳正常

2）follower副本最近replicalagtimemaxms（默认是10秒）时间内从leader同步过最新消息。需要注意的是，一定要拉取到最新消息，如果最近replicalagtimemaxms时间内拉取过消息，但不是最新的，比如落后follower在追赶leader过程中，也不会成为ISR。

follower在同步leader过程中，follower和leader都会维护几个参数，来表示他们之间的同步情况。leader和follower都会为自己的消息队列维护LEO（Last End Offset）和HW（High Watermark）。leader还会为每一个follower维护一个LEO。LEO表示leader或follower队列写入的最后一条消息的offset。HW表示的offset对应的消息写入了所有的ISR。当leader发现所有follower的LEO的最小值大于HW时，则会增加HW值到这个最小值LEO。follower拉取leader的消息时，同时能获取到leader维护的HW值，如果follower发现自己维护的HW值小于leader发送过来的HW值，也会增加本地的HW值到leader的HW值。这样我们可以得到一个不等式： follower HW <= leader HW <= follower LEO <= leader LEO 。HW对应的log又叫做committed log，consumer消费partititon的消息时，只能消费到offset值小于或等于HW值的消息的，由于这个原因，kafka系统又称为分布式committed log消息系统。

kafka的消息内容存储在logdirs参数配置的目录下。kafka每个partition的数据存放在本地磁盘logdirs目录下的一个单独的目录下，目录命名规范为 ${topicName}-${partitionId} ，每个partition由多个LogSegment组成，每个LogSegment由一个数据文件(命名规范为： {baseOffset}index）和一个时间戳索引文件（命名规范为：${baseOffset}timeindex）组成，文件名的baseOffset就是相应LogSegment中第一条消息的offset。index文件存储的是消息的offset到该消息在相应log文件中的偏移，便于快速在log文件中快速找到指定offset的消息。index是一个稀疏索引，每隔一定间隔大小的offset才会建立相应的索引(比如每间隔10条消息建立一个索引)。timeindex也是一个稀疏索引文件，这样可以根据消息的时间找到对应的消息。

可以考虑将消息日志存放到多个磁盘中，这样多个磁盘可以并发访问，增加消息读写的吞吐量。这种情况下，logdirs配置的是一个目录列表，kafka会根据每个目录下partition的数量，将新分配的partition放到partition数最少的目录下。如果我们新增了一个磁盘，你会发现新分配的partition都出现在新增的磁盘上。

kafka提供了两个参数logsegmentbytes和logsegmentms来控制LogSegment文件的大小。logsegmentbytes默认值是1GB，当LogSegment大小达到logsegmentbytes规定的阈值时，kafka会关闭当前LogSegment，生成一个新的LogSegment供消息写入，当前供消息写入的LogSegment称为活跃（Active）LogSegment。logsegmentms表示最大多长时间会生成一个新的LogSegment，logsegmentms没有默认值。当这两个参数都配置了值，kafka看哪个阈值先达到，触发生成新的LogSegment。

kafka还提供了logretentionms和logretentionbytes两个参数来控制消息的保留时间。当消息的时间超过了logretentionms配置的阈值（默认是168小时，也就是一周），则会被认为是过期的，会被kafka自动删除。或者是partition的总的消息大小超过了logretentionbytes配置的阈值时，最老的消息也会被kafka自动删除，使相应partition保留的总消息大小维持在logretentionbytes阈值以下。这个地方需要注意的是，kafka并不是以消息为粒度进行删除的，而是以LogSegment为粒度删除的。也就是说，只有当一个LogSegment的最后一条消息的时间超过logretentionms阈值时，该LogSegment才会被删除。这两个参数都配置了值时，也是只要有一个先达到阈值，就会执行相应的删除策略

当我们使用KafkaProducer向kafka发送消息时非常简单，只要构造一个包含消息key、value、接收topic信息的ProducerRecord对象就可以通过KafkaProducer的send()向kafka发送消息了，而且是线程安全的。KafkaProducer支持通过三种消息发送方式

KafkaProducer客户端虽然使用简单，但是一条消息从客户端到topic partition的日志文件，中间需要经历许多的处理过程。KafkaProducer的内部结构如下所示：

从图中可以看出，消息的发送涉及两类线程，一类是调用KafkaProducersend()方法的应用程序线程，因为KafkaProducersend()是多线程安全的，所以这样的线程可以有多个；另一类是与kafka集群通信，实际将消息发送给kafka集群的Sender线程，当我们创建一个KafkaProducer实例时，会创建一个Sender线程，通过该KafkaProducer实例发送的所有消息最终通过该Sender线程发送出去。RecordAccumulator则是一个消息队列，是应用程序线程与Sender线程之间消息传递的桥梁。当我们调用KafkaProducersend()方法时，消息并没有直接发送出去，只是写入了RecordAccumulator中相应的队列中，最终需要Sender线程在适当的时机将消息从RecordAccumulator队列取出来发送给kafka集群。

消息的发送过程如下：

在使用KafkaConsumer实例消费kafka消息时，有一个特性我们要特别注意，就是KafkaConsumer不是多线程安全的，KafkaConsumer方法都在调用KafkaConsumer的应用程序线程中运行（除了consumer向kafka集群发送的心跳，心跳在一个专门的单独线程中发送），所以我们调用KafkaConsumer的所有方法均需要保证在同一个线程中调用，除了KafkaConsumerwakeup()方法，它设计用来通过其它线程向consumer线程发送信号，从而终止consumer执行。

跟producer一样，consumer要与kafka集群通信，消费kafka消息，首先需要获取消费的topic partition leader replica所在的broker地址等信息，这些信息可以通过向kafka集群任意broker发送Metadata请求消息获取。

我们知道，一个consumer group有多个consumer，一个topic有多个partition，而且topic的partition在同一时刻只能被consumer group内的一个consumer消费，那么consumer在消费partition消息前需要先确定消费topic的哪个partition。partition的分配通过group coordinator来实现。基本过程如下：

我们可以通过实现接口orgapachekafkaclientsconsumerinternalsPartitionAssignor自定义partition分配策略，但是kafka已经提供了三种分配策略可以直接使用。

partition分配完后，每个consumer知道了自己消费的topic partition，通过metadata请求可以获取相应partition的leader副本所在的broker信息，然后就可以向broker poll消息了。但是consumer从哪个offset开始poll消息？所以consumer在第一次向broker发送FetchRequest poll消息之前需要向Group Coordinator发送OffsetFetchRequest获取消费消息的起始位置。Group Coordinator会通过key {topic}-${partition}查询 __consumer_offsets topic中是否有offset的有效记录，如果存在，则将consumer所属consumer group最近已提交的offset返回给consumer。如果没有（可能是该partition是第一次分配给该consumer group消费，也可能是该partition长时间没有被该consumer group消费），则根据consumer配置参数autooffsetreset值确定consumer消费的其实offset。如果autooffsetreset值为latest，表示从partition的末尾开始消费，如果值为earliest，则从partition的起始位置开始消费。当然，consumer也可以随时通过KafkaConsumerseek()方法人工设置消费的起始offset。

kafka broker在收到FetchRequest请求后，会使用请求中topic partition的offset查一个skiplist表（该表的节点key值是该partition每个LogSegment中第一条消息的offset值）确定消息所属的LogSegment，然后继续查LogSegment的稀疏索引表（存储在index文件中），确定offset对应的消息在LogSegment文件中的位置。为了提升消息消费的效率，consumer通过参数fetchminbytes和maxpartitionfetchbytes告诉broker每次拉取的消息总的最小值和每个partition的最大值（consumer一次会拉取多个partition的消息）。当kafka中消息较少时，为了让broker及时将消息返回给consumer，consumer通过参数fetchmaxwaitms告诉broker即使消息大小没有达到fetchminbytes值，在收到请求后最多等待fetchmaxwaitms时间后，也将当前消息返回给consumer。fetchminbytes默认值为1MB，待fetchmaxwaitms默认值为500ms。

为了提升消息的传输效率，kafka采用零拷贝技术让内核通过DMA把磁盘中的消息读出来直接发送到网络上。因为kafka写入消息时将消息写入内存中就返回了，如果consumer跟上了producer的写入速度，拉取消息时不需要读磁盘，直接从内存获取消息发送出去就可以了。

为了避免发生再平衡后，consumer重复拉取消息，consumer需要将已经消费完的消息的offset提交给group coordinator。这样发生再平衡后，consumer可以从上次已提交offset出继续拉取消息。

kafka提供了多种offset提交方式

partition offset提交和管理对kafka消息系统效率来说非常关键，它直接影响了再平衡后consumer是否会重复拉取消息以及重复拉取消息的数量。如果offset提交的比较频繁，会增加consumer和kafka broker的消息处理负载，降低消息处理效率；如果offset提交的间隔比较大，再平衡后重复拉取的消息就会比较多。还有比较重要的一点是，kafka只是简单的记录每次提交的offset值，把最后一次提交的offset值作为最新的已提交offset值，作为再平衡后消息的起始offset，而什么时候提交offset，每次提交的offset值具体是多少，kafka几乎不关心（这个offset对应的消息应该存储在kafka中，否则是无效的offset），所以应用程序可以先提交3000，然后提交2000，再平衡后从2000处开始消费，决定权完全在consumer这边。

kafka中的topic partition与consumer group中的consumer的消费关系其实是一种配对关系，当配对双方发生了变化时，kafka会进行再平衡，也就是重新确定这种配对关系，以提升系统效率、高可用性和伸缩性。当然，再平衡也会带来一些负面效果，比如在再平衡期间，consumer不能消费kafka消息，相当于这段时间内系统是不可用的。再平衡后，往往会出现消息的重复拉取和消费的现象。

触发再平衡的条件包括：

需要注意的是，kafka集群broker的增减或者topic partition leader重新选主这类集群状态的变化并不会触发在平衡

有两种情况与日常应用开发比较关系比较密切：

consumer在调用subscribe()方法时，支持传入一个ConsumerRebalanceListener监听器，ConsumerRebalanceListener提供了两个方法，onPartitionRevoked()方法在consumer停止消费之后，再平衡开始之前被执行。可以发现，这个地方是提交offset的好时机。onPartitonAssigned()方法则会在重新进行partition分配好了之后，但是新的consumer还未消费之前被执行。

我们在提到kafka时，首先想到的是它的吞吐量非常大，这也是很多人选择kafka作为消息传输组件的重要原因。

以下是保证kafka吞吐量大的一些设计考虑：

但是kafka是不是总是这么快？我们同时需要看到kafka为了追求快舍弃了一些特性：

所以，kafka在消息独立、允许少量消息丢失或重复、不关心消息顺序的场景下可以保证非常高的吞吐量，但是在需要考虑消息事务、严格保证消息顺序等场景下producer和consumer端需要进行复杂的考虑和处理，可能会比较大的降低kafka的吞吐量，例如对可靠性和保序要求比较高的控制类消息需要非常谨慎的权衡是否适合使用kafka。

我们通过producer向kafka集群发送消息，总是期望消息能被consumer成功消费到。最不能忍的是producer收到了kafka集群消息写入的正常响应，但是consumer仍然没有消费到消息。

kafka提供了一些机制来保证消息的可靠传递，但是有一些因素需要仔细权衡考虑，这些因素往往会影响kafka的吞吐量，需要在可靠性与吞吐量之间求得平衡：

kafka只保证partition消息顺序，不保证topic级别的顺序，而且保证的是partition写入顺序与读取顺序一致，不是业务端到端的保序。

如果对保序要求比较高，topic需要只设置一个partition。这时可以把参数maxinflightrequestsperconnection设置为1，而retries设置为大于1的数。这样即使发生了可恢复型错误，仍然能保证消息顺序，但是如果发生不可恢复错误，应用层进行重试的话，就无法保序了。也可以采用同步发送的方式，但是这样也极大的降低了吞吐量。如果消息携带了表示顺序的字段，可以在接收端对消息进行重新排序以保证最终的有序。

有时需要简单的用shell去检查一个topic下边某一个partition的某个offset的消息。

之前一直用 kafkatoolsConsoleConsumer， 用的心酸。

后来发现一个稍强大的工具。

>

earliest： 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费

latest： 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

none： topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

简单来说，如果partition里已经有数据，但还没有消费，earliest就会从没消费的起始点来消费，反观latest就不会去消费；如果partition已经有已消费的数据，再放新的数据进去，那么它们都会从新的数据开始消费。

offset会保存在kafka内部，一开始发送数据到kafka的时候就有offset，只是有没有提交而已。而使用spring-kafka时，客户端在监听topic的时候，它有2种提交offset的方式：

1、自动提交，设置enableautocommit=true，更新的频率根据参数autocommitintervalms来定。这种方式也被称为at most once，fetch到消息后就可以更新offset，无论是否消费成功。

2、手动提交，设置enableautocommit=false，这种方式称为at least once。fetch到消息后，等消费完成再调用方法consumercommitSync()，手动更新offset；如果消费失败，则offset也不会更新，此条消息会被重复消费一次。

spring-kafka版本255，官网 >

以上就是关于Kafka的日志存储全部的内容，包括:Kafka的日志存储、Kafka的消息格式及offset是如何设置的、kafka系列之(3)——Coordinator与offset管理和Consumer Rebalance等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！