二、Kafka 工作流程及生产者、消费者

以你之姓@ 2023-06-03 15:53 12阅读 0赞

**目录**

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

1、工作流程

2、文件存储机制

二、Kafka 生产者

1、分区策略

2、数据可靠性保证

（1）副本数据同步策略

2） ISR

3） ack 应答机制

4）故障处理细节

5）、Exactly Once 语义

三、Kafka 消费者

1、消费方式

2、分区分配策略（待研究）

3、offset 的维护

4、消费者组案例

--------------------

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

## 1、工作流程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

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

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

## 2、文件存储机制 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

由于生产者生产的消息会不断追加到 log 文件末尾， **为防止 log 文件过大导致数据定位效率低下， Kafka 采取了分片和索引机制，**将每个 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文件的结构示意图。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2][]

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

# 二、Kafka 生产者 #

## 1、分区策略 ##

**1）分区的原因**

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

**（2） 可以提高并发，**因为可以**以 Partition 为单位读写**了。

**2） 分区的原则**

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

![2019100623575336.png][]

（1） **指明 partition** 的情况下，直接将指明的值直接作为 partiton 值；

（2）**没有指明 partition 值但有 key 的情况下，将** **key 的 hash 值与 topic 的 partition数进行取余得到 partition 值**；

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 3][]

### （1）副本数据同步策略 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 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 故障，那么将会丢失数据；

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

### 4）故障处理细节 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 5][]

**LEO：指的是每个副本最大的 offset；**

**HW：指的是消费者能见到的最大的 offset， ISR 队列中最小的 LEO。**

**a、 follower 故障**

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

**b、 leader 故障**

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

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

### 5）、Exactly Once 语义 ###

将服务器的 **ACK 级别设置为-1**，可以保证 Producer 到 Server 之间不会丢失数据，即 **At Least 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

[ kafka 幂等生产者及事务(kafka0.11之后版本新特性)][kafka _kafka0.11]

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

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

# 三、Kafka 消费者 #

## **1、消费方式** ##

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

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

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

## **2、分区分配策略（待研究）** ##

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

Kafka 有两种分配策略，一是 RoundRobin，一是 Range。

**RoundRobin**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 6][]

**Range**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 7][]

## 3、offset 的维护 ##

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 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 hadoop102: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 hadoop102:2181 --formatter
    "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageForm
    atter" --consumer.config config/consumer.properties --frombeginning

## 4、消费者组案例 ##

**1） 需求：测试同一个消费者组中的消费者， 同一时刻只能有一个消费者消费。**

**2） 案例实操**

（1）在 hadoop102、 hadoop103 上修改/opt/module/kafka/config/consumer.properties 配置文件中的 group.id 属性为任意组名。

vi consumer.properties
    group.id=atguigu

（2）在 hadoop102、 hadoop103 上分别启动消费者

bin/kafka-console-consumer.sh \
    --zookeeper hadoop102:2181 --topic first --consumer.config
    config/consumer.properties
    [atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --
    bootstrap-server hadoop102:9092 --topic first --consumer.config
    config/consumer.properties

（3）在 hadoop104 上启动生产者

bin/kafka-console-producer.sh \
    --broker-list hadoop102:9092 --topic first
    >hello world

4）查看 hadoop102 和 hadoop103 的接收者。

同一时刻只有一个消费者接收到消息。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: /images/20230601/6c6e9248b52944f69db71cc38d7923f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1]: /images/20230601/bd51572ddd0542debe2a5490a8f9c169.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2]: /images/20230601/0421b422bdcc400585dbd66598b71074.png
[2019100623575336.png]: /images/20230601/8c4b346372134943bcd6f0d6937981a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 3]: /images/20230601/b88bb361342247f38ccc416215dec086.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 4]: /images/20230601/93cf12d031d44af6984756174973048a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 5]: /images/20230601/b352db90147b42339fef89af0ae76f67.png
[kafka _kafka0.11]: https://www.cnblogs.com/felixzh/p/10184762.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 6]: /images/20230601/212b7043ce654ae9a631c84b3f5213b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 7]: /images/20230601/2d9fe6bce73c44109d50dbdaca8f320d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 8]: /images/20230601/f5c51ce21cfc40948a5bad1dbaf3f62f.png