Kafka-工作流程及各个组件的工作机制

淡淡的烟草味﹌ 2022-09-09 00:23 147阅读 0赞

# Kafka 架构深入 #

## Kafka 工作流程及文件存储机制 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

Kafka 中消息是以 **topic** 进行分类的,生产者生产消息,消费者消费消息,都是面向 topic的。

topic 是逻辑上的概念,而 partition 是物理上的概念,每个 partition 对应于一个 log 文件,该 log 文件中存储的就是 producer 生产的数据。Producer 生产的数据会被不断追加到该log 文件末端,且每条数据都有自己的 offset。消费者组中的每个消费者,都会实时记录自己消费到了哪个 offset,以便出错恢复时,从上次的位置继续消费。

当生产这往一个不存在的topic发消息的时候，系统会自动创建改topic，默认是一个分区一个副本。

**文件存储的机制**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]

由于生产者生产的消息会不断追加到 log 文件末尾,为防止 log 文件过大导致数据定位效率低下,**Kafka 采取了分片（默认1G）和索引机制**,将每个 partition 分为多个 segment。

每个 segment对应两个文件——“.index”文件和“.log”文件。这些文件位于一个文件夹下,该文件夹的命名规则为: topic 名称+分区序号。例如, first 这个 topic 有三个分区,则其对应的文件夹为 first-0,first-1,first-2。

00000000000000000000.index
    00000000000000000000.log
    00000000000000170410.index
    00000000000000170410.log
    00000000000000239430.index
    00000000000000239430.log

index 和 log 文件以当前 segment 的第一条消息的 offset 命名。下图为 index 文件和 log文件的结构示意图。

**index文件和log文件详解**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]

**“.index”文件存储大量的索引信息,“.log”文件存储大量的数据**,索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

.log文件存储的数据量最大是1G，超过1G会产生一个新的.log文件

## Kafka 生产者 ##

### 生产者分区策略 ###

**1)分区的原因**

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

(2)**可以提高并发**,因为可以以 Partition 为单位读写了

**2)分区的原则**

我们需要将 producer 发送的数据封装成一个 **ProducerRecord** 对象。  
![在这里插入图片描述][421dff133d8c434c90e39deaf133d13f.png]

(1)指明 partition 的情况下,直接将指明的值直接作为 partiton 值;

(2)没有指明 partition 值但有 key 的情况下,将 key 的 hash 值与 topic 的 partition数进行取余得到 partition 值;

(3)既没有 partition 值又没有 key 值的情况下,第一次调用时随机生成一个整数(后面每次调用在这个整数上自增),将这个值与 topic 可用的 partition 总数取余得到 partition值,也就是常说的 round-robin 算法（轮询）。

### 数据可靠性保证 ###

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

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]

**1)副本数据同步策略**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 4]

Kafka 选择了第二种方案,原因如下:

1.同样为了容忍 n 台节点的故障,第一种方案需要 2n+1 个副本,而第二种方案只需要 n+1个副本,而 Kafka 的每个分区都有大量的数据,第一种方案会造成大量数据的冗余。

2.虽然第二种方案的网络延迟会比较高,但网络延迟对 Kafka 的影响较小。

**2)ISR**

采用第二种方案之后,设想以下情景: leader 收到数据,所有 follower 都开始同步数据,但有一个 follower,因为某种故障,迟迟不能与 leader 进行同步,那 leader 就要一直等下去,直到它完成同步,才能发送 ack。这个问题怎么解决呢?

**Leader 维护了一个动态的 in-sync replica set (ISR),意为和 leader 保持同步的 follower 集合。当 ISR 中的 follower 完成数据的同步之后, leader 就会给 follower 发送 ack。如果 follower长 时 间 未 向 leader 同 步 数 据 , 则 该 follower 将 被 踢 出 ISR，在这个时间内则将其加回来 , 该 时 间 阈 值 由replica.lag.time.max.ms 参数设定。Leader 发生故障之后,就会从 ISR 中选举新的 leader。**

**3)ack 应答机制**

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

所以 Kafka 为用户提供了三种可靠性级别,用户根据对可靠性和延迟的要求进行权衡,选择以下的配置。

**acks 参数配置:**

**acks:**

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

**1**:producer 等待 broker 的 ack,partition 的 leader 落盘成功后返回 ack,如果在 follower同步成功之前 leader 故障,那么将会**丢失数据**;

**acks=1数据丢失案例**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 5]

**-1(all)**:producer 等待 broker 的 ack,partition 的 leader 和 follower （ISR中的follower）全部落盘成功后才返回 ack。但是如果在 follower 同步完成后,broker 发送 ack 之前,leader 发生故障,那么会造成**数据重复**。

**acks=-1数据重复案例**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 6]

**4)故障处理细节**

**Log文件中的HW和LEO**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 7]

**LEO:指的是每个副本最大的 offset;**

\*\*HW:指的是消费者能见到的最大的 offset,ISR 队列中最小的 LEO。\*\*消费者只能看到以上的12，即HW。高水位。

**(1)follower 故障**

follower 发生故障后会被临时踢出 ISR,待该 follower 恢复后, follower 会读取本地磁盘记录的上次的 HW,并将 log 文件高于 HW 的部分截取掉,从 HW 开始向 leader 进行同步。等该 \*\*follower 的 LEO 大于等于该 Partition 的 HW,\*\*即 follower 追上 leader 之后,就可以重新加入 ISR 了。

**(2)leader 故障**

leader 发生故障之后,会从 ISR 中选出一个新的 leader,之后,为保证多个副本之间的数据存储一致性,其余的 follower 会先将各自的 log 文件**高于 HW 的部分截掉**,然后从新的 leader同步数据。

**注意:这只能保证副本之间的数据一致性,并不能保证数据不丢失或者不重复。**

### Exactly Once 语义 ###

精准一次性。

将服务器的 ACK 级别设置为-1,可以保证 Producer 到 Server 之间不会丢失数据,即 **AtLeast Once 语义**。

相对的,将服务器 ACK 级别设置为 0,可以保证生产者每条消息只会被发送一次,即 **At Most Once 语义**。

At Least Once 可以保证数据不丢失,但是不能保证数据不重复;相对的,At Least Once可以保证数据不重复,但是不能保证数据不丢失。

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

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

At Least Once + **幂等性** = Exactly Once

幂等性就是为了解决数据重复问题的，它作用于broker上，也就是说不需要各个消费者自己去去重了，它在kk就完成去重了。

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

但是 PID 重启就会变化,同时不同的 Partition 也具有不同主键,所以幂等性无法保证跨分区跨会话的 Exactly Once。幂等性只能解决不跨分区不跨会话的数据重复问题。

## Kafka 消费者 ##

### 消费方式 ###

**consumer 采用 pull(拉)模式从 broker 中读取数据。**

\*\*push (推)模式很难适应消费速率不同的消费者,因为消息发送速率是由 broker 决定的。\*\*它的目标是尽可能以最快速度传递消息,但是这样很容易造成 consumer 来不及处理消息,典型的表现就是拒绝服务以及网络拥塞。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。

**pull 模式不足之处是,如果 kafka 没有数据,消费者可能会陷入循环中,一直返回空数据**。针对这一点,Kafka 的消费者在消费数据时会传入一个时长参数 timeout,如果当前没有数据可供消费,consumer 会等待一段时间之后再返回,这段时长即为 timeout。

### 消费者分区分配策略 ###

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

同一个消费者组里面的消费者不能同时消费同一个分区，只能同时消费一个主题。

Kafka 有两种分配策略,一是 RoundRobin轮询,一是 Range。默认是Range策略

**1)RoundRobin**  
![在这里插入图片描述][0c1c05dd784a413dacf3c93154cf7c71.png]

轮询模式下（面向组），消费者组会将所订阅的所有topic下的所有分区做一个hash排序，然后再按照轮询的方式发给各个消费者。好处在于各个消费者之间消费的分区最多差一个；但是要保证消费者组里面的消费者订阅的topic是完全相同的，否则就会乱掉，消费者没有订阅的topic分区但是收到了对应的消息。

**2)Range**  
![在这里插入图片描述][5edf0bc329124d85b785dafbd9ad7e00.png]

Range的前提是按照单个主题来分的，可能会带来消费者消费数据不对等问题，比如：有两个topic1（0,1,2三个分区），topic2（10,11,12三个分区），消费者组中有两个消费者A，B，在Range策略下，A会分到0,1,10,11，而B只会分到2和12。

因此如果两个主题，AB都订阅了，那个可以采用轮询的方式；若A订阅了主题1，B订阅了主题2，那么采用Rang的方式。

什么时候会触发策略的执行呢？

当消费者组中消费者的个数发生变化时，都会触发策略的执行。

### offset 的维护 ###

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

通过组+主题+分区唯一确定一个offset，按照组来进行offset保存的。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 8]

**Kafka 0.9 版本之前,consumer 默认将 offset 保存在 Zookeeper 中,从 0.9 版本开始,consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中,该 topic 为\_\_consumer\_offsets。**

1)修改配置文件 consumer.properties

exclude.internal.topics=false
    ## 表示可以用普通的消费者，消费系统的主题

2)读取 offset

0.11.0.0 之前版本:
    bin/kafka-console-consumer.sh --topic __consumer_offsets -- zookeeper hadoop113:2181 --formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter" -- consumer.config config/consumer.properties --from-beginning
    
    0.11.0.0 之后版本(含):
    
    bin/kafka-console-consumer.sh --topic __consumer_offsets -- zookeeper hadoop113:2181 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning

### 消费者组案例 ###

需求:测试同一个消费者组中的消费者,同一时刻只能有一个消费者消费。

(1)在 hadoop113、hadoop114 上修改/opt/module/kafka/config/consumer.properties 配置文件中的 group.id 属性为任意组名。
    vi consumer.properties
    group.id=sanguo
    
    (2)在 hadoop113、hadoop114 上分别启动消费者
    bin/kafka-console-consumer.sh --zookeeper hadoop113:2181 --topic first --consumer.config config/consumer.properties
    
    bin/kafka-console-consumer.sh --bootstrap-server hadoop113:9092 --topic first --consumer.config config/consumer.properties
    
    (3)在 hadoop115 上启动生产者
    bin/kafka-console-producer.sh --broker-list hadoop113:9092 --topic first
    
    (4)查看 hadoop113 和 hadoop114 的接收者。
    同一时刻同一个消费者组中只有一个消费者接收到某个分区的数据；
    不同组之间可以共同消费同一主题；
    消费者组内消费者数量改变之后，会重新触发分区分配；

## Kafka 高效读写数据 ##

**1)顺序写磁盘**

Kafka 的 producer 生产数据,要写入到 log 文件中,写的过程是一直追加到文件末端,为顺序写。官网有数据表明,同样的磁盘,顺序写能到 600M/s,而随机写只有 100K/s。这与磁盘的机械机构有关,顺序写之所以快,是因为其省去了大量磁头寻址的时间。

**2)零复制技术**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_16_color_FFFFFF_t_70_g_se_x_16]

## Zookeeper 在 Kafka 中的作用 ##

Kafka 集群中有一个 broker 会被选举为 Controller（抢占）,负责管理集群 broker 的上下线,所有 topic 的**分区副本分配和 leader 选举**等工作。

Controller 的管理工作都是依赖于 Zookeeper 的。

以下为 partition 的 leader 选举过程:

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 9]

## Kafka 事务 ##

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

### Producer 事务 ###

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

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

### Consumer 事务 ###

保证的是精准一次性消费。

上述事务机制主要是从 Producer 方面考虑,对于 Consumer 而言,事务的保证就会相对较弱,尤其时无法保证 Commit 的信息被精确消费。这是由于 Consumer 可以通过 offset 访问任意信息,而且不同的 Segment File 生命周期不同,同一事务的消息可能会出现重启后被删除的情况。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/4384e14aaa394fdba7c4790cd55eb85e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/8b2b62303a854960a819350f9519a2ad.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/5fb9c3776b8645daa3c1ad231f5c6309.png
[421dff133d8c434c90e39deaf133d13f.png]: /images/20220829/106a4c47e43d4a29bef725bc7133a5e1.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/442ab62c61f740bba0dc982813b90ba9.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/77f03d09cb654fe1983a233fd2f44260.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220829/be4a04ece6504821bbea7ea8d4be1f7f.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 6]: /images/20220829/3bead6dc0134449db01ccec2020a347e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 7]: /images/20220829/486340c91de3472ea126e35168a3a612.png
[0c1c05dd784a413dacf3c93154cf7c71.png]: /images/20220829/ca861dc0a3db437f8e5381482d443cde.png
[5edf0bc329124d85b785dafbd9ad7e00.png]: /images/20220829/58992d0ac903492490b5f6941ea6a25b.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 8]: /images/20220829/24eded42fe404f089e8dd541a674b82e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_16_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/ac9bbdf969a444dba110d808bc959532.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 9]: /images/20220829/a5bb1be1cb8d49fc95f811cdc492632d.png