Kafka源码分析(四) - Server端-请求处理框架

news/2024/10/7 20:27:38

系列文章目录

https://zhuanlan.zhihu.com/p/367683572

一. 总体结构

先给一张概览图：

服务端请求处理过程涉及到两个模块：kafka.network和kafka.server。

1.1 kafka.network

该包是kafka底层模块，提供了服务端NIO通信能力基础。

有4个核心类：SocketServer、Acceptor、Processor、RequestChannel。各自角色如下：

  • SocketServer：服务端的抽象，是服务端通信的入口；

  • Acceptor：Reactor通信模式中处理连接ACCEPT事件的线程/线程池所执行的任务；

  • Processor：Reactor通信模式中处理连接可读/可写事件的线程/线程池所执行的任务；

  • RequestChannel：请求队列，存储已经解析好的请求以等待处理；

对于上层模块而言，该基础模块有两个输入和一个输出

  1. 输入：IP+端口号，该模块会对目标端口实现监听；

  2. 输出：解析好的请求，通过RequestChannel进行输出；

  3. 输入：待发送的Response，通过Processor.responseQueue来完成输入；

1.2 kafka.server

该包在kafka.network的基础上实现各种请求的处理逻辑，主要包含KafkaServer和KafkaApis两个类。其中：

  • KafkaServer：Kafka服务端的抽象，统一维护Kafka服务端的各流程和状态；

  • KakfaApis：维护了各类请求对应的业务逻辑，通过KafkaServer.apis字段组合到KafkaServer之中；

二. Server的端口监听

整体流程如图：

接下来按调用顺序依次分析各方法

2.1 KafkaServer.startup()

关于端口监听的核心逻辑分4步，代码如下(用注释说明各部分的目的)：

def startup() {// 省略无关代码... ...// 1. 创建SocketServersocketServer = new SocketServer(config, metrics, time, credentialProvider)// 2. 启动端口监听// (在这里完成了Acceptor的创建和端口ACCEPT事件的监听)// (startupProcessors = false表示暂不启动Processor处理线程)socketServer.startup(startupProcessors = false)// 3. 启动请求处理过程中的相关依赖// (这也是第2步中不启动Processor处理线程的原因，有依赖项需要处理)... ...// 4. 启动端口可读/可写事件处理线程(即Processor线程)socketServer.startProcessors()// 省略无关代码... ...
}

2.2 SocketServer.startup(Boolean)

代码及说明性注释如下：

def startup(startupProcessors: Boolean = true) {this.synchronized {// 省略无关代码... ...// 1. 创建Accetpor和Processor的实例,// 同时页完成了Acceptor对端口ACCEPT事件的监听createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)// 2. [可选]启动各Acceptor对应的Processor线程if (startupProcessors) {startProcessors()}}
}

2.3 ScocketServer.createAcceptorAndProcessor()

直接上注释版的代码，流程分3步：

// 入参解释
// processorsPerListener: 对于每个IP:Port, 指定Reactor模式子线程池大小, 
//                        即处理端口可读/可写事件的线程数(Processor线程);
// endpoints: 接收请求的IP:Port列表;
def createAcceptorAndProcessors(processorsPerListener: Int,endpoints: Seq[EndPoint]): Unit = synchronized {// 省略无关代码... ...endpoints.foreach { endpoint =>// 省略无关代码... ...// 1. 创建Acceptor对象// 在此步骤中调用Acceptor.openServerSocket, 完成了对端口ACCEPT事件的监听val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)// 2. 创建了与acceptor对应的Processor对象列表// (这里并未真正启动Processor线程)addProcessors(acceptor, endpoint, processorsPerListener)// 3. 启动Acceptor线程KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()// 省略无关代码... ...}}

2.4 Acceptor.openServerSocket()

该方法中没什么特殊点，就是java NIO的标准流程：

def openServerSocket(host: String, port: Int): ServerSocketChannel = {// 1. 构建InetSocketAddress对象val socketAddress =if (host == null || host.trim.isEmpty)new InetSocketAddress(port)elsenew InetSocketAddress(host, port)// 2. 构建ServerSocketChannel对象, 并设置必要参数值val serverChannel = ServerSocketChannel.open()serverChannel.configureBlocking(false)if (recvBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)serverChannel.socket().setReceiveBufferSize(recvBufferSize)// 3. 端口绑定, 实现事件监听try {serverChannel.socket.bind(socketAddress)info("Awaiting socket connections on %s:%d.".format(socketAddress.getHostString, serverChannel.socket.getLocalPort))} catch {case e: SocketException =>throw new KafkaException("Socket server failed to bind to %s:%d: %s.".format(socketAddress.getHostString, port, e.getMessage), e)}// 4. 返回ServerSocketChannel对象, 用于后续register到Selector中serverChannel}

2.5 SocketServer.startProcessor()

从这步开始，仅剩的工作就是启动Processor线程，代码都非常简单。比如本方法只是遍历Acceptor列表，并调用Acceptor.startProcessors()

def startProcessors(): Unit = synchronized {acceptors.values.asScala.foreach { _.startProcessors() }info(s"Started processors for ${acceptors.size} acceptors")
}

2.6 Acceptor.startProcessors()

该方法很简明，直接上代码

def startProcessors(): Unit = synchronized {if (!processorsStarted.getAndSet(true)) {startProcessors(processors)}
}def startProcessors(processors: Seq[Processor]): Unit = synchronized {processors.foreach { processor =>KafkaThread.nonDaemon(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",processor).start()}
}

三. 请求/响应的格式

3.1 格式概述

请求和响应都由两部分组成：Header和Body。RequestHeader中包含ApiKey、ApiVersion、CorrelationId、ClientId；ResponseHeader中只包含CorrelationId字段。接下来逐个讲解这些字段。

  • ApiKey

    2字节整型，指明请求的类型；比如0代表Produce请求，1代表Fetch请求；具体id和请求类型之间的映射关系可在 org.apache.kafka.common.protocol.ApiKeys 中找到；

  • ApiVersion

    随着API的升级迭代，各类型请求的请求体格式可能有变更；这个2字节的整型指明了请求体结构的版本；

  • CorrelationId

    4字节整型，在Response中传回，Kafka Server端不处理，用于客户端内部关联业务数据；

  • ClientId

    可变长字符串，标识客户端；

3.2 请求体/响应体的具体格式

各业务操作(比如Produce、Fetch等)对应的请求体和响应体格式都维护在 org.apache.kafka.common.protocol.ApiKeys 中。接下来以Produce为例讲解ApiKeys是如何表达数据格式的。

ApiKeys是个枚举类，其核心属性如下：

public enum ApiKeys {// 省略部分代码... ...// 上文提到的请求类型对应的idpublic final short id;// 业务操作名称public final String name;// 各版本请求体格式public final Schema[] requestSchemas;// 各版本响应体格式public final Schema[] responseSchemas;// 省略部分代码... ...
}

其中PRODUCE枚举项的定义如下

PRODUCE(0, "Produce", ProduceRequest.schemaVersions(), ProduceResponse.schemaVersions())

可以看到各版本的请求格式维护在 ProduceRequest.schemaVersions()，代码如下

public static Schema[] schemaVersions() {return new Schema[] {PRODUCE_REQUEST_V0, PRODUCE_REQUEST_V1, PRODUCE_REQUEST_V2, PRODUCE_REQUEST_V3,PRODUCE_REQUEST_V4, PRODUCE_REQUEST_V5, PRODUCE_REQUEST_V6};
}

这里只是简单返回了一个Schema数组。一个Schema对象代表了一种数据格式。请求头中的ApiVersion指明了请求体的格式对应数组的第几项(从0开始)。

接下来我们看看Schema是如何表达数据格式的。其结构如下

Schema有两个字段：fields和fieldsByName。其中fields是体现数据格式的关键，它指明了字段的排序和各字段类型；而fieldsByName只是按字段名重新组织的Map，用于根据名称查找对应字段。

BoundField只是Field的简单封装。Field有两个核心字段：name和type。其中name表示字段名称，type表示字段类型。常见的Type如下：

Type.BOOLEAN;
Type.INT8;
Type.INT16;
Type.INT32;// 可通过org.apache.kafka.common.protocol.types.Type查看全部类型
... ...

回到PRODUCE API，通过查看Schema的定义，能看到其V0版本的请求体和响应体的结构如下：

四. 请求的处理流程

  1. Acceptor监听到ACCEPT事件(TCP创建连接"第一次握手"的SYN)；

  2. Acceptor将将连接注册到Processor列表内的其中一个，由该Processor监听这个连接的后续可读可写事件；

  3. Processor接收到完整请求后，会将Request追加到RequestChannel中进行排队，等待后续处理；

  4. KafkaServer中有个requestHandlerPool的字段，KafkaRequestHandlerPool类型，代表请求处理线程池；KafkaRequestHandler就是其中的线程，会从RequestChannel拉请求进行处理；

  5. KafkaRequestHandler将拉到的Request传入KafkaApis.handle(Request)方法进行处理；

  6. KafkaApis根据不同的ApiKey调用不同的方法进行处理，处理完毕后会将Response最终写入对应的Processor的ResponseQueue中等待发送；KafkaApis.handle(Request)的方法结构如下：

def handle(request: RequestChannel.Request) {try {// 省略部分代码... ...request.header.apiKey match {case ApiKeys.PRODUCE => handleProduceRequest(request)case ApiKeys.FETCH => handleFetchRequest(request)case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)case ApiKeys.METADATA => handleTopicMetadataRequest(request)case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)// 省略部分代码... ...}} catch {case e: FatalExitError => throw ecase e: Throwable => handleError(request, e)} finally {request.apiLocalCompleteTimeNanos = time.nanoseconds}
}

  1. Processor从自己的ResponseQueue中拉取待发送的Respnose；

  2. Processor将Response发给客户端；

五. 总结

才疏学浅，未能窥其十之一二，随时欢迎各位交流补充。若文章质量还算及格，可以点赞收藏加以鼓励，后续我继续更新。

知乎主页：https://www.zhihu.com/people/hao_zhihu
关注收藏不迷路，第一时间接收技术文章推送

微信公众号：
微信二维码

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.ryyt.cn/news/28208.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈，一经查实，立即删除！

相关文章

一次通过dump文件分析OutOfMemoryError异常代码定位过程

一次通过dump文件分析OutOfMemoryError异常代码定位过程

OutOfMemoryError是Java程序中常见的异常，通常出现在内存不足时，导致程序无法运行。借助MAT内存分析工具分析可能的内存泄漏代码问题定位。OutOfMemoryError是Java程序中常见的异常，通常出现在内存不足时，导致程序无法运行。 当出现OutOfMemoryError异常时，可能的现象是这…
阅读更多...
前端埋点数据采集（二）mock应用系统10万条前端埋点数据

前端埋点数据采集（二）mock应用系统10万条前端埋点数据

前端埋点数据采集（二）mock应用系统10万条前端埋点数据 上一期我们分享了前端埋点数据采集（一）采集系统架构设计 我们说应用系统的数据，采集到大数据平台来，然后再到数仓。但是很多实际场景是应用系统、大数据平台、数仓平台各自并没有完成系统的搭建和开发。假设现在一个…
阅读更多...
windows下安装Jenkins以及配置分布式agent节点

windows下安装Jenkins以及配置分布式agent节点

安装Jenkins： 1.Jenkins稳定版本的war包路径：https://get.jenkins.io/war-stable/ 2.jdk下载：https://www.oracle.com/java/technologies/downloads/ 3.启动Jenkins：命令行运行java -jar jenkins.war 至此可以通过浏览器127.0.0.1:8080,连接上本地Jenkins配置分布式agent节…
阅读更多...
locust压测

locust压测

目录locust1.依赖2. 实例2.1 压测方式2.2 locust服务端2.3 待压测接口服务3. 参考文档 locust 1.依赖 pip install locust2. 实例 2.1 压测方式 1. 压测方式 1.1 前台自编辑方式修改文件名为locustfile.py 并在控制台使用locust启动前台服务 用户自定义压测参数并开启压测1.2 …
阅读更多...
如何快速找出文件夹里的全部带有符号纯符号的文件

如何快速找出文件夹里的全部带有符号纯符号的文件

参考此文章：如何快速找出文件夹里的全部带有中文&纯中文的文件 只需要根据自己的需求，把下面相关的设置调整好即可
阅读更多...
前端埋点数据采集（一）采集系统架构设计

前端埋点数据采集（一）采集系统架构设计

前几次内容分享，我们逐步搭建了传统hadoop大数据平台、zookeeper、kafka集群等。假设现在一个场景是：今天产品经理提了1个优先级极高的需求：应用系统前端埋点数据都需要被采集到大数据平台hadoop上。“前端埋点数据的采集系列”，主要分为以下5个部分： 一、采集系统架构设计…
阅读更多...
如何快速找出文件夹里的全部带有数字纯数字的文件

如何快速找出文件夹里的全部带有数字纯数字的文件

参考此文章：如何快速找出文件夹里的全部带有中文&纯中文的文件 只需要根据自己的需求，把下面相关的设置调整好即可
阅读更多...
更优性能与性价比，从自建 ELK 迁移到 SLS 开始

更优性能与性价比，从自建 ELK 迁移到 SLS 开始

本文介绍了 SLS 基本能力，并和开源自建 ELK 做了对比，可以看到 SLS 相比开源 ELK 有较大优势。背景 ELK (Elasticsearch、Logstash、Kibana) 是当下开源领域主流的日志解决方案，在可观测场景下有比较广泛的应用。 随着数字化进程加速，机器数据日志增加，自建 ELK 在面临大规…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站 尧图建网站 广利置业