kafka架构生产、消费、broker如何分工

末蓝、 2022-11-03 11:09 94阅读 0赞

### 分工合作 ###

我们将kafka的分成三个部分：

*  生产者
 *  存储Broker
 *  消费者

然后看看这三个部分都分别做什么事情。

### kafka生产者都干了什么事情 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3JhbmRvbXBlb3BsZQ_size_16_color_FFFFFF_t_70]  
这是一张生产者大图，我们可以从一下几点来看下：

#### KafkaProducer：生产消息 ####

对消息塞上Topic、分区、key、消息内容等数据。这里就可以了解到消息分区的逻辑其实是在客户端这边处理的，并不是在broker上确定分区的，这里主要看分区计算的逻辑：

*  对于指定了 key 的消息，partition 位置的计算方式为：Utils.murmur2(key) % numPartitions，即先对 key 进行哈希计算，然后在于 partition 个数求余，从而得到该条消息应该被存储在哪个 partition 上。
 *  对于没有指定 key 的消息，partition 位置的计算方式为：采用 round-robin 方式确定 partition 位置，即采用轮询的方式，平均的将消息分布到不同的 partition 上，从而避免某些 partition 数据量过大影响 Broker 和消费端性能。

#### RecordAccumulator：消息队列 ####

RecordAccumulator 是一个记录收集器，用于收集客户端发送的消息，并将收集到的消息暂存到客户端缓存中。  
这里暂存其实是一个队列，将消息放到队列中， N 条消息组成一批，一批消息最多存储 N 条，超过后则新建一个组来存储新消息。

#### Sender：发送 ####

第一步：扫描记录收集器中满足条件的批数据，然后将 partition -> 批数据映射转换成 BrokerId -> N 批数据的映射。  
第二步：Sender 线程会为每个 BrokerId 创建一个客户端请求，然后将请求交给 NetWorkClient，由 NetWrokClient 去真正发送网络请求到 Broker。

### broker 要做什么事情 ###

#### 幂等性 ####

为了保证消息过来的重复性问题，一半都会有幂等判断，Kafka引入了Producer ID（即PID）和Sequence Number。对于每个PID，该Producer发送消息的每个<Topic, Partition>都对应一个单调递增的Sequence Number。同样，Broker端也会为每个<PID, Topic, Partition>维护一个序号，并且每Commit一条消息时将其对应序号递增。对于接收的每条消息，如果其序号比Broker维护的序号）大一，则Broker会接受它，否则将其丢弃。

#### 偏移量 ####

每个分区中的消息被赋予了唯一整数标识，称之为偏移量，Broker仅提供基于offset的读取方式，不维护consumer的消费offset，以前的zookeeper可以存储consumer的消费偏移量。

#### 数据存储 ####

partition物理上由多个segment文件组成，每个segment大小相等，顺序读写。每个segment数据文件以该段中最小的offset命名，文件扩展名为.log，同时还有一个同名的以.index结尾的索引文件，不过该文件不是对每条消息都有索引，而是每隔一定字节的数据建立一条索引。这样在查找指定offset的Message的时候，用二分查找就可以定位到该Message在哪个segment数据文件中。然后加载对应的.index索引文件到内存，同样二分法找出小于等于给定offset的最大的那个offset记录（相对offset，position）；最后，根据position到.log文件中，顺序查找出offset等于给定offset值的消息。在顺序查找具体的消息。也就是2次二分查找，一次顺序查找。

#### 高性能写入磁盘 ####

Kafka Broker 在收到消息时会返回一个响应，如果写入成功，会返回一个 RecordMetaData 对象，它包含了主题和分区信息，以及记录在分区里的偏移量，上面两种的时间戳类型也会返回给用户。如果写入失败，会返回一个错误。生产者在收到错误之后会尝试重新发送消息，几次之后如果还是失败的话，就返回错误消息。  
先写操作系统的页缓存（Page Cache），然后由操作系统自行决定何时刷到磁盘。

因此 Kafka 达到高吞吐、低延迟的原因主要有以下 4 点：

*  页缓存是在内存中分配的，所以消息写入的速度很快。而且kafka broker 奔溃的时候，数据还不丢失，重启后还可以继续提供服务
 *  Kafka 不必和底层的文件系统进行交互，所有繁琐的 I/O 操作都由操作系统来处理。
 *  Kafka 采用追加写的方式，避免了磁盘随机写操作。
 *  使用以 sendfile 为代表的零拷贝技术提高了读取数据的效率。

Broker中保存的数据是有有效期的，比如7天，一旦超过了有效期，对应的数据将被移除以释放磁盘空间。只要数据在有效期内，Consumer可以重复读取而不受限制。

#### Replication设计 ####

### 消费者做什么 ###

*  连接 ZK 集群，拿到对应 topic 的 partition 信息和 partition 的 leader 的相关信息
 *  连接到对应 leader 对应的 broker
 *  consumer 将自己保存的 offset 发送给 leader
 *  leader 根据 offset 等信息定位到 segment（索引文件和日志文件）
 *  根据索引文件中的内容，定位到日志文件中该偏移量对应的开始位置读取相应长度的数据并返回给 consumer

如果一个新的消费者来消费TOPIC，那么以前的消息如何：

*  （1）使用一个全新的"group.id"（就是之前没有被任何消费者使用过）;
 *  （2）指定"auto.offset.reset"参数的值为earliest；  
    从何处开始消费,latest 表示消费最新消息,earliest 表示从头开始消费,none表示抛出异常,默认latest。

注意：从kafka-0.9版本及以后，kafka的消费者组和offset信息就不存zookeeper了，而是存到broker服务器上，所以，如果你为某个消费者指定了一个消费者组名称（group.id），那么，一旦这个消费者启动，这个消费者组名和它要消费的那个topic的offset信息就会被记录在broker服务器上。

### 参考博客 ###

[kafka基础知识][kafka]  
[重要：Kafka第3篇之一条消息如何被存储到Broker上][Kafka_3_Broker]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3JhbmRvbXBlb3BsZQ_size_16_color_FFFFFF_t_70]: /images/20221024/033c4c33933a427f844addebf39902f8.png
[kafka]: https://www.cnblogs.com/cxuanBlog/p/11949238.html
[Kafka_3_Broker]: https://zhuanlan.zhihu.com/p/137811719