字节二面：能说说 Kafka 处理请求的流程么？越详细越好

超、凢脫俗 2023-09-28 13:17 75阅读 0赞

今天来讲讲 `Kafka Broker` 端处理请求的全流程，剖析下底层的网络通信是如何实现的、Reactor 在 kafka 上的应用。

再说说社区为何在 2.3 版本将请求类型划分成两大类，又是如何实现两类请求处理的优先级。

## 叨叨 ##

不过在进入今天主题之前我想先叨叨几句，就源码这个事儿，不同人有不同的看法。

有些人听到源码这两个词就被吓到了，这么多代码怎么看。奔进去就像无头苍蝇，一路断点跟下来，跳来跳去，算了拜拜了您嘞。

而有些人觉得源码有啥用，看了和没看一样，看了也用不上。

其实上面两种想法我都有过，哈哈哈。那为什么我会开始看 `Kafka` 源码呢？

其实就是我有个同事在自学 `go`，然后想用 go 写个消息队列，在画架构图的时候就来问我，这消息队列好像有点东西啊，消息收发，元数据管理，消息如何持久一堆问题过来，我直呼顶不住。

这市面上 `Kafka`、`RocketMQ` 都是现成的方案，于是乎我就看起了源码。

所以促使我看源码的初始动力，竟然是为了在同事前面装逼！！

我是先看了 `RocketMQ`，因为毕竟是 `Java` 写的，而 `Kafka Broker` 都是 `scala` 写的。

梳理了一波 `RocketMQ` 之后，我又想看看 `Kafka` 是怎么做的，于是乎我又看起了 `Kafka`。

在源码分析之前我先总结性的说了说 `Kafka` 底层的通信模型。应对面试官询问 `Kafka` 请求全过程已经够了。

## Reactor 模式 ##

在扯到 `Kafka` 之前我们先来说说 `Reactor模式`，基本上只要是底层的高性能网络通信就离不开 `Reactor模式`。像 Netty、Redis 都是使用 `Reactor模式`。

像我们以前刚学网络编程的时候以下代码可是非常的熟悉，新来一个请求，要么在当前线程直接处理了，要么新起一个线程处理。

![format_png][]

在早期这样的编程是没问题的，但是随着互联网的快速发展，单线程处理不过来，也不能充分的利用计算机资源。

而每个请求都新起一个线程去处理，资源的要求就太高了，并且创建线程也是一个重操作。

说到这有人想到了，那搞个线程池不就完事了嘛，还要啥 `Reactor`。 ![format_png 1][]

池化技术确实能缓解资源的问题，但是池子是有限的，池子里的一个线程不还是得候着某个连接，等待指示嘛。现在的互联网时代早已突破 `C10K` 了。

因此引入的 `IO多路复用`，由**一个线程来监视一堆连接**，同步等待一个或多个 IO 事件的到来，然后将事件分发给对应的 `Handler` 处理，这就叫 `Reactor模式`。

网络通信模型的发展如下 > 单线程 => 多线程 => 线程池 => Reactor 模型

Kafka 所采用的 `Reactor模型`如下 ![图来自Doug Lea大神的 Scalable IO in Java][Doug Lea_ Scalable IO in Java]

## Kafka Broker 网络通信模型 ##

简单来说就是，Broker 中有个 `Acceptor(mainReactor)` 监听新连接的到来，与新连接建连之后轮询选择一个 `Processor(subReactor)` 管理这个连接。

而 `Processor` 会监听其管理的连接，当事件到达之后，读取封装成 `Request`，并将 `Request` 放入共享请求队列中。

然后 IO 线程池不断的从该队列中取出请求，执行真正的处理。处理完之后将响应发送到对应的 `Processor` 的响应队列中，然后由 `Processor` 将 `Response` 返还给客户端。

每个 `listener` 只有一个 `Acceptor线程`，因为它只是作为新连接建连再分发，没有过多的逻辑，很轻量，一个足矣。

`Processor` 在 Kafka 中称之为网络线程，默认网络线程池有 3 个线程，对应的参数是 `num.network.threads`。并且可以根据实际的业务动态增减。

还有个 IO 线程池，即 `KafkaRequestHandlerPool`，执行真正的处理，对应的参数是 `num.io.threads`，默认值是 8。IO 线程处理完之后会将 `Response` 放入对应的 `Processor` 中，由 `Processor` 将响应返还给客户端。

可以看到网络线程和 IO 线程之间利用的经典的生产者 - 消费者模式，不论是用于处理 Request 的共享请求队列，还是 IO 处理完返回的 Response。

这样的好处是什么？生产者和消费者之间解耦了，可以对生产者或者消费者做独立的变更和扩展。并且可以平衡两者的处理能力，例如消费不过来了，我多加些 IO 线程。

如果你看过其他中间件源码，你会发现生产者 - 消费者模式真的是太常见了，所以面试题经常会有手写一波生产者 - 消费者。

![format_png 2][]

## 源码级别剖析网络通信模型 ##

Kafka 网络通信组件主要由两大部分构成：**SocketServer** 和 **KafkaRequestHandlerPool**。

### SocketServer ###

![format_png 3][] 可以看出 `SocketServer` 旗下管理着，`Acceptor 线程`、`Processor 线程`和 `RequestChannel`等对象。

`data-plane` 和 `control-plane` 稍后再做分析，先看看 `RequestChannel` 是什么。

#### RequestChannel ####

![format_png 4][] 关键的属性和方法都已经在下面代码中注释了，可以看出这个对象主要就是**管理 `Processor`和作为传输 `Request` 和 `Response` 的中转站**。

### Acceptor ###

接下来我们再看看 `Acceptor`

![format_png 5][]

可以看到它继承了 `AbstractServerThread`，接下来再看看它 run 些啥

![format_png 6][] 再来看看 `accept(key)` 做了啥 ![format_png 7][]

很简单，标准 `selector` 的处理，获取准备就绪事件，调用 `serverSocketChannel.accept()` 得到 `socketChannel`，将 `socketChannel` 交给通过轮询选择出来的 `Processor`，之后由它来处理 IO 事件。 \#\#Processor 接下来我们再看看 `Processor`，相对而言比 `Acceptor` 复杂一些。

先来看看三个关键的成员

![format_png 8][]

再来看看主要的处理逻辑。

![format_png 9][]

可以看到 `Processor` 主要是将底层读事件 IO 数据封装成 `Request` 存入队列中，然后将 IO 线程塞入的 `Response`，返还给客户端，并处理 `Response` 的回调逻辑。

\#KafkaRequestHandlerPool

IO 线程池，实际处理请求的线程。

![format_png 10][]

再来看看 IO 线程都干了些啥

![format_png 11][]

很简单，核心就是不断的从 `requestChannel` 拿请求，然后调用 handle 处理请求。

`handle` 方法是位于 `KafkaApis` 类中，可以理解为通过 `switch`，根据请求头里面不同的 `apikey` 调用不同的 `handle` 来处理请求。

![format_png 12][]

我们再举例看下较为简单的处理 `LIST_OFFSETS` 的过程，即 `handleListOffsetRequest`，来完成一个请求的闭环。

我用红色箭头标示了调用链。表明处理完请求之后是塞给对应的 `Processor` 的。 ![format_png 13][]

最后再来个更详细的总览图，把源码分析到的类基本上都对应的加上去了。

![format_png 14][]

## 请求处理优先级 ##

上面提到的 `data-plane` 和 `control-plane` 是时候揭开面纱了。这两个对应的就是数据类请求和控制类请求。

为什么需要分两类请求呢？直接在请求里面用 key 标明请求是要读写数据啊还是更新元数据不就行了吗？

简单点的说比如我们想删除某个 topic，我们肯定是想这个 topic 马上被删除的，而此时 producer 还一直往这个 topic 写数据，那这个情况可能是我们的删除请求排在第 N 个... 等前面的写入请求处理好了才轮到删除的请求。实际上前面哪些往这个 topic 写入的请求都是没用的，平白的消耗资源。

再或者说进行 `Preferred Leader` 选举时候，`producer` 将 `ack` 设置为 `all` 时候，`老leader` 还在等着 `follower` 写完数据向他报告呢，谁知 `follower` 已经成为了`新leader`，而通知它 leader 已经变更的请求由于被一堆数据类型请求堵着呢，`老leader` 就傻傻的在等着，直到超时。

就是为了解决这种情况，社区将请求分为两类。

那如何让控制类的请求优先被处理？优先队列？

社区采取的是两套 `Listener`，即数据类型一个 `listener`，控制类一个 `listener`。

对应的就是我们上面讲的**网络通信模型，在 kafka 中有两套！** kafka 通过两套监听变相的实现了请求优先级，毕竟数据类型请求肯定很多，控制类肯定少，这样看来控制类肯定比大部分数据类型先被处理！

迂回战术啊。

控制类的和数据类区别就在于，就一个 `Porcessor线程`，并且请求队列写死的长度为 20。

## 最后 ##

看源码主要就是得耐心，耐心跟下去。然后再跳出来看。你会发现不过如此，哈哈哈。

感谢阅读，更多的java课程学习路线,笔记,面试等架构资料，需要的同学可以私信我（资料）即可免费获取！

![b99c864442b1495fb943f4a218ecc9f1.jpeg][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS0zNzBlZjAwYTUxZmNjYTU1LnBuZw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS1lYTQwMmFmOTUxYjYyZDAzLnBuZw?x-oss-process=image/format,png
[Doug Lea_ Scalable IO in Java]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS05YWNlYWMzM2E1NWQxOTIwLnBuZw?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS1iM2ZjODA3Y2VlYTExNDQyLnBuZw?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS03MGU1Nzg2N2U2YzU3YTk3LnBuZw?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS0wYzAyOGE3MWRhZDUwYmYxLnBuZw?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS0wNzg0NzA1MWZiMjg5OGY4LnBuZw?x-oss-process=image/format,png
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS01YjMzNzhhNzhlMDZkYjZkLnBuZw?x-oss-process=image/format,png
[format_png 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS04YjViMTAyMWQ0YjkyODg3LnBuZw?x-oss-process=image/format,png
[format_png 8]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS1mYjBmODBlMzg5ZTkwNWFkLnBuZw?x-oss-process=image/format,png
[format_png 9]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS01NGI4N2Q1MGYyYmZhNjMxLnBuZw?x-oss-process=image/format,png
[format_png 10]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS03MzIzYTc5ZDA1YjZlZDQ0LnBuZw?x-oss-process=image/format,png
[format_png 11]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS1jZWVmYTNjNzZiOTlhOTA4LnBuZw?x-oss-process=image/format,png
[format_png 12]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS1mOTYzYTJkZTE4OTBhYzk1LnBuZw?x-oss-process=image/format,png
[format_png 13]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS0xZWMyMGE4N2M4ZTU4NDg0LnBuZw?x-oss-process=image/format,png
[format_png 14]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xNjAzNDI3OS00ZTI0OTZiZDc5M2VmNjc5LnBuZw?x-oss-process=image/format,png
[b99c864442b1495fb943f4a218ecc9f1.jpeg]: https://img-blog.csdnimg.cn/b99c864442b1495fb943f4a218ecc9f1.jpeg