1. Kafka生产、消费数据工作流程

1.1 Kafka数据写入流程

• 生产者先从 zookeeper 的 "/brokers/topics/主题名/partitions/分区名/state"节点找到该 partition 的leader
  

• 生产者在ZK中找到该ID找到对应的broker
  

• broker进程上的leader将消息写入到本地log中

• follower从leader上拉取消息，写入到本地log，并向leader发送ACK

• leader接收到所有的ISR中的Replica的ACK后，并向生产者返回ACK。

1.2 Kafka数据消费流程

1.2.1 两种消费模式

• kafka采用拉取模型，由消费者自己记录消费状态，每个消费者互相独立地顺序拉取每个分区的消息
• 消费者可以按照任意的顺序消费消息。比如，消费者可以重置到旧的偏移量，重新处理之前已经消费过的消息；或者直接跳到最近的位置，从当前的时刻开始消费。

1.2.2 Kafka消费数据流程

• 每个consumer都可以根据分配策略（默认RangeAssignor），获得要消费的分区
• 获取到consumer对应的offset（默认从ZK中获取上一次消费的offset）
• 找到该分区的leader，拉取数据
• 消费者提交offset

1.2.3 总结

• 写流程
  • 通过ZooKeeper找partition对应的leader，leader是负责写的
  • producer开始写入数据
  • ISR里面的follower开始同步数据，并返回给leader ACK
  • 返回给producer ACK
• 读流程
  • 通过ZooKeeper找partition对应的leader，leader是负责读的
  • 通过ZooKeeper找到消费者对应的offset
  • 然后开始从offset往后顺序拉取数据
  • 提交offset（自动提交——每隔多少秒提交一次offset、手动提交——放入到事务中提交）

2. Kafka的数据存储形式

• 一个topic由多个分区组成
• 一个分区（partition）由多个segment（段）组成
• 一个segment（段）由多个文件组成（log、index、timeindex）

2.1 存储日志

接下来，我们来看一下Kafka中的数据到底是如何在磁盘中存储的。

• Kafka中的数据是保存在配置文件的log.dirs对应的目录下
• 消息是保存在以：「主题名-分区ID」的文件夹中的

# 比如
drwxr-xr-x 2 root root  141 May 19 18:17 TEST-1
drwxr-xr-x 2 root root  141 May 19 18:17 TEST-2
drwxr-xr-x 2 root root  141 May 19 18:17 topic_event-0
drwxr-xr-x 2 root root  141 May 19 18:17 topic_start-0

• 数据文件夹中包含以下内容：

[root@node2 kafka-logs]# cd test-0
[root@node1 test-0]# ll -h
total 16M
-rw-r--r-- 1 root root  10M May 20 15:32 00000000000000000000.index
-rw-r--r-- 1 root root 9.9M May 20 15:32 00000000000000000000.log
-rw-r--r-- 1 root root  10M May 20 15:32 00000000000000000000.timeindex
-rw-r--r-- 1 root root    8 May 20 15:30 leader-epoch-checkpoint

这些分别对应：

文件名	说明
00000000000000000000.index	索引文件，根据offset查找数据就是通过该索引文件来操作的
00000000000000000000.log	日志数据文件
00000000000000000000.timeindex	时间索引
leader-epoch-checkpoint	持久化每个partition leader对应的LEO（log end offset、日志文件中下一条待写入消息的offset）

• 每个日志文件的文件名为起始偏移量，因为每个分区的起始偏移量是0，所以，分区的日志文件都以0000000000000000000.log开始
• 默认的每个日志文件最大为「log.segment.bytes =102410241024」1G
• 为了简化根据offset查找消息，Kafka日志文件名设计为开始的偏移量
• 每个index、log、timeindex文件就是一个段

2.2 观察测试

为了方便测试观察，新创建一个topic：「test_10m」，该topic每个日志数据文件最大为10M

kafka-topics.sh --create --zookeeper node1.itcast.cn --topic test_10m --replication-factor 2 --partitions 3 --config segment.bytes=10485760

使用之前的生产者程序往「test_10m」主题中生产数据，可以观察到如下：

2.3 写入消息

• 新的消息总是写入到最后的一个日志文件中
• 该文件如果到达指定的大小（默认为：1GB）时，将滚动到一个新的文件中
  

2.4 读取消息

• 根据「offset」首先需要找到存储数据的 segment 段（注意：offset指定分区的全局偏移量）
• 然后根据这个「全局分区offset」找到相对于文件的「segment段offset」
  
• 最后再根据 「segment段offset」读取消息
• 为了提高查询效率，每个文件都会维护对应的范围内存，查找的时候就是使用简单的二分查找

2.5 总结

• Kafka的数据组织结构
  • topic
  • partition
  • segment
    • .log数据文件
    • .index（稀疏索引）
    • .timeindex（根据时间做的索引）
• 深入了解读数据的流程
  • 消费者的offset是一个针对partition全局offset
  • 可以根据这个offset找到segment段
  • 接着需要将全局的offset转换成segment的局部offset
  • 根据局部的offset，就可以从（.index稀疏索引）找到对应的数据位置
  • 开始顺序读取