kafka是存储到本地磁盘么_一文讲清Kafka工作流程和存储机制

一、Kafka 文件存储机制

topic构成

Kafka 中消息是以 topic 进行分类的&＃xff0c;生产者生产消息&＃xff0c;消费者消费消息&＃xff0c;都是面向 topic 的。

在 Kafka 中&＃xff0c;一个 topic 可以分为多个 partition&＃xff0c;一个 partition 分为多个 segment&＃xff0c;每个 segment 对应两个文件&＃xff1a;.index 和 .log 文件

topic 是逻辑上的概念&＃xff0c;而 patition 是物理上的概念&＃xff0c;每个 patition 对应一个 log 文件&＃xff0c;而 log 文件中存储的就是 producer 生产的数据&＃xff0c;patition 生产的数据会被不断的添加到 log 文件的末端&＃xff0c;且每条数据都有自己的 offset。

消费组中的每个消费者&＃xff0c;都是实时记录自己消费到哪个 offset&＃xff0c;以便出错恢复&＃xff0c;从上次的位置继续消费。

消息存储原理

由于生产者生产的消息会不断追加到 log 文件末尾&＃xff0c;为防止 log 文件过大导致数据定位效率低下&＃xff0c;Kafka 采取了分片和索引机制&＃xff0c;将每个 partition 分为多个 segment。每个 segment 对应两个文件——.index文件和 .log文件。这些文件位于一个文件夹下&＃xff0c;该文件夹的命名规则为&＃xff1a;topic名称&＃43;分区序号。

如下&＃xff0c;我们创建一个只有一个分区一个副本的 topic

> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic starfish

然后可以在 kafka-logs 目录(server.properties 默认配置)下看到会有个名为 starfish-0 的文件夹。如果&＃xff0c;starfish 这个 topic 有三个分区&＃xff0c;则其对应的文件夹为 starfish-0&＃xff0c;starfish-1&＃xff0c;starfish-2。

这些文件的含义如下&＃xff1a;

index 和 log 文件以当前 segment 的第一条消息的 offset 命名。偏移量 offset 是一个 64 位的长整形数&＃xff0c;固定是20 位数字&＃xff0c;长度未达到&＃xff0c;用 0 进行填补&＃xff0c;索引文件和日志文件都由此作为文件名命名规则。所以从上图可以看出&＃xff0c;我们的偏移量是从 0 开始的&＃xff0c;.index 和 .log 文件名称都为 00000000000000000000。

接着往 topic 中发送一些消息&＃xff0c;并启动消费者消费

> bin /kafka-console-producer.sh --bootstrap-server localhost:9092 --topic starfishone

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic starfish --from-beginningone

查看 .log 文件下是否有数据 one

内容存在一些”乱码“&＃xff0c;因为数据是经过序列化压缩的。

那么数据文件 .log 大小有限制吗&＃xff0c;能保存多久时间&＃xff1f;这些我们都可以通过 Kafka 目录下 conf/server.properties 配置文件修改&＃xff1a;

# log文件存储时间&＃xff0c;单位为小时&＃xff0c;这里设置为1周log.retention.hours&＃61;168# log文件大小的最大值&＃xff0c;这里为1g&＃xff0c;超过这个值&＃xff0c;则会创建新的segment(也就是新的.index和.log文件)log.segment.bytes&＃61;1073741824

比如&＃xff0c;当生产者生产数据量较多&＃xff0c;一个 segment 存储不下触发分片时&＃xff0c;在日志 topic 目录下你会看到类似如下所示的文件&＃xff1a;

00000000000000000000.index00000000000000000000.log00000000000000170410.index00000000000000170410.log00000000000000239430.index00000000000000239430.log

下图展示了Kafka查找数据的过程&＃xff1a;

.index文件 存储大量的索引信息&＃xff0c;.log文件 存储大量的数据&＃xff0c;索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

比如现在要查找偏移量 offset 为 3 的消息&＃xff0c;根据 .index 文件命名我们可以知道&＃xff0c;offset 为 3 的索引应该从00000000000000000000.index 里查找。根据上图所示&＃xff0c;其对应的索引地址为 756-911&＃xff0c;所以 Kafka 将读取00000000000000000000.log 756~911区间的数据。

二、Kafka 生产过程

Kafka 生产者用于生产消息。通过前面的内容我们知道&＃xff0c;Kafka 的 topic 可以有多个分区&＃xff0c;那么生产者如何将这些数据可靠地发送到这些分区&＃xff1f;生产者发送数据的不同的分区的依据是什么&＃xff1f;针对这两个疑问&＃xff0c;这节简单记录下。

3.2.1 写入流程

producer 写入消息流程如下&＃xff1a;

1. producer 先从 zookeeper 的 "/brokers/.../state"节点找到该 partition 的 leader
2. producer 将消息发送给该 leader
3. leader 将消息写入本地 log
4. followers 从 leader pull 消息&＃xff0c;写入本地 log 后向 leader 发送 ACK
5. leader 收到所有 ISR 中的 replication 的 ACK 后&＃xff0c;增加 HW(high watermark&＃xff0c;最后 commit 的 offset)并向 producer 发送 ACK

2.1 写入方式

producer 采用推(push) 模式将消息发布到 broker&＃xff0c;每条消息都被追加(append) 到分区(patition) 中&＃xff0c;属于顺序写磁盘(顺序写磁盘效率比随机写内存要高&＃xff0c;保障 kafka 吞吐率)。

2.2 分区(Partition)

消息发送时都被发送到一个 topic&＃xff0c;其本质就是一个目录&＃xff0c;而 topic 是由一些 Partition Logs(分区日志)组成

分区的原因&＃xff1a;

1. 方便在集群中扩展&＃xff0c;每个 Partition 可以通过调整以适应它所在的机器&＃xff0c;而一个 topic 又可以有多个 Partition 组成&＃xff0c;因此整个集群就可以适应任意大小的数据了&＃xff1b;

2. 可以提高并发&＃xff0c;因为可以以 Partition 为单位读写了。

分区的原则&＃xff1a;

我们需要将 producer 发送的数据封装成一个 ProducerRecord 对象。

public ProducerRecord (String topic, Integer partition, Long timestamp, K key, V value, Iterable headers)public ProducerRecord (String topic, Integer partition, Long timestamp, K key, V value)public ProducerRecord (String topic, Integer partition, K key, V value, Iterable headers)public ProducerRecord (String topic, Integer partition, K key, V value)public ProducerRecord (String topic, K key, V value)public ProducerRecord (String topic, V value)

1. 指明 partition 的情况下&＃xff0c;直接将指明的值直接作为 partiton 值&＃xff1b;
2. 没有指明 partition 值但有 key 的情况下&＃xff0c;将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值&＃xff1b;
3. 既没有 partition 值又没有 key 值的情况下&＃xff0c;第一次调用时随机生成一个整数(后面每次调用在这个整数上自增)&＃xff0c;将这个值与 topic 可用的 partition 总数取余得到 partition 值&＃xff0c;也就是常说的 round-robin 算法。

2.3 副本(Replication)

同一个 partition 可能会有多个 replication( 对应 server.properties 配置中的 default.replication.factor&＃61;N)。没有 replication 的情况下&＃xff0c;一旦 broker 宕机&＃xff0c;其上所有 patition 的数据都不可被消费&＃xff0c;同时 producer 也不能再将数据存于其上的 patition。引入 replication 之后&＃xff0c;同一个 partition 可能会有多个 replication&＃xff0c;而这时需要在这些 replication 之间选出一 个 leader&＃xff0c; producer 和 consumer 只与这个 leader 交互&＃xff0c;其它 replication 作为 follower 从 leader 中复制数据。

2.4 数据可靠性保证

为保证 producer 发送的数据&＃xff0c;能可靠的发送到指定的 topic&＃xff0c;topic 的每个 partition 收到 producer 数据后&＃xff0c;都需要向 producer 发送 ack(acknowledgement确认收到)&＃xff0c;如果 producer 收到 ack&＃xff0c;就会进行下一轮的发送&＃xff0c;否则重新发送数据。

a) 副本数据同步策略主要有如下两种

Kafka 选择了第二种方案&＃xff0c;原因如下&＃xff1a;

• 同样为了容忍 n 台节点的故障&＃xff0c;第一种方案需要的副本数相对较多&＃xff0c;而 Kafka 的每个分区都有大量的数据&＃xff0c;第一种方案会造成大量的数据冗余&＃xff1b;
• 虽然第二种方案的网络延迟会比较高&＃xff0c;但网络延迟对 Kafka 的影响较小。

b) ISR

采用第二种方案之后&＃xff0c;设想一下情景&＃xff1a;leader 收到数据&＃xff0c;所有 follower 都开始同步数据&＃xff0c;但有一个 follower 挂了&＃xff0c;迟迟不能与 leader 保持同步&＃xff0c;那 leader 就要一直等下去&＃xff0c;直到它完成同步&＃xff0c;才能发送 ack&＃xff0c;这个问题怎么解决呢&＃xff1f;

leader 维护了一个动态的 in-sync replica set(ISR)&＃xff0c;意为和 leader 保持同步的 follower 集合。当 ISR 中的follower 完成数据的同步之后&＃xff0c;leader 就会给 follower 发送 ack。如果 follower 长时间未向 leader 同步数据&＃xff0c;则该 follower 将会被踢出 ISR&＃xff0c;该时间阈值由 replica.lag.time.max.ms 参数设定。leader 发生故障之后&＃xff0c;就会从 ISR 中选举新的 leader。(之前还有另一个参数&＃xff0c;0.9 版本之后 replica.lag.max.messages 参数被移除了)

c) ack应答机制

对于某些不太重要的数据&＃xff0c;对数据的可靠性要求不是很高&＃xff0c;能够容忍数据的少量丢失&＃xff0c;所以没必要等 ISR 中的follower全部接收成功。

所以Kafka为用户提供了三种可靠性级别&＃xff0c;用户根据对可靠性和延迟的要求进行权衡&＃xff0c;选择以下的acks 参数配置

• 0&＃xff1a;producer 不等待 broker 的 ack&＃xff0c;这一操作提供了一个最低的延迟&＃xff0c;broker 一接收到还没有写入磁盘就已经返回&＃xff0c;当 broker 故障时有可能丢失数据&＃xff1b;

• 1&＃xff1a;producer 等待 broker 的 ack&＃xff0c;partition 的 leader 落盘成功后返回 ack&＃xff0c;如果在 follower 同步成功之前 leader 故障&＃xff0c;那么将会丢失数据&＃xff1b;

• -1(all)&＃xff1a;producer 等待 broker 的 ack&＃xff0c;partition 的 leader 和 follower 全部落盘成功后才返回 ack。但是如果在 follower 同步完成后&＃xff0c;broker 发送 ack 之前&＃xff0c;leader 发生故障&＃xff0c;那么就会造成数据重复。

d) 故障处理

由于我们并不能保证 Kafka 集群中每时每刻 follower 的长度都和 leader 一致(即数据同步是有时延的)&＃xff0c;那么当leader 挂掉选举某个 follower 为新的 leader 的时候(原先挂掉的 leader 恢复了成为了 follower)&＃xff0c;可能会出现leader 的数据比 follower 还少的情况。为了解决这种数据量不一致带来的混乱情况&＃xff0c;Kafka 提出了以下概念&＃xff1a;

• LEO(Log End Offset)&＃xff1a;指的是每个副本最后一个offset&＃xff1b;
• HW(High Wather)&＃xff1a;指的是消费者能见到的最大的 offset&＃xff0c;ISR 队列中最小的 LEO。

消费者和 leader 通信时&＃xff0c;只能消费 HW 之前的数据&＃xff0c;HW 之后的数据对消费者不可见。

针对这个规则&＃xff1a;

• 当follower发生故障时&＃xff1a;follower 发生故障后会被临时踢出 ISR&＃xff0c;待该 follower 恢复后&＃xff0c;follower 会读取本地磁盘记录的上次的 HW&＃xff0c;并将 log 文件高于 HW 的部分截取掉&＃xff0c;从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW&＃xff0c;即 follower 追上 leader 之后&＃xff0c;就可以重新加入 ISR 了。
• 当leader发生故障时&＃xff1a;leader 发生故障之后&＃xff0c;会从 ISR 中选出一个新的 leader&＃xff0c;之后&＃xff0c;为保证多个副本之间的数据一致性&＃xff0c;其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉&＃xff0c;然后从新的 leader 同步数据。

所以数据一致性并不能保证数据不丢失或者不重复&＃xff0c;这是由 ack 控制的。HW 规则只能保证副本之间的数据一致性&＃xff01;

2.5 Exactly Once语义

将服务器的 ACK 级别设置为 -1&＃xff0c;可以保证 Producer 到 Server 之间不会丢失数据&＃xff0c;即 At Least Once 语义。相对的&＃xff0c;将服务器 ACK 级别设置为 0&＃xff0c;可以保证生产者每条消息只会被发送一次&＃xff0c;即 At Most Once语义。

At Least Once 可以保证数据不丢失&＃xff0c;但是不能保证数据不重复。相对的&＃xff0c;At Most Once 可以保证数据不重复&＃xff0c;但是不能保证数据不丢失。但是&＃xff0c;对于一些非常重要的信息&＃xff0c;比如说交易数据&＃xff0c;下游数据消费者要求数据既不重复也不丢失&＃xff0c;即 Exactly Once 语义。在 0.11 版本以前的 Kafka&＃xff0c;对此是无能为力的&＃xff0c;只能保证数据不丢失&＃xff0c;再在下游消费者对数据做全局去重。对于多个下游应用的情况&＃xff0c;每个都需要单独做全局去重&＃xff0c;这就对性能造成了很大的影响。

0.11 版本的 Kafka&＃xff0c;引入了一项重大特性&＃xff1a;幂等性。所谓的幂等性就是指 Producer 不论向 Server 发送多少次重复数据。Server 端都会只持久化一条&＃xff0c;幂等性结合 At Least Once 语义&＃xff0c;就构成了 Kafka 的 Exactily Once 语义&＃xff0c;即&＃xff1a;At Least Once &＃43; 幂等性 &＃61; Exactly Once

要启用幂等性&＃xff0c;只需要将 Producer 的参数中 enable.idompotence 设置为 true 即可。Kafka 的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer 在初始化的时候会被分配一个 PID&＃xff0c;发往同一 Partition 的消息会附带 Sequence Number。而 Broker 端会对 做缓存&＃xff0c;当具有相同主键的消息提交时&＃xff0c;Broker 只会持久化一条。

但是 PID 重启就会变化&＃xff0c;同时不同的 Partition 也具有不同主键&＃xff0c;所以幂等性无法保证跨分区会话的 Exactly Once。

三、Broker 保存消息

3.1 存储方式

物理上把 topic 分成一个或多个 patition(对应 server.properties 中的 num.partitions&＃61;3 配置)&＃xff0c;每个 patition 物理上对应一个文件夹(该文件夹存储该 patition 的所有消息和索引文件)。

3.2 存储策略

无论消息是否被消费&＃xff0c; kafka 都会保留所有消息。有两种策略可以删除旧数据&＃xff1a;

1. 基于时间&＃xff1a;log.retention.hours&＃61;168

2. 基于大小&＃xff1a;log.retention.bytes&＃61;1073741824 需要注意的是&＃xff0c;因为 Kafka 读取特定消息的时间复杂度为 O(1)&＃xff0c;即与文件大小无关&＃xff0c; 所以这里删除过期文件与提高 Kafka 性能无关。

四、Kafka 消费过程

Kafka 消费者采用 pull 拉模式从 broker 中消费数据。与之相对的 push(推)模式很难适应消费速率不同的消费者&＃xff0c;因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息&＃xff0c;但是这样很容易造成 consumer 来不及处理消息。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。

pull 模式不足之处是&＃xff0c;如果 kafka 没有数据&＃xff0c;消费者可能会陷入循环中&＃xff0c;一直返回空数据。为了避免这种情况&＃xff0c;我们在我们的拉请求中有参数&＃xff0c;允许消费者请求在等待数据到达的“长轮询”中进行阻塞(并且可选地等待到给定的字节数&＃xff0c;以确保大的传输大小&＃xff0c;或者传入等待超时时间)。

4.1 消费者组

消费者是以 consumer group 消费者组的方式工作&＃xff0c;由一个或者多个消费者组成一个组&＃xff0c; 共同消费一个 topic。每个分区在同一时间只能由 group 中的一个消费者读取&＃xff0c;但是多个 group 可以同时消费这个 partition。在图中&＃xff0c;有一个由三个消费者组成的 group&＃xff0c;有一个消费者读取主题中的两个分区&＃xff0c;另外两个分别读取一个分区。某个消费者读取某个分区&＃xff0c;也可以叫做某个消费者是某个分区的拥有者。

在这种情况下&＃xff0c;消费者可以通过水平扩展的方式同时读取大量的消息。另外&＃xff0c;如果一个消费者失败了&＃xff0c;那么其他的 group 成员会自动负载均衡读取之前失败的消费者读取的分区。

消费者组最为重要的一个功能是实现广播与单播的功能。一个消费者组可以确保其所订阅的 Topic 的每个分区只能被从属于该消费者组中的唯一一个消费者所消费&＃xff1b;如果不同的消费者组订阅了同一个 Topic&＃xff0c;那么这些消费者组之间是彼此独立的&＃xff0c;不会受到相互的干扰。

如果我们希望一条消息可以被多个消费者所消费&＃xff0c;那么可以将这些消费者放到不同的消费者组中&＃xff0c;这实际上就是广播的效果&＃xff1b;如果希望一条消息只能被一个消费者所消费&＃xff0c;那么可以将这些消费者放到同一个消费者组中&＃xff0c;这实际上就是单播的效果。

4.2 分区分配策略

一个 consumer group 中有多个 consumer&＃xff0c;一个 topic 有多个 partition&＃xff0c;所以必然会涉及到 partition 的分配问题&＃xff0c;即确定哪个 partition 由哪个 consumer 来消费。

Kafka 有两种分配策略&＃xff0c;一是 RoundRobin&＃xff0c;一是 Range。

RoundRobin

RoundRobin 即轮询的意思&＃xff0c;比如现在有一个三个消费者 ConsumerA、ConsumerB 和 ConsumerC 组成的消费者组&＃xff0c;同时消费 TopicA 主题消息&＃xff0c;TopicA 分为 7 个分区&＃xff0c;如果采用 RoundRobin 分配策略&＃xff0c;过程如下所示&＃xff1a;

这种轮询的方式应该很好理解。但如果消费者组消费多个主题的多个分区&＃xff0c;会发生什么情况呢&＃xff1f;比如现在有一个两个消费者 ConsumerA 和 ConsumerB 组成的消费者组&＃xff0c;同时消费 TopicA 和 TopicB 主题消息&＃xff0c;如果采用RoundRobin 分配策略&＃xff0c;过程如下所示&＃xff1a;

注&＃xff1a;TAP0 表示 TopicA Partition0 分区数据&＃xff0c;以此类推。

这种情况下&＃xff0c;采用 RoundRobin 算法分配&＃xff0c;多个主题会被当做一个整体来看&＃xff0c;这个整体包含了各自的 Partition&＃xff0c;比如在 Kafka-clients 依赖中&＃xff0c;与之对应的对象为 TopicPartition。接着将这些 TopicPartition 根据其哈希值进行排序&＃xff0c;排序后采用轮询的方式分配给消费者。

但这会带来一个问题&＃xff1a;假如上图中的消费者组中&＃xff0c;ConsumerA 只订阅了 TopicA 主题&＃xff0c;ConsumerB 只订阅了TopicB 主题&＃xff0c;采用 RoundRobin 轮询算法后&＃xff0c;可能会出现 ConsumerA 消费了 TopicB 主题分区里的消息&＃xff0c;ConsumerB 消费了 TopicA 主题分区里的消息。

综上所述&＃xff0c;RoundRobin 算法只适用于消费者组中消费者订阅的主题相同的情况。同时会发现&＃xff0c;采用 RoundRobin 算法&＃xff0c;消费者组里的消费者之间消费的消息个数最多相差 1 个。

Range

Kafka 默认采用 Range 分配策略&＃xff0c;Range 顾名思义就是按范围划分的意思。

比如现在有一个三个消费者 ConsumerA、ConsumerB 和 ConsumerC 组成的消费者组&＃xff0c;同时消费 TopicA 主题消息&＃xff0c;TopicA分为7个分区&＃xff0c;如果采用 Range 分配策略&＃xff0c;过程如下所示&＃xff1a;

假如现在有一个两个消费者 ConsumerA 和 ConsumerB 组成的消费者组&＃xff0c;同时消费 TopicA 和 TopicB 主题消息&＃xff0c;如果采用 Range 分配策略&＃xff0c;过程如下所示&＃xff1a;

Range 算法并不会把多个主题分区当成一个整体。

从上面的例子我们可以总结出Range算法的一个弊端&＃xff1a;那就是同一个消费者组内的消费者消费的消息数量相差可能较大。

4.3 offset 的维护

由于 consumer 在消费过程中可能会出现断电宕机等故障&＃xff0c;consumer 恢复后&＃xff0c;需要从故障前的位置继续消费&＃xff0c;所以 consumer 需要实时记录自己消费到了哪个 offset&＃xff0c;以便故障恢复后继续消费。

Kafka 0.9 版本之前&＃xff0c;consumer 默认将 offset 保存在 Zookeeper 中&＃xff0c;从 0.9 版本开始&＃xff0c;consumer 默认将 offset保存在 Kafka 一个内置的 topic 中&＃xff0c;该 topic 为 _consumer_offsets。

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic starfish --from-beginningone

消费 topic 后&＃xff0c;查看 kafka-logs 目录&＃xff0c;会发现多出 50 个分区。

默认情况下__consumer_offsets 有 50 个分区&＃xff0c;如果你的系统中 consumer group 也很多的话&＃xff0c;那么这个命令的输出结果会很多

五、Kafka事务

Kafka 从 0.11 版本开始引入了事务支持。事务可以保证 Kafka 在 Exactly Once 语义的基础上&＃xff0c;生产和消费可以跨分区和会话&＃xff0c;要么全部成功&＃xff0c;要么全部失败。

5.1 Producer事务

为了了实现跨分区跨会话的事务&＃xff0c;需要引入一个全局唯一的 TransactionID&＃xff0c;并将 Producer 获得的 PID 和Transaction ID 绑定。这样当 Producer 重启后就可以通过正在进行的 TransactionID 获得原来的 PID。

为了管理 Transaction&＃xff0c;Kafka 引入了一个新的组件 Transaction Coordinator。Producer 就是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。Transaction Coordinator 还负责将事务所有写入 Kafka 的一个内部 Topic&＃xff0c;这样即使整个服务重启&＃xff0c;由于事务状态得到保存&＃xff0c;进行中的事务状态可以得到恢复&＃xff0c;从而继续进行。

5.2 Consumer事务

对 Consumer 而言&＃xff0c;事务的保证就会相对较弱&＃xff0c;尤其是无法保证 Commit 的消息被准确消费。这是由于Consumer 可以通过 offset 访问任意信息&＃xff0c;而且不同的 SegmentFile 生命周期不同&＃xff0c;同一事务的消息可能会出现重启后被删除的情况。

来源&＃xff1a;JavaKeeper(ID&＃xff1a;JavaKeeper)