Kafka--原理--工作流程

蔚落 2022-11-05 03:06 488阅读 0赞

原文网址：[Kafka--原理--工作流程\_IT利刃出鞘的博客-CSDN博客][Kafka--_--_IT_-CSDN]

# 简介 #

本文介绍Kafka的工作流程（工作原理）。

# 总体流程 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70][]

# 生产者推送消息 #

## **1 写入方式** ##

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

## **2 分区（Partition）** ##

Kafka集群有多个消息代理服务器（broker-server）组成，发布到Kafka集群的每条消息都有一个类别，用主题（topic）来表示。通常，不同应用产生不同类型的数据，可以设置不同的主题。一个主题一般会有多个消息的订阅者，当生产者发布消息到某个主题时，订阅了这个主题的消费者都可以接收到生成者写入的新消息。

Kafka集群为每个主题维护了分布式的分区（partition）日志文件，物理意义上可以把主题（topic）看作进行了分区的日志文件（partition log）。主题的每个分区都是一个有序的、不可变的记录序列，新的消息会不断追加到日志中。分区中的每条消息都会按照时间顺序分配到一个单调递增的顺序编号，叫做偏移量（offset），这个偏移量能够唯一地定位当前分区中的每一条消息。

消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成，其组织结构如下图所示：

下图中的topic有3个分区，每个分区的偏移量都从0开始，不同分区之间的偏移量都是独立的，不会相互影响。

![70][]

![70 1][]

我们可以看到，每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值。

发布到Kafka主题的每条消息包括键值和时间戳。消息到达服务器端的指定分区后，都会分配到一个自增的偏移量。原始的消息内容和分配的偏移量以及其他一些元数据信息最后都会存储到分区日志文件中。消息的键也可以不用设置，这种情况下消息会均衡地分布到不同的分区。

**分区的原因**

（1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；  
（2）可以提高并发，因为可以以Partition为单位读写了。

传统消息系统在服务端保持消息的顺序，如果有多个消费者消费同一个消息队列，服务端会以消费存储的顺序依次发送给消费者。但由于消息是异步发送给消费者的，消息到达消费者的顺序可能是无序的，这就意味着在并行消费时，传统消息系统无法很好地保证消息被顺序处理。虽然我们可以设置一个专用的消费者只消费一个队列，以此来解决消息顺序的问题，但是这就使得消费处理无法真正执行。

Kafka比传统消息系统有更强的顺序性保证，它使用主题的分区作为消息处理的并行单元。Kafka以分区作为最小的粒度，将每个分区分配给消费者组中不同的而且是唯一的消费者，并确保一个分区只属于一个消费者，即这个消费者就是这个分区的唯一读取线程。那么，只要分区的消息是有序的，消费者处理的消息顺序就有保证。每个主题有多个分区，不同的消费者处理不同的分区，所以Kafka不仅保证了消息的有序性，也做到了消费者的负载均衡。

**分区的原则**

（1）指定了patition，则直接使用；  
（2）未指定patition但指定key，通过对key的value进行hash出一个patition  
（3）patition和key都未指定，使用轮询选出一个patition。

DefaultPartitioner类

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (keyBytes == null) {
            int nextValue = nextValue(topic);
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            if (availablePartitions.size() > 0) {
                int part = Utils.toPositive(nextValue) % availablePartitions.size();
                return availablePartitions.get(part).partition();
            } else {
                // no partitions are available, give a non-available partition
                return Utils.toPositive(nextValue) % numPartitions;
            }
        } else {
            // hash the keyBytes to choose a partition
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }

## **3 副本（Replication）** ##

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

## **4 写入流程** ##

producer写入消息流程如下：

![70 2][]

1）producer先从zookeeper的 “/brokers/…/state”节点找到该partition的leader  
2）producer将消息发送给该leader  
3）leader将消息写入本地log  
4）followers从leader pull消息，写入本地log后向leader发送ACK  
5）leader收到所有ISR中的replication的ACK后，增加HW（high watermark，最后commit 的offset）并向producer发送ACK

# Broker存储 #

## 1 存储方式 ##

由于Producer产生的消息会不断的追加到日志文件的末尾，这样将对消息文件的维护以及以消费的消息的清理带来严重的影响，因此，Kafka引入的分片和索引的设计。每个Partition物理上对应一个文件夹（该文件夹存储该patition的所有消息和索引文件）：“topic名称-分区序号”。每个Partition分为多个Segment,Segment分为两类文件：“.index”索引文件与“.log”数据文件，其中索引文件和数据文件都在Partition对应的文件夹中。

**实际例子**

[joker@hadoop102 logs]$ ll
    drwxrwxr-x. 2 joker joker 4096 8月 6 14:37 first-0
    drwxrwxr-x. 2 joker joker 4096 8月 6 14:35 first-1
    drwxrwxr-x. 2 joker joker 4096 8月 6 14:37 first-2
    [joker@hadoop102 logs]$ cd first-0
    [joker@hadoop102 first-0]$ ll
    -rw-rw-r–. 1 joker joker 10485760 8月 6 14:33 00000000000000000000.index
    -rw-rw-r–. 1 joker joker 219 8月 6 15:07 00000000000000000000.log
    -rw-rw-r–. 1 joker joker 10485756 8月 6 14:33 00000000000000000000.timeindex
    -rw-rw-r–. 1 joker joker 8 8月 6 14:37 leader-epoch-checkpoint

**图片示意**

![da3a444b18f808c660f71cdae59a4fff.png][]

假设test-topic有3个分区，则对应的文件夹名称为：test-topic-0、test-topic-1、test-topic-2。 partition文件夹下文件形如：

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

可以看到有索引文件与数据文件，有3个Segment。 这两个文件的命令规则为：Partition全局的第一个Segment从0开始，后续每个Segment文件名为上一个Segment文件最后一条消息的offset值，数值大小为64位，20位数字字符长度，没有数字用0填充。

Segment文件的详细内容：

![d4d1e0d960b8f775ea061fcb4c427771.png][]

索引文件存储的元数据指向数据文件中的message的物理偏移地址。

**从partition中通过offset查找message**

以上图为例，读取offset=170418的消息，首先查找segment文件，其中00000000000000000000.index 为最开始的文件，第二个文件为 00000000000000170410.index（起始偏移为 170410 1=170411），而第三个文件为 00000000000000239430.index（起始偏移为 239430 1=239431），所以这个 offset=170418 就落到了第二个文件之中。其它后续文件可以依次类推，以其偏移量命名并排列这些文件，然后根据二分查找法就可以快速定位到具体文件位置。其次根据 00000000000000170410.index 文件中的 \[8,1325\] 定位到 00000000000000170410.log 文件中的 1325 的位置进行读取。要是读取 offset=170418 的消息，从 00000000000000170410.log 文件中的 1325的位置进行读取，那么，如何确定何时读完本条消息呢？ 这个问题由消息的物理结构解决，消息都具有固定的物理结构，包括：offset（8 Bytes）、消息体的大小（4 Bytes）、crc32（4 Bytes）、magic（1 Byte）、attributes（1 Byte）、key length（4 Bytes）、key（K Bytes）、payload（N Bytes）等等字段，可以确定一条消息的大小，即读取到哪里截止。

## 2 存储策略 ##

无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据：  
1）基于时间：log.retention.hours=168  
2）基于大小：log.retention.bytes=1073741824  
需要注意的是，因为Kafka读取特定消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高 Kafka 性能无关。

## 3 Zookeeper存储结构 ##

注意：producer不在zk中注册，消费者在zk中注册。

# 消费者拉取消息 #

kafka提供了两套consumer API：高级Consumer API和低级API。

## 1 消费模型 ##

消息由生产者发布到Kafka集群后，会被消费者消费。消息的消费模型有两种：推送模型（push）和拉取模型（pull）。

基于推送模型（push）的消息系统，由消息代理记录消费者的消费状态。消息代理在将消息推送到消费者后，标记这条消息为已消费，但这种方式无法很好地保证消息被处理。比如，消息代理把消息发送出去后，当消费进程挂掉或者由于网络原因没有收到这条消息时，就有可能造成消息丢失（因为消息代理已经把这条消息标记为已消费了，但实际上这条消息并没有被实际处理）。如果要保证消息被处理，消息代理发送完消息后，要设置状态为“已发送”，只有收到消费者的确认请求后才更新为“已消费”，这就需要消息代理中记录所有的消费状态，这种做法显然是不可取的。

Kafka采用拉取模型，由消费者自己记录消费状态，每个消费者互相独立地顺序读取每个分区的消息。如下图所示，有两个消费者（不同消费者组）拉取同一个主题的消息，消费者A的消费进度是3，消费者B的消费进度是6。消费者拉取的最大上限通过最高水位（watermark）控制，生产者最新写入的消息如果还没有达到备份数量，对消费者是不可见的。这种由消费者控制偏移量的优点是：消费者可以按照任意的顺序消费消息。比如，消费者可以重置到旧的偏移量，重新处理之前已经消费过的消息；或者直接跳到最近的位置，从当前的时刻开始消费。

![70 3][]

在一些消息系统中，消息代理会在消息被消费之后立即删除消息。如果有不同类型的消费者订阅同一个主题，消息代理可能需要冗余地存储同一消息；或者等所有消费者都消费完才删除，这就需要消息代理跟踪每个消费者的消费状态，这种设计很大程度上限制了消息系统的整体吞吐量和处理延迟。Kafka的做法是生产者发布的所有消息会一致保存在Kafka集群中，不管消息有没有被消费。用户可以通过设置保留时间来清理过期的数据，比如，设置保留策略为两天。那么，在消息发布之后，它可以被不同的消费者消费，在两天之后，过期的消息就会自动清理掉。

## 2 高级API ##

**1）高级API优点**

高级API 写起来简单  
不需要自行去管理offset，系统通过zookeeper自行管理。  
不需要管理分区，副本等情况，.系统自动管理。  
消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据（默认设置1分钟更新一下zookeeper中存的offset）  
可以使用group来区分对同一个topic 的不同程序访问分离开来（不同的group记录不同的offset，这样不同程序读取同一个topic才不会因为offset互相影响）

**2）高级API缺点**

不能自行控制offset（对于某些特殊需求来说）  
不能细化控制如分区、副本、zk等

## 3 低级API ##

**1）低级 API 优点**

能够让开发者自己控制offset，想从哪里读取就从哪里读取。  
自行控制连接分区，对分区自定义进行负载均衡  
对zookeeper的依赖性降低（如：offset不一定非要靠zk存储，自行存储offset即可，比如存在文件或者内存中）

**2）低级API缺点**

太过复杂，需要自行控制offset，连接哪个分区，找到分区leader 等。

## 4 消费者组 ##

![70 4][]

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

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

## 5 消费方式 ##

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

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

对于Kafka而言，pull模式更合适，它可简化broker的设计，consumer可自主控制消费消息的速率，同时consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。

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

## 6 消费者组案例 ##

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

2）案例实操  
（1）在hadoop102、hadoop103上修改/opt/module/kafka/config/consumer.properties配置文件中的group.id属性为任意组名。  
\[joker@hadoop103 config\]$ vi consumer.properties  
group.id=joker

（2）在hadoop102、hadoop103上分别启动消费者  
\[joker@hadoop102 kafka\]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties  
\[joker@hadoop103 kafka\]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties

（3）在hadoop104上启动生产者  
\[joker@hadoop104 kafka\]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first  
hello world

（4）查看hadoop102和hadoop103的接收者。  
同一时刻只有一个消费者接收到消息。

# 其他网址 #

[Kafka工作流程分析\_Da.的博客-CSDN博客\_kafka工作流程][Kafka_Da._-CSDN_kafka]  
[Kafka简介、基本原理、执行流程与使用场景\_徐刘根的博客-CSDN博客\_kafka简介及使用][Kafka_-CSDN_kafka]  
[深入了解Kafka【二】工作流程及文件存储机制][Kafka]

[Kafka--_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/114332733
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70]: /images/20221023/36f4a287afdd4e93b2e10b70cd4630fd.png
[70]: /images/20221023/3ff75c3e27c74cfcb0943798bb42904d.png
[70 1]: /images/20221023/3a2eb884029648bd8cc9b3dfd38b5c59.png
[70 2]: /images/20221023/91fefcaff0fe4be29f5017e77e59c2cb.png
[da3a444b18f808c660f71cdae59a4fff.png]: /images/20221023/f5b15ece03594b02895469994e49c64e.png
[d4d1e0d960b8f775ea061fcb4c427771.png]: /images/20221023/ce9bb53059954e75aeff2ad7f8e1b5c4.png
[70 3]: /images/20221023/b079d05d969f4f62956a970b6c9a9eac.png
[70 4]: /images/20221023/3d0a2084d85645f5b286665f4e9065df.png
[Kafka_Da._-CSDN_kafka]: https://blog.csdn.net/qq_35641192/article/details/80956244
[Kafka_-CSDN_kafka]: https://blog.csdn.net/xlgen157387/article/details/77266719
[Kafka]: https://juejin.cn/post/6844904009388523528