Kafka：broker如何处理请求

桃扇骨 2024-03-24 14:18 101阅读 0赞

## 前言 ##

![format_png][]

生产者生产消息，调用Kafka Client提供的API通过send方法发送到broker之后，broker会根据消息**请求类型**有对应的处理手法，**并且不止是生产者投递的消息，还有broker与broker之间也会传递请求**。

首先讲一下Kafka的请求类型。

## 请求类型 ##

### 元数据请求 ###

> 解决客户端去找分区首领所在哪个broker的问题

元数据请求里面包含了：

*  客户端**感兴趣的主题列表**。
 *  服务器端的响应消息里指明了这些**主题所包含的分区**，
 *  每个**分区**都有哪些**副本**，
 *  以及哪个**副本是首领**。

有了这些元数据，客户端就知道该向哪个broker发送生产请求或者获取请求。

一般客户端都会把这些信息缓存起来。时不时会通过**发送元数据请求**来刷新这些信息。

当发生分区变化，比如新的broker加入，那么部分分区就会被分配到新的broker里，可能会有首领。

此时如果客户端还拿着旧的元数据信息，那么就会发生请求失败，返回一个**非首领错误**。

客户端在重试之前，会先发送元数据请求，去同步一下自己的元数据信息。然后再进行重试。

**总结：其实就是Redis Clusters里面的全局key映射关系**

### 生产请求 ###

> 生产者发送的请求，可以理解为**Kafka写入消息请求**

如果之前有配置那个acks，那么首先会进行一个**写入消息检查**。

包含首领副本的broker收到生产请求后，会对请求做一些验证：

*  对方是否具有主题的写入权限。
 *  请求里面包含的acks值是否有效。
 *  如果是acks=all，是否有足够多的同步副本保证消息已经被写入。

之后，消息会被写入**本地磁盘**。

在消息被写入分区的首领之后，broker开始检查acks配置参数，如果是0，或者是1，就立客返回响应。如果是all。那么会先把请求保存到一个叫做**炼狱**的`缓冲区`里，直到首领发现所有跟随者副本都复制了消息，响应才会返回给客户端。（ZAB里面是过半就行，这里要全部）

### 获取请求 ###

> 获取请求**解决的是跟随者跟首领数据一致性**的问题

跟随者向首领发出请求，请求同步数据一致性。

跟随者会指定某主题下的某分区的**偏移量**一个区间的消息，让首领发送给他。

首领在收到请求后，**会先检查请求是否有效**，比如偏移量是否在分区上存在等等。

**客户端可以指定broker最多可以从一个分区里面返回多少数据**，因为服务器返回数据给我客户端，我客户端需要预分配一些内存空间来存储这些消息的。

如果是合法的偏移量，那么broker就会按照获取请求上面的要求，从分区里读取消息，然后返回给客户端。

**Kafka使用零复制技术向客户端发送消息。** 其实就是直接把**消息从文件（Linux文件系统缓存）里发送到网络通道**，不需要经过任何中间缓冲区。

一般来说，其他的数据库会在把数据发送给客户端之前，会把他们保存在**本地缓存**里。

## 处理请求 ##

> 这里讲的处理请求，是指**broker处理客户端、分区副本和控制器发送给分区首领的请求**。

所有的请求消息都包含一个**标准消息头**：

*  `Reqiest type`：（也就是API key）
 *  `Requiest version`：（broker**可以处理不同版本的客户端请求**，并根据客户0端版本做出不同的响应）
 *  `Correlation ID`：具有**唯一性**的数据，**用于标识请求消息**，同时也会出现在响应消息和错误日志里面（用于诊断问题）
 *  `Client ID`：客户端Id

类似http一样，也是**请求-响应模式**。

broker是按照**请求到达的顺序**来处理他们。

![format_png 1][]

请求的类型：

*  `生产请求`：生产者发送的消息，包含客户端要写入Kafka的消息
 *  `获取请求`：跟随者向首领获取信息的请求。

#### Acceptor线程 ####

> 用来**创建连接**的

broker会在它所监听的每一个端口上运行一个**Acceptor线程**，这个线程会创建一个连接，并把它交给**Processor线程**去处理。

Processor线程可以有一个或者多个。数量可以配置。

#### Processor线程 ####

> 用来**搬运**请求的，也被称为**网络线程**

网络线程把来自客户端的请求，搬运到**请求队列**中，从**响应队列**中搬运消息到指定的客户端。他是一个消息搬运工。

#### IO线程 ####

> 处理请求队列中的消息，然后存入响应队列

请求消息被放到请求队列后，IO线程会负责处理他们

`生产请求`和`获取请求`都**必须发送给分区的首领副本**。所以，如果一个broker收到一个针对特定分区的请求，而该分区的首领在另一个broker上，那么就会收到一个错误响应。**叫做“非首领”错误**。

**所以Kafka客户端需要自己负责把生产请求和获取请求发送到正确的broker上**。（redis cluster集群里面是通过全局key表和moved指令来实现这个功能的）

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/140d5474f367464a97d2e8450910880e.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/66e88ef286304f8dbffc2123749c3baa.png