Kafka的基本概念，结构，持久化

Posted 2021-05-02 进击的大杂烩

Kafka® is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.

kafka 的基本概念

• Broker: 缓存代理，Kafa 集群中的一台或多台服务器统称为 broker

• Topic: 每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

• Partition：Topic 物理上的分组，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列。partition 中的每条消息都会被分配一个有序的 id（offset）

• Producer: 消息的生产者，负责发布消息到Kafka broker

• Consumer: 消息的消费者，向Kafka broker读取消息的客户端

• Consumer Group: 消费者组，可以并行消费Topic中partition的消息，每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

• Message：消息，是通信的基本单位，每个 producer 可以向一个 topic（主题）发布一些消息

kafka 拓扑结构

如上图所示，一个典型的Kafka集群中包含若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息

数据持久化

1. Topic & Partition

Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息和索引文件。若创建一个topic test01，且有3分区(partition)，2个副本(replication-factor均为2)则整个集群上会相应会生成6个文件夹（本文所用集群共3个节点），如下表所示。

节点	目录
192.168.100.101
192.168.100.102
192.168.100.103

通过命令也能验证这一点

# ./bin/kafka-topics.sh --zookeeper 192.168.100.101:2181/kafka --describe --topic test01
Topic:test01    PartitionCount:3        ReplicationFactor:2     Configs:
        Topic: test01   Partition: 0    Leader: 2       Replicas: 2,1   Isr: 2,1
        Topic: test01   Partition: 1    Leader: 3       Replicas: 3,2   Isr: 2,3
        Topic: test01   Partition: 2    Leader: 1       Replicas: 1,3   Isr: 3,1

2. 持久化的表现

• 任何发布到相应 Topic 的消息都会根据策略(轮训或Hash)写入到这个 Topic 的一个 Partition 中，并被直接追加到log文件的尾部，每条消息再文件中的位置成为offset(偏移量，相当于唯一标识符)，partition 是以文件的形式存储在文件系统中。

• log文件根据broker配置文件server.properties中的配置项{log.retention.hours}和{log.retention.bytes}的相应配置来自动清理log文件。

3. 日志文件

在kafka的日志存储 partition 的目录中。日志文件的名字由该日志文件第一条消息的offset number + .log 组成。为了提高log文件消息的检索速度，每个日志文件(.log)对应一个索引文件(.index)，索引文件。并不会对每条消息建立索引，而是每隔一定字节数(根据broker配置文件server.properties中的配置项{log.index.interval.bytes})建立一条索引(目的是减少索引文件的大小)。每个日志文件的大小与broker配置文件server.properties中的配置项{log.segment.bytes}一致。一个partition的索引示意图：

索引index文件是二进制的数据结构是：4Bytes的offset相对偏移，4Bytes的物理文件偏移。通过下面的操作来查看index文件。

# hexdump -C data/test01-2/00000000000000000000.index

输出如下，由于文件过长，截取开始两行：
00000000  00 00 da e1 00 10 3e 5e  00 01 b5 bf 00 20 7c 15  |......>^..... |.|
00000010  00 02 28 a5 00 30 b3 48  00 02 2c a1 00 40 d7 58  |..(..0.H..,..@.X|
可见第一条索引的offset为：00 00 da e1，对应的文件中的偏移(seek)为：00 10 3e 5e
解析为十进制为：56033 --> 1064542

参考:
[kafka官方文档]http://kafka.apache.org/